Allow new values to be added to an existing enum type.

After much expenditure of effort, we've got this to the point where the
performance penalty is pretty minimal in typical cases.

Andrew Dunstan, reviewed by Brendan Jurd, Dean Rasheed, and Tom Lane

1 files changed, 176 insertions, 83 deletions

diff --git a/src/backend/utils/adt/enum.c b/src/backend/utils/adt/enum.c
index e5747a46bcd..dd168dd9730 100644
--- a/src/backend/utils/adt/enum.c
+++ b/src/backend/utils/adt/enum.c
@@ -13,18 +13,22 @@
  */
 #include "postgres.h"
 
+#include "access/genam.h"
+#include "access/heapam.h"
+#include "catalog/indexing.h"
 #include "catalog/pg_enum.h"
 #include "fmgr.h"
+#include "libpq/pqformat.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
-#include "utils/lsyscache.h"
+#include "utils/fmgroids.h"
+#include "utils/snapmgr.h"
 #include "utils/syscache.h"
-#include "libpq/pqformat.h"
-#include "miscadmin.h"
+#include "utils/typcache.h"
 
 
+static Oid	enum_endpoint(Oid enumtypoid, ScanDirection direction);
 static ArrayType *enum_range_internal(Oid enumtypoid, Oid lower, Oid upper);
-static int	enum_elem_cmp(const void *left, const void *right);
 
 
 /* Basic I/O support */
@@ -155,13 +159,63 @@ enum_send(PG_FUNCTION_ARGS)
 
 /* Comparison functions and related */
 
+/*
+ * enum_cmp_internal is the common engine for all the visible comparison
+ * functions, except for enum_eq and enum_ne which can just check for OID
+ * equality directly.
+ */
+static int
+enum_cmp_internal(Oid arg1, Oid arg2, FunctionCallInfo fcinfo)
+{
+	TypeCacheEntry *tcache;
+
+	/* Equal OIDs are equal no matter what */
+	if (arg1 == arg2)
+		return 0;
+
+	/* Fast path: even-numbered Oids are known to compare correctly */
+	if ((arg1 & 1) == 0 && (arg2 & 1) == 0)
+	{
+		if (arg1 < arg2)
+			return -1;
+		else
+			return 1;
+	}
+
+	/* Locate the typcache entry for the enum type */
+	tcache = (TypeCacheEntry *) fcinfo->flinfo->fn_extra;
+	if (tcache == NULL)
+	{
+		HeapTuple	enum_tup;
+		Form_pg_enum en;
+		Oid			typeoid;
+
+		/* Get the OID of the enum type containing arg1 */
+		enum_tup = SearchSysCache1(ENUMOID, ObjectIdGetDatum(arg1));
+		if (!HeapTupleIsValid(enum_tup))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
+					 errmsg("invalid internal value for enum: %u",
+							arg1)));
+		en = (Form_pg_enum) GETSTRUCT(enum_tup);
+		typeoid = en->enumtypid;
+		ReleaseSysCache(enum_tup);
+		/* Now locate and remember the typcache entry */
+		tcache = lookup_type_cache(typeoid, 0);
+		fcinfo->flinfo->fn_extra = (void *) tcache;
+	}
+
+	/* The remaining comparison logic is in typcache.c */
+	return compare_values_of_enum(tcache, arg1, arg2);
+}
+
 Datum
 enum_lt(PG_FUNCTION_ARGS)
 {
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_BOOL(a < b);
+	PG_RETURN_BOOL(enum_cmp_internal(a, b, fcinfo) < 0);
 }
 
 Datum
@@ -170,7 +224,7 @@ enum_le(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_BOOL(a <= b);
+	PG_RETURN_BOOL(enum_cmp_internal(a, b, fcinfo) <= 0);
 }
 
 Datum
@@ -197,7 +251,7 @@ enum_ge(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_BOOL(a >= b);
+	PG_RETURN_BOOL(enum_cmp_internal(a, b, fcinfo) >= 0);
 }
 
 Datum
@@ -206,7 +260,7 @@ enum_gt(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_BOOL(a > b);
+	PG_RETURN_BOOL(enum_cmp_internal(a, b, fcinfo) > 0);
 }
 
 Datum
@@ -215,7 +269,7 @@ enum_smaller(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_OID(a <= b ? a : b);
+	PG_RETURN_OID(enum_cmp_internal(a, b, fcinfo) < 0 ? a : b);
 }
 
 Datum
@@ -224,7 +278,7 @@ enum_larger(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	PG_RETURN_OID(a >= b ? a : b);
+	PG_RETURN_OID(enum_cmp_internal(a, b, fcinfo) > 0 ? a : b);
 }
 
 Datum
@@ -233,24 +287,63 @@ enum_cmp(PG_FUNCTION_ARGS)
 	Oid			a = PG_GETARG_OID(0);
 	Oid			b = PG_GETARG_OID(1);
 
-	if (a > b)
-		PG_RETURN_INT32(1);
-	else if (a == b)
+	if (a == b)
 		PG_RETURN_INT32(0);
+	else if (enum_cmp_internal(a, b, fcinfo) > 0)
+		PG_RETURN_INT32(1);
 	else
 		PG_RETURN_INT32(-1);
 }
 
 /* Enum programming support functions */
 
+/*
+ * enum_endpoint: common code for enum_first/enum_last
+ */
+static Oid
+enum_endpoint(Oid enumtypoid, ScanDirection direction)
+{
+	Relation	enum_rel;
+	Relation	enum_idx;
+	SysScanDesc enum_scan;
+	HeapTuple	enum_tuple;
+	ScanKeyData skey;
+	Oid			minmax;
+
+	/*
+	 * Find the first/last enum member using pg_enum_typid_sortorder_index.
+	 * Note we must not use the syscache, and must use an MVCC snapshot here.
+	 * See comments for RenumberEnumType in catalog/pg_enum.c for more info.
+	 */
+	ScanKeyInit(&skey,
+				Anum_pg_enum_enumtypid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(enumtypoid));
+
+	enum_rel = heap_open(EnumRelationId, AccessShareLock);
+	enum_idx = index_open(EnumTypIdSortOrderIndexId, AccessShareLock);
+	enum_scan = systable_beginscan_ordered(enum_rel, enum_idx,
+										   GetTransactionSnapshot(),
+										   1, &skey);
+
+	enum_tuple = systable_getnext_ordered(enum_scan, direction);
+	if (HeapTupleIsValid(enum_tuple))
+		minmax = HeapTupleGetOid(enum_tuple);
+	else
+		minmax = InvalidOid;
+
+	systable_endscan_ordered(enum_scan);
+	index_close(enum_idx, AccessShareLock);
+	heap_close(enum_rel, AccessShareLock);
+
+	return minmax;
+}
+
 Datum
 enum_first(PG_FUNCTION_ARGS)
 {
 	Oid			enumtypoid;
-	Oid			min = InvalidOid;
-	CatCList   *list;
-	int			num,
-				i;
+	Oid			min;
 
 	/*
 	 * We rely on being able to get the specific enum type from the calling
@@ -263,21 +356,14 @@ enum_first(PG_FUNCTION_ARGS)
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("could not determine actual enum type")));
 
-	list = SearchSysCacheList1(ENUMTYPOIDNAME, ObjectIdGetDatum(enumtypoid));
-	num = list->n_members;
-	for (i = 0; i < num; i++)
-	{
-		Oid			valoid = HeapTupleHeaderGetOid(list->members[i]->tuple.t_data);
-
-		if (!OidIsValid(min) || valoid < min)
-			min = valoid;
-	}
-
-	ReleaseCatCacheList(list);
+	/* Get the OID using the index */
+	min = enum_endpoint(enumtypoid, ForwardScanDirection);
 
-	if (!OidIsValid(min))		/* should not happen */
-		elog(ERROR, "no values found for enum %s",
-			 format_type_be(enumtypoid));
+	if (!OidIsValid(min))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("enum %s contains no values",
+						format_type_be(enumtypoid))));
 
 	PG_RETURN_OID(min);
 }
@@ -286,10 +372,7 @@ Datum
 enum_last(PG_FUNCTION_ARGS)
 {
 	Oid			enumtypoid;
-	Oid			max = InvalidOid;
-	CatCList   *list;
-	int			num,
-				i;
+	Oid			max;
 
 	/*
 	 * We rely on being able to get the specific enum type from the calling
@@ -302,21 +385,14 @@ enum_last(PG_FUNCTION_ARGS)
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("could not determine actual enum type")));
 
-	list = SearchSysCacheList1(ENUMTYPOIDNAME, ObjectIdGetDatum(enumtypoid));
-	num = list->n_members;
-	for (i = 0; i < num; i++)
-	{
-		Oid			valoid = HeapTupleHeaderGetOid(list->members[i]->tuple.t_data);
-
-		if (!OidIsValid(max) || valoid > max)
-			max = valoid;
-	}
-
-	ReleaseCatCacheList(list);
+	/* Get the OID using the index */
+	max = enum_endpoint(enumtypoid, BackwardScanDirection);
 
-	if (!OidIsValid(max))		/* should not happen */
-		elog(ERROR, "no values found for enum %s",
-			 format_type_be(enumtypoid));
+	if (!OidIsValid(max))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("enum %s contains no values",
+						format_type_be(enumtypoid))));
 
 	PG_RETURN_OID(max);
 }
@@ -377,51 +453,68 @@ static ArrayType *
 enum_range_internal(Oid enumtypoid, Oid lower, Oid upper)
 {
 	ArrayType  *result;
-	CatCList   *list;
-	int			total,
-				i,
-				j;
+	Relation	enum_rel;
+	Relation	enum_idx;
+	SysScanDesc enum_scan;
+	HeapTuple	enum_tuple;
+	ScanKeyData skey;
 	Datum	   *elems;
+	int			max,
+				cnt;
+	bool        left_found;
 
-	list = SearchSysCacheList1(ENUMTYPOIDNAME, ObjectIdGetDatum(enumtypoid));
-	total = list->n_members;
+	/*
+	 * Scan the enum members in order using pg_enum_typid_sortorder_index.
+	 * Note we must not use the syscache, and must use an MVCC snapshot here.
+	 * See comments for RenumberEnumType in catalog/pg_enum.c for more info.
+	 */
+	ScanKeyInit(&skey,
+				Anum_pg_enum_enumtypid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(enumtypoid));
+
+	enum_rel = heap_open(EnumRelationId, AccessShareLock);
+	enum_idx = index_open(EnumTypIdSortOrderIndexId, AccessShareLock);
+	enum_scan = systable_beginscan_ordered(enum_rel, enum_idx,
+										   GetTransactionSnapshot(),
+										   1, &skey);
+
+	max = 64;
+	elems = (Datum *) palloc(max * sizeof(Datum));
+	cnt = 0;
+	left_found = !OidIsValid(lower);
+
+	while (HeapTupleIsValid(enum_tuple = systable_getnext_ordered(enum_scan, ForwardScanDirection)))
+	{
+		Oid		enum_oid = HeapTupleGetOid(enum_tuple);
 
-	elems = (Datum *) palloc(total * sizeof(Datum));
+		if (!left_found && lower == enum_oid)
+			left_found = true;
 
-	j = 0;
-	for (i = 0; i < total; i++)
-	{
-		Oid			val = HeapTupleGetOid(&(list->members[i]->tuple));
+		if (left_found)
+		{
+			if (cnt >= max)
+			{
+				max *= 2;
+				elems = (Datum *) repalloc(elems, max * sizeof(Datum));
+			}
 
-		if ((!OidIsValid(lower) || lower <= val) &&
-			(!OidIsValid(upper) || val <= upper))
-			elems[j++] = ObjectIdGetDatum(val);
-	}
+			elems[cnt++] = ObjectIdGetDatum(enum_oid);
+		}
 
-	/* shouldn't need the cache anymore */
-	ReleaseCatCacheList(list);
+		if (OidIsValid(upper) && upper == enum_oid)
+			break;
+	}
 
-	/* sort results into OID order */
-	qsort(elems, j, sizeof(Datum), enum_elem_cmp);
+	systable_endscan_ordered(enum_scan);
+	index_close(enum_idx, AccessShareLock);
+	heap_close(enum_rel, AccessShareLock);
 
+	/* and build the result array */
 	/* note this hardwires some details about the representation of Oid */
-	result = construct_array(elems, j, enumtypoid, sizeof(Oid), true, 'i');
+	result = construct_array(elems, cnt, enumtypoid, sizeof(Oid), true, 'i');
 
 	pfree(elems);
 
 	return result;
 }
-
-/* qsort comparison function for Datums that are OIDs */
-static int
-enum_elem_cmp(const void *left, const void *right)
-{
-	Oid			l = DatumGetObjectId(*((const Datum *) left));
-	Oid			r = DatumGetObjectId(*((const Datum *) right));
-
-	if (l < r)
-		return -1;
-	if (l > r)
-		return 1;
-	return 0;
-}