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;
-}