Relax transactional restrictions on ALTER TYPE ... ADD VALUE (redux).

Originally committed as 15bc038 (plus some follow-ups), this was
reverted in 28e0727 due to a problem discovered in parallel
workers.  This new version corrects that problem by sending the
list of uncommitted enum values to parallel workers.

Here follows the original commit message describing the change:

To prevent possibly breaking indexes on enum columns, we must keep
uncommitted enum values from getting stored in tables, unless we
can be sure that any such column is new in the current transaction.

Formerly, we enforced this by disallowing ALTER TYPE ... ADD VALUE
from being executed at all in a transaction block, unless the target
enum type had been created in the current transaction.  This patch
removes that restriction, and instead insists that an uncommitted enum
value can't be referenced unless it belongs to an enum type created
in the same transaction as the value.  Per discussion, this should be
a bit less onerous.  It does require each function that could possibly
return a new enum value to SQL operations to check this restriction,
but there aren't so many of those that this seems unmaintainable.

Author: Andrew Dunstan and Tom Lane, with parallel query fix by Thomas Munro
Reviewed-by: Tom Lane
Discussion: https://postgr.es/m/CAEepm%3D0Ei7g6PaNTbcmAh9tCRahQrk%3Dr5ZWLD-jr7hXweYX3yg%40mail.gmail.com
Discussion: https://postgr.es/m/4075.1459088427%40sss.pgh.pa.us

Author: macdice

doc/src/sgml/ref/alter_type.sgml

@@ -290,8 +290,9 @@ ALTER TYPE <replaceable class="parameter">name</replaceable> RENAME VALUE <repla
   <title>Notes</title>
 
   <para>
-   <command>ALTER TYPE ... ADD VALUE</command> (the form that adds a new value to an
-   enum type) cannot be executed inside a transaction block.
+   If <command>ALTER TYPE ... ADD VALUE</command> (the form that adds a new
+   value to an enum type) is executed inside a transaction block, the new
+   value cannot be used until after the transaction has been committed.
   </para>
 
   <para>

src/backend/access/transam/parallel.c

@@ -19,6 +19,7 @@
 #include "access/session.h"
 #include "access/xact.h"
 #include "access/xlog.h"
+#include "catalog/pg_enum.h"
 #include "catalog/index.h"
 #include "catalog/namespace.h"
 #include "commands/async.h"
@@ -71,6 +72,7 @@
 #define PARALLEL_KEY_SESSION_DSM			UINT64CONST(0xFFFFFFFFFFFF000A)
 #define PARALLEL_KEY_REINDEX_STATE			UINT64CONST(0xFFFFFFFFFFFF000B)
 #define PARALLEL_KEY_RELMAPPER_STATE		UINT64CONST(0xFFFFFFFFFFFF000C)
+#define PARALLEL_KEY_ENUMBLACKLIST			UINT64CONST(0xFFFFFFFFFFFF000D)
 
 /* Fixed-size parallel state. */
 typedef struct FixedParallelState
@@ -210,6 +212,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
 	Size		tstatelen = 0;
 	Size		reindexlen = 0;
 	Size		relmapperlen = 0;
+	Size		enumblacklistlen = 0;
 	Size		segsize = 0;
 	int			i;
 	FixedParallelState *fps;
@@ -263,8 +266,10 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
 		relmapperlen = EstimateRelationMapSpace();
 		shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
+		enumblacklistlen = EstimateEnumBlacklistSpace();
+		shm_toc_estimate_chunk(&pcxt->estimator, enumblacklistlen);
 		/* If you add more chunks here, you probably need to add keys. */
-		shm_toc_estimate_keys(&pcxt->estimator, 9);
+		shm_toc_estimate_keys(&pcxt->estimator, 10);
 
 		/* Estimate space need for error queues. */
 		StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==
@@ -340,6 +345,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		char	   *error_queue_space;
 		char	   *session_dsm_handle_space;
 		char	   *entrypointstate;
+		char	   *enumblacklistspace;
 		Size		lnamelen;
 
 		/* Serialize shared libraries we have loaded. */
@@ -389,6 +395,12 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		shm_toc_insert(pcxt->toc, PARALLEL_KEY_RELMAPPER_STATE,
 					   relmapperspace);
 
+		/* Serialize enum blacklist state. */
+		enumblacklistspace = shm_toc_allocate(pcxt->toc, enumblacklistlen);
+		SerializeEnumBlacklist(enumblacklistspace, enumblacklistlen);
+		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENUMBLACKLIST,
+					   enumblacklistspace);
+
 		/* Allocate space for worker information. */
 		pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
 
@@ -1222,6 +1234,7 @@ ParallelWorkerMain(Datum main_arg)
 	char	   *tstatespace;
 	char	   *reindexspace;
 	char	   *relmapperspace;
+	char	   *enumblacklistspace;
 	StringInfoData msgbuf;
 	char	   *session_dsm_handle_space;
 
@@ -1408,6 +1421,11 @@ ParallelWorkerMain(Datum main_arg)
 	relmapperspace = shm_toc_lookup(toc, PARALLEL_KEY_RELMAPPER_STATE, false);
 	RestoreRelationMap(relmapperspace);
 
+	/* Restore enum blacklist. */
+	enumblacklistspace = shm_toc_lookup(toc, PARALLEL_KEY_ENUMBLACKLIST,
+										false);
+	RestoreEnumBlacklist(enumblacklistspace);
+
 	/*
 	 * We've initialized all of our state now; nothing should change
 	 * hereafter.

src/backend/access/transam/xact.c

@@ -31,6 +31,7 @@
 #include "access/xloginsert.h"
 #include "access/xlogutils.h"
 #include "catalog/namespace.h"
+#include "catalog/pg_enum.h"
 #include "catalog/storage.h"
 #include "commands/async.h"
 #include "commands/tablecmds.h"
@@ -2135,6 +2136,7 @@ CommitTransaction(void)
 	AtCommit_Notify();
 	AtEOXact_GUC(true, 1);
 	AtEOXact_SPI(true);
+	AtEOXact_Enum();
 	AtEOXact_on_commit_actions(true);
 	AtEOXact_Namespace(true, is_parallel_worker);
 	AtEOXact_SMgr();
@@ -2413,6 +2415,7 @@ PrepareTransaction(void)
 	/* PREPARE acts the same as COMMIT as far as GUC is concerned */
 	AtEOXact_GUC(true, 1);
 	AtEOXact_SPI(true);
+	AtEOXact_Enum();
 	AtEOXact_on_commit_actions(true);
 	AtEOXact_Namespace(true, false);
 	AtEOXact_SMgr();
@@ -2615,6 +2618,7 @@ AbortTransaction(void)
 
 		AtEOXact_GUC(false, 1);
 		AtEOXact_SPI(false);
+		AtEOXact_Enum();
 		AtEOXact_on_commit_actions(false);
 		AtEOXact_Namespace(false, is_parallel_worker);
 		AtEOXact_SMgr();

src/backend/catalog/pg_enum.c

@@ -28,13 +28,26 @@
 #include "utils/builtins.h"
 #include "utils/catcache.h"
 #include "utils/fmgroids.h"
+#include "utils/hsearch.h"
+#include "utils/memutils.h"
 #include "utils/syscache.h"
 #include "utils/tqual.h"
 
 
 /* Potentially set by pg_upgrade_support functions */
 Oid			binary_upgrade_next_pg_enum_oid = InvalidOid;
 
+/*
+ * Hash table of enum value OIDs created during the current transaction by
+ * AddEnumLabel.  We disallow using these values until the transaction is
+ * committed; otherwise, they might get into indexes where we can't clean
+ * them up, and then if the transaction rolls back we have a broken index.
+ * (See comments for check_safe_enum_use() in enum.c.)  Values created by
+ * EnumValuesCreate are *not* blacklisted; we assume those are created during
+ * CREATE TYPE, so they can't go away unless the enum type itself does.
+ */
+static HTAB *enum_blacklist = NULL;
+
 static void RenumberEnumType(Relation pg_enum, HeapTuple *existing, int nelems);
 static int	sort_order_cmp(const void *p1, const void *p2);
 
@@ -168,6 +181,23 @@ EnumValuesDelete(Oid enumTypeOid)
 	heap_close(pg_enum, RowExclusiveLock);
 }
 
+/*
+ * Initialize the enum blacklist for this transaction.
+ */
+static void
+init_enum_blacklist(void)
+{
+	HASHCTL		hash_ctl;
+
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+	hash_ctl.keysize = sizeof(Oid);
+	hash_ctl.entrysize = sizeof(Oid);
+	hash_ctl.hcxt = TopTransactionContext;
+	enum_blacklist = hash_create("Enum value blacklist",
+								 32,
+								 &hash_ctl,
+								 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+}
 
 /*
  * AddEnumLabel
@@ -460,6 +490,13 @@ AddEnumLabel(Oid enumTypeOid,
 	heap_freetuple(enum_tup);
 
 	heap_close(pg_enum, RowExclusiveLock);
+
+	/* Set up the blacklist hash if not already done in this transaction */
+	if (enum_blacklist == NULL)
+		init_enum_blacklist();
+
+	/* Add the new value to the blacklist */
+	(void) hash_search(enum_blacklist, &newOid, HASH_ENTER, NULL);
 }
 
 
@@ -547,6 +584,39 @@ RenameEnumLabel(Oid enumTypeOid,
 }
 
 
+/*
+ * Test if the given enum value is on the blacklist
+ */
+bool
+EnumBlacklisted(Oid enum_id)
+{
+	bool		found;
+
+	/* If we've made no blacklist table, all values are safe */
+	if (enum_blacklist == NULL)
+		return false;
+
+	/* Else, is it in the table? */
+	(void) hash_search(enum_blacklist, &enum_id, HASH_FIND, &found);
+	return found;
+}
+
+
+/*
+ * Clean up enum stuff after end of top-level transaction.
+ */
+void
+AtEOXact_Enum(void)
+{
+	/*
+	 * Reset the blacklist table, as all our enum values are now committed.
+	 * The memory will go away automatically when TopTransactionContext is
+	 * freed; it's sufficient to clear our pointer.
+	 */
+	enum_blacklist = NULL;
+}
+
+
 /*
  * RenumberEnumType
  *		Renumber existing enum elements to have sort positions 1..n.
@@ -620,3 +690,72 @@ sort_order_cmp(const void *p1, const void *p2)
 	else
 		return 0;
 }
+
+Size
+EstimateEnumBlacklistSpace(void)
+{
+	size_t		entries;
+
+	if (enum_blacklist)
+		entries = hash_get_num_entries(enum_blacklist);
+	else
+		entries = 0;
+
+	/* Add one for the terminator. */
+	return sizeof(Oid) * (entries + 1);
+}
+
+void
+SerializeEnumBlacklist(void *space, Size size)
+{
+	Oid		   *serialized = (Oid *) space;
+
+	/*
+	 * Make sure the hash table hasn't changed in size since the caller
+	 * reserved the space.
+	 */
+	Assert(size == EstimateEnumBlacklistSpace());
+
+	/* Write out all the values from the hash table, if there is one. */
+	if (enum_blacklist)
+	{
+		HASH_SEQ_STATUS status;
+		Oid		   *value;
+
+		hash_seq_init(&status, enum_blacklist);
+		while ((value = (Oid *) hash_seq_search(&status)))
+			*serialized++ = *value;
+	}
+
+	/* Write out the terminator. */
+	*serialized = InvalidOid;
+
+	/*
+	 * Make sure the amount of space we actually used matches what was
+	 * estimated.
+	 */
+	Assert((char *) (serialized + 1) == ((char *) space) + size);
+}
+
+void
+RestoreEnumBlacklist(void *space)
+{
+	Oid		   *serialized = (Oid *) space;
+
+	Assert(!enum_blacklist);
+
+	/*
+	 * As a special case, if the list is empty then don't even bother to
+	 * create the hash table.  This is the usual case, since enum alteration
+	 * is expected to be rare.
+	 */
+	if (!OidIsValid(*serialized))
+		return;
+
+	/* Read all the values into a new hash table. */
+	init_enum_blacklist();
+	do
+	{
+		hash_search(enum_blacklist, serialized++, HASH_ENTER, NULL);
+	} while (OidIsValid(*serialized));
+}

src/backend/commands/typecmds.c

@@ -1270,10 +1270,10 @@ DefineEnum(CreateEnumStmt *stmt)
 
 /*
  * AlterEnum
- *		ALTER TYPE on an enum.
+ *		Adds a new label to an existing enum.
  */
 ObjectAddress
-AlterEnum(AlterEnumStmt *stmt, bool isTopLevel)
+AlterEnum(AlterEnumStmt *stmt)
 {
 	Oid			enum_type_oid;
 	TypeName   *typename;
@@ -1291,6 +1291,8 @@ AlterEnum(AlterEnumStmt *stmt, bool isTopLevel)
 	/* Check it's an enum and check user has permission to ALTER the enum */
 	checkEnumOwner(tup);
 
+	ReleaseSysCache(tup);
+
 	if (stmt->oldVal)
 	{
 		/* Rename an existing label */
@@ -1299,27 +1301,6 @@ AlterEnum(AlterEnumStmt *stmt, bool isTopLevel)
 	else
 	{
 		/* Add a new label */
-
-		/*
-		 * Ordinarily we disallow adding values within transaction blocks,
-		 * because we can't cope with enum OID values getting into indexes and
-		 * then having their defining pg_enum entries go away.  However, it's
-		 * okay if the enum type was created in the current transaction, since
-		 * then there can be no such indexes that wouldn't themselves go away
-		 * on rollback.  (We support this case because pg_dump
-		 * --binary-upgrade needs it.)  We test this by seeing if the pg_type
-		 * row has xmin == current XID and is not HEAP_UPDATED.  If it is
-		 * HEAP_UPDATED, we can't be sure whether the type was created or only
-		 * modified in this xact.  So we are disallowing some cases that could
-		 * theoretically be safe; but fortunately pg_dump only needs the
-		 * simplest case.
-		 */
-		if (HeapTupleHeaderGetXmin(tup->t_data) == GetCurrentTransactionId() &&
-			!(tup->t_data->t_infomask & HEAP_UPDATED))
-			 /* safe to do inside transaction block */ ;
-		else
-			PreventInTransactionBlock(isTopLevel, "ALTER TYPE ... ADD");
-
 		AddEnumLabel(enum_type_oid, stmt->newVal,
 					 stmt->newValNeighbor, stmt->newValIsAfter,
 					 stmt->skipIfNewValExists);
@@ -1329,8 +1310,6 @@ AlterEnum(AlterEnumStmt *stmt, bool isTopLevel)
 
 	ObjectAddressSet(address, TypeRelationId, enum_type_oid);
 
-	ReleaseSysCache(tup);
-
 	return address;
 }
 

src/backend/tcop/utility.c

@@ -1439,7 +1439,7 @@ ProcessUtilitySlow(ParseState *pstate,
 				break;
 
 			case T_AlterEnumStmt:	/* ALTER TYPE (enum) */
-				address = AlterEnum((AlterEnumStmt *) parsetree, isTopLevel);
+				address = AlterEnum((AlterEnumStmt *) parsetree);
 				break;
 
 			case T_ViewStmt:	/* CREATE VIEW */