Restructure function-internal caching in the range type code.

Move the responsibility for caching specialized information about range types into the type cache, so that the catalog lookups only have to occur once per session. Rearrange APIs a bit so that fn_extra caching is actually effective in the GiST support code. (Use of OidFunctionCallN is bad enough for performance in itself, but it also prevents the function from exploiting fn_extra caching.) The range I/O functions are still not very bright about caching repeated lookups, but that seems like material for a separate patch. Also, avoid unnecessary use of memcpy to fetch/store the range type OID and flags, and don't use the full range_deserialize machinery when all we need to see is the flags value. Also fix API error in range_gist_penalty --- it was failing to set *penalty for any case involving an empty range.
2025-09-03 15:22:11 +03:00 · 2011-11-15 13:05:45 -05:00
parent ad50934eaa
commit 37ee4b75db
5 changed files with 705 additions and 607 deletions
--- a/src/backend/utils/adt/rangetypes.c
+++ b/src/backend/utils/adt/rangetypes.c
--- a/src/backend/utils/adt/rangetypes_gist.c
+++ b/src/backend/utils/adt/rangetypes_gist.c
@@ -17,8 +17,6 @@
 #include "access/gist.h"
 #include "access/skey.h"
 #include "utils/builtins.h"
-#include "utils/fmgroids.h"
-#include "utils/lsyscache.h"
 #include "utils/rangetypes.h"


@@ -36,22 +34,24 @@
 #define RANGESTRAT_OVERRIGHT			11
 #define RANGESTRAT_ADJACENT				12

+#define RangeIsEmpty(r)  (range_get_flags(r) & RANGE_EMPTY)
+
 /*
 * Auxiliary structure for picksplit method.
 */
 typedef struct
 {
-	int			index;
-	RangeType  *data;
-	FunctionCallInfo fcinfo;
+	int			index;			/* original index in entryvec->vector[] */
+	RangeType  *data;			/* range value to sort */
+	TypeCacheEntry *typcache;	/* range type's info */
 }	PickSplitSortItem;

-static RangeType *range_super_union(FunctionCallInfo fcinfo, RangeType * r1,
+static RangeType *range_super_union(TypeCacheEntry *typcache, RangeType * r1,
 				  RangeType * r2);
-static bool range_gist_consistent_int(FunctionCallInfo fcinfo,
+static bool range_gist_consistent_int(FmgrInfo *flinfo,
 						  StrategyNumber strategy, RangeType * key,
 						  RangeType * query);
-static bool range_gist_consistent_leaf(FunctionCallInfo fcinfo,
+static bool range_gist_consistent_leaf(FmgrInfo *flinfo,
 						   StrategyNumber strategy, RangeType * key,
 						   RangeType * query);
 static int	sort_item_cmp(const void *a, const void *b);
@@ -66,15 +66,13 @@ range_gist_consistent(PG_FUNCTION_ARGS)
 	/* Oid subtype = PG_GETARG_OID(3); */
 	bool	   *recheck = (bool *) PG_GETARG_POINTER(4);
 	RangeType  *key = DatumGetRangeType(entry->key);
+	TypeCacheEntry *typcache;
 	RangeType  *query;
 	RangeBound	lower;
 	RangeBound	upper;
-	bool		empty;
-	Oid			rngtypid;

+	/* All operators served by this function are exact */
 	*recheck = false;
-	range_deserialize(fcinfo, key, &lower, &upper, &empty);
-	rngtypid = lower.rngtypid;

 	switch (strategy)
 	{
@@ -85,18 +83,19 @@ range_gist_consistent(PG_FUNCTION_ARGS)
 		 */
 		case RANGESTRAT_CONTAINS_ELEM:
 		case RANGESTRAT_ELEM_CONTAINED_BY:
-			lower.rngtypid = rngtypid;
-			lower.inclusive = true;
+			typcache = range_get_typcache(fcinfo, RangeTypeGetOid(key));
+
 			lower.val = dquery;
-			lower.lower = true;
 			lower.infinite = false;
-			upper.rngtypid = rngtypid;
-			upper.inclusive = true;
+			lower.inclusive = true;
+			lower.lower = true;
+
 			upper.val = dquery;
-			upper.lower = false;
 			upper.infinite = false;
-			query = DatumGetRangeType(make_range(fcinfo,
-												 &lower, &upper, false));
+			upper.inclusive = true;
+			upper.lower = false;
+
+			query = make_range(typcache, &lower, &upper, false);
 			break;

 		default:
@@ -105,10 +104,10 @@ range_gist_consistent(PG_FUNCTION_ARGS)
 	}

 	if (GIST_LEAF(entry))
-		PG_RETURN_BOOL(range_gist_consistent_leaf(fcinfo, strategy,
+		PG_RETURN_BOOL(range_gist_consistent_leaf(fcinfo->flinfo, strategy,
 												  key, query));
 	else
-		PG_RETURN_BOOL(range_gist_consistent_int(fcinfo, strategy,
+		PG_RETURN_BOOL(range_gist_consistent_int(fcinfo->flinfo, strategy,
 												 key, query));
 }

@@ -118,13 +117,16 @@ range_gist_union(PG_FUNCTION_ARGS)
 	GistEntryVector *entryvec = (GistEntryVector *) PG_GETARG_POINTER(0);
 	GISTENTRY  *ent = entryvec->vector;
 	RangeType  *result_range;
+	TypeCacheEntry *typcache;
 	int			i;

 	result_range = DatumGetRangeType(ent[0].key);

+	typcache = range_get_typcache(fcinfo, RangeTypeGetOid(result_range));
+
 	for (i = 1; i < entryvec->n; i++)
 	{
-		result_range = range_super_union(fcinfo, result_range,
+		result_range = range_super_union(typcache, result_range,
 										 DatumGetRangeType(ent[i].key));
 	}

@@ -155,6 +157,7 @@ range_gist_penalty(PG_FUNCTION_ARGS)
 	float	   *penalty = (float *) PG_GETARG_POINTER(2);
 	RangeType  *orig = DatumGetRangeType(origentry->key);
 	RangeType  *new = DatumGetRangeType(newentry->key);
+	TypeCacheEntry *typcache;
 	RangeType  *s_union;
 	FmgrInfo   *subtype_diff;
 	RangeBound	lower1,
@@ -163,34 +166,54 @@ range_gist_penalty(PG_FUNCTION_ARGS)
 				upper2;
 	bool		empty1,
 				empty2;
-	float		lower_diff,
+	float8		lower_diff,
 				upper_diff;
-	RangeTypeInfo rngtypinfo;

-	s_union = range_super_union(fcinfo, orig, new);
+	if (RangeTypeGetOid(orig) != RangeTypeGetOid(new))
+		elog(ERROR, "range types do not match");

-	range_deserialize(fcinfo, orig, &lower1, &upper1, &empty1);
-	range_deserialize(fcinfo, s_union, &lower2, &upper2, &empty2);
+	typcache = range_get_typcache(fcinfo, RangeTypeGetOid(orig));

-	range_gettypinfo(fcinfo, lower1.rngtypid, &rngtypinfo);
-	subtype_diff = &rngtypinfo.subdiffFn;
+	subtype_diff = &typcache->rng_subdiff_finfo;

+	s_union = range_super_union(typcache, orig, new);
+
+	range_deserialize(typcache, orig, &lower1, &upper1, &empty1);
+	range_deserialize(typcache, s_union, &lower2, &upper2, &empty2);
+
+	/* if orig isn't empty, s_union can't be either */
 	Assert(empty1 || !empty2);

 	if (empty1 && empty2)
-		return 0;
+	{
+		*penalty = 0;
+		PG_RETURN_POINTER(penalty);
+	}
 	else if (empty1 && !empty2)
 	{
 		if (lower2.infinite || upper2.infinite)
+		{
 			/* from empty to infinite */
-			return get_float8_infinity();
-		else if (subtype_diff->fn_addr != NULL)
+			*penalty = get_float4_infinity();
+			PG_RETURN_POINTER(penalty);
+		}
+		else if (OidIsValid(subtype_diff->fn_oid))
+		{
 			/* from empty to upper2-lower2 */
-			return DatumGetFloat8(FunctionCall2(subtype_diff,
-												upper2.val, lower2.val));
+			*penalty = DatumGetFloat8(FunctionCall2Coll(subtype_diff,
+														typcache->rng_collation,
+														upper2.val,
+														lower2.val));
+			if (*penalty < 0)
+				*penalty = 0;		/* subtype_diff is broken */
+			PG_RETURN_POINTER(penalty);
+		}
 		else
+		{
 			/* wild guess */
-			return 1.0;
+			*penalty = 1.0;
+			PG_RETURN_POINTER(penalty);
+		}
 	}

 	Assert(lower2.infinite || !lower1.infinite);
@@ -199,15 +222,20 @@ range_gist_penalty(PG_FUNCTION_ARGS)
 		lower_diff = get_float8_infinity();
 	else if (lower2.infinite && lower1.infinite)
 		lower_diff = 0;
-	else if (subtype_diff->fn_addr != NULL)
+	else if (OidIsValid(subtype_diff->fn_oid))
 	{
-		lower_diff = DatumGetFloat8(FunctionCall2(subtype_diff,
-												  lower1.val, lower2.val));
+		lower_diff = DatumGetFloat8(FunctionCall2Coll(subtype_diff,
+													  typcache->rng_collation,
+													  lower1.val,
+													  lower2.val));
 		if (lower_diff < 0)
 			lower_diff = 0;		/* subtype_diff is broken */
 	}
-	else	/* only know whether there is a difference or not */
-		lower_diff = (float) range_cmp_bounds(fcinfo, &lower1, &lower2);
+	else
+	{
+		/* only know whether there is a difference or not */
+		lower_diff = (float) range_cmp_bounds(typcache, &lower1, &lower2);
+	}

 	Assert(upper2.infinite || !upper1.infinite);

@@ -215,15 +243,20 @@ range_gist_penalty(PG_FUNCTION_ARGS)
 		upper_diff = get_float8_infinity();
 	else if (upper2.infinite && upper1.infinite)
 		upper_diff = 0;
-	else if (subtype_diff->fn_addr != NULL)
+	else if (OidIsValid(subtype_diff->fn_oid))
 	{
-		upper_diff = DatumGetFloat8(FunctionCall2(subtype_diff,
-												  upper2.val, upper1.val));
+		upper_diff = DatumGetFloat8(FunctionCall2Coll(subtype_diff,
+													  typcache->rng_collation,
+													  upper2.val,
+													  upper1.val));
 		if (upper_diff < 0)
 			upper_diff = 0;		/* subtype_diff is broken */
 	}
-	else	/* only know whether there is a difference or not */
-		upper_diff = (float) range_cmp_bounds(fcinfo, &upper2, &upper1);
+	else
+	{
+		/* only know whether there is a difference or not */
+		upper_diff = (float) range_cmp_bounds(typcache, &upper2, &upper1);
+	}

 	Assert(lower_diff >= 0 && upper_diff >= 0);

@@ -243,6 +276,7 @@ range_gist_picksplit(PG_FUNCTION_ARGS)
 {
 	GistEntryVector *entryvec = (GistEntryVector *) PG_GETARG_POINTER(0);
 	GIST_SPLITVEC *v = (GIST_SPLITVEC *) PG_GETARG_POINTER(1);
+	TypeCacheEntry *typcache;
 	OffsetNumber i;
 	RangeType  *pred_left;
 	RangeType  *pred_right;
@@ -253,23 +287,28 @@ range_gist_picksplit(PG_FUNCTION_ARGS)
 	OffsetNumber *right;
 	OffsetNumber maxoff;

+	/* use first item to look up range type's info */
+	pred_left = DatumGetRangeType(entryvec->vector[FirstOffsetNumber].key);
+	typcache = range_get_typcache(fcinfo, RangeTypeGetOid(pred_left));
+
+	/* allocate result and work arrays */
 	maxoff = entryvec->n - 1;
 	nbytes = (maxoff + 1) * sizeof(OffsetNumber);
-	sortItems = (PickSplitSortItem *) palloc(
-										 maxoff * sizeof(PickSplitSortItem));
 	v->spl_left = (OffsetNumber *) palloc(nbytes);
 	v->spl_right = (OffsetNumber *) palloc(nbytes);
+	sortItems = (PickSplitSortItem *) palloc(maxoff * sizeof(PickSplitSortItem));

 	/*
-	 * Preparing auxiliary array and sorting.
+	 * Prepare auxiliary array and sort the values.
 	 */
 	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
 	{
 		sortItems[i - 1].index = i;
 		sortItems[i - 1].data = DatumGetRangeType(entryvec->vector[i].key);
-		sortItems[i - 1].fcinfo = fcinfo;
+		sortItems[i - 1].typcache = typcache;
 	}
 	qsort(sortItems, maxoff, sizeof(PickSplitSortItem), sort_item_cmp);
+
 	split_idx = maxoff / 2;

 	left = v->spl_left;
@@ -278,29 +317,27 @@ range_gist_picksplit(PG_FUNCTION_ARGS)
 	v->spl_nright = 0;

 	/*
-	 * First half of segs goes to the left datum.
+	 * First half of items goes to the left datum.
 	 */
-	pred_left = DatumGetRangeType(sortItems[0].data);
+	pred_left = sortItems[0].data;
 	*left++ = sortItems[0].index;
 	v->spl_nleft++;
 	for (i = 1; i < split_idx; i++)
 	{
-		pred_left = range_super_union(fcinfo, pred_left,
-									  DatumGetRangeType(sortItems[i].data));
+		pred_left = range_super_union(typcache, pred_left, sortItems[i].data);
 		*left++ = sortItems[i].index;
 		v->spl_nleft++;
 	}

 	/*
-	 * Second half of segs goes to the right datum.
+	 * Second half of items goes to the right datum.
 	 */
-	pred_right = DatumGetRangeType(sortItems[split_idx].data);
+	pred_right = sortItems[split_idx].data;
 	*right++ = sortItems[split_idx].index;
 	v->spl_nright++;
 	for (i = split_idx + 1; i < maxoff; i++)
 	{
-		pred_right = range_super_union(fcinfo, pred_right,
-									   DatumGetRangeType(sortItems[i].data));
+		pred_right = range_super_union(typcache, pred_right, sortItems[i].data);
 		*right++ = sortItems[i].index;
 		v->spl_nright++;
 	}
@@ -316,11 +353,16 @@ range_gist_picksplit(PG_FUNCTION_ARGS)
 Datum
 range_gist_same(PG_FUNCTION_ARGS)
 {
-	Datum		r1 = PG_GETARG_DATUM(0);
-	Datum		r2 = PG_GETARG_DATUM(1);
+	/* Datum r1 = PG_GETARG_DATUM(0); */
+	/* Datum r2 = PG_GETARG_DATUM(1); */
 	bool	   *result = (bool *) PG_GETARG_POINTER(2);

-	*result = DatumGetBool(OidFunctionCall2(F_RANGE_EQ, r1, r2));
+	/*
+	 * We can safely call range_eq using our fcinfo directly; it won't notice
+	 * the third argument.  This allows it to use fn_extra for caching.
+	 */
+	*result = DatumGetBool(range_eq(fcinfo));
+
 	PG_RETURN_POINTER(result);
 }

@@ -330,9 +372,11 @@ range_gist_same(PG_FUNCTION_ARGS)
 *----------------------------------------------------------
 */

-/* return the smallest range that contains r1 and r2 */
+/*
+ * Return the smallest range that contains r1 and r2
+ */
 static RangeType *
-range_super_union(FunctionCallInfo fcinfo, RangeType * r1, RangeType * r2)
+range_super_union(TypeCacheEntry *typcache, RangeType * r1, RangeType * r2)
 {
 	RangeBound	lower1,
 				lower2;
@@ -343,20 +387,20 @@ range_super_union(FunctionCallInfo fcinfo, RangeType * r1, RangeType * r2)
 	RangeBound *result_lower;
 	RangeBound *result_upper;

-	range_deserialize(fcinfo, r1, &lower1, &upper1, &empty1);
-	range_deserialize(fcinfo, r2, &lower2, &upper2, &empty2);
+	range_deserialize(typcache, r1, &lower1, &upper1, &empty1);
+	range_deserialize(typcache, r2, &lower2, &upper2, &empty2);

 	if (empty1)
 		return r2;
 	if (empty2)
 		return r1;

-	if (range_cmp_bounds(fcinfo, &lower1, &lower2) <= 0)
+	if (range_cmp_bounds(typcache, &lower1, &lower2) <= 0)
 		result_lower = &lower1;
 	else
 		result_lower = &lower2;

-	if (range_cmp_bounds(fcinfo, &upper1, &upper2) >= 0)
+	if (range_cmp_bounds(typcache, &upper1, &upper2) >= 0)
 		result_upper = &upper1;
 	else
 		result_upper = &upper2;
@@ -367,161 +411,182 @@ range_super_union(FunctionCallInfo fcinfo, RangeType * r1, RangeType * r2)
 	if (result_lower == &lower2 && result_upper == &upper2)
 		return r2;

-	return DatumGetRangeType(make_range(fcinfo, result_lower, result_upper,
-										false));
+	return make_range(typcache, result_lower, result_upper, false);
 }

+/*
+ * trick function call: call the given function with given FmgrInfo
+ *
+ * To allow the various functions called here to cache lookups of range
+ * datatype information, we use a trick: we pass them the FmgrInfo struct
+ * for the GiST consistent function.  This relies on the knowledge that
+ * none of them consult FmgrInfo for anything but fn_extra, and that they
+ * all use fn_extra the same way, i.e. as a pointer to the typcache entry
+ * for the range data type.  Since the FmgrInfo is long-lived (it's actually
+ * part of the relcache entry for the index, typically) this essentially
+ * eliminates lookup overhead during operations on a GiST range index.
+ */
+static Datum
+TrickFunctionCall2(PGFunction proc, FmgrInfo *flinfo, Datum arg1, Datum arg2)
+{
+	FunctionCallInfoData fcinfo;
+	Datum		result;
+
+	InitFunctionCallInfoData(fcinfo, flinfo, 2, InvalidOid, NULL, NULL);
+
+	fcinfo.arg[0] = arg1;
+	fcinfo.arg[1] = arg2;
+	fcinfo.argnull[0] = false;
+	fcinfo.argnull[1] = false;
+
+	result = (*proc) (&fcinfo);
+
+	if (fcinfo.isnull)
+		elog(ERROR, "function %p returned NULL", proc);
+
+	return result;
+}
+
+/*
+ * GiST consistent test on an index internal page
+ */
 static bool
-range_gist_consistent_int(FunctionCallInfo fcinfo, StrategyNumber strategy,
+range_gist_consistent_int(FmgrInfo *flinfo, StrategyNumber strategy,
 						  RangeType * key, RangeType * query)
 {
-	Oid			proc;
-	RangeBound	lower1,
-				lower2;
-	RangeBound	upper1,
-				upper2;
-	bool		empty1,
-				empty2;
-	bool		retval;
+	PGFunction	proc;
 	bool		negate = false;
-
-	range_deserialize(fcinfo, key, &lower1, &upper1, &empty1);
-	range_deserialize(fcinfo, query, &lower2, &upper2, &empty2);
+	bool		retval;

 	switch (strategy)
 	{
 		case RANGESTRAT_EQ:
-			proc = F_RANGE_CONTAINS;
+			proc = range_contains;
 			break;
 		case RANGESTRAT_NE:
 			return true;
 			break;
 		case RANGESTRAT_OVERLAPS:
-			proc = F_RANGE_OVERLAPS;
+			proc = range_overlaps;
 			break;
 		case RANGESTRAT_CONTAINS_ELEM:
 		case RANGESTRAT_CONTAINS:
-			proc = F_RANGE_CONTAINS;
+			proc = range_contains;
 			break;
 		case RANGESTRAT_ELEM_CONTAINED_BY:
 		case RANGESTRAT_CONTAINED_BY:
 			return true;
 			break;
 		case RANGESTRAT_BEFORE:
-			if (empty1)
+			if (RangeIsEmpty(key))
 				return false;
-			proc = F_RANGE_OVERRIGHT;
+			proc = range_overright;
 			negate = true;
 			break;
 		case RANGESTRAT_AFTER:
-			if (empty1)
+			if (RangeIsEmpty(key))
 				return false;
-			proc = F_RANGE_OVERLEFT;
+			proc = range_overleft;
 			negate = true;
 			break;
 		case RANGESTRAT_OVERLEFT:
-			if (empty1)
+			if (RangeIsEmpty(key))
 				return false;
-			proc = F_RANGE_AFTER;
+			proc = range_after;
 			negate = true;
 			break;
 		case RANGESTRAT_OVERRIGHT:
-			if (empty1)
+			if (RangeIsEmpty(key))
 				return false;
-			proc = F_RANGE_BEFORE;
+			proc = range_before;
 			negate = true;
 			break;
 		case RANGESTRAT_ADJACENT:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			if (DatumGetBool(OidFunctionCall2(F_RANGE_ADJACENT,
-											  RangeTypeGetDatum(key),
-											  RangeTypeGetDatum(query))))
+			if (DatumGetBool(TrickFunctionCall2(range_adjacent, flinfo,
+												RangeTypeGetDatum(key),
+												RangeTypeGetDatum(query))))
 				return true;
-			proc = F_RANGE_OVERLAPS;
+			proc = range_overlaps;
 			break;
 		default:
 			elog(ERROR, "unrecognized range strategy: %d", strategy);
-			proc = InvalidOid;
+			proc = NULL;		/* keep compiler quiet */
 			break;
 	}

-	retval = DatumGetBool(OidFunctionCall2(proc, RangeTypeGetDatum(key),
-										   RangeTypeGetDatum(query)));
-
+	retval = DatumGetBool(TrickFunctionCall2(proc, flinfo,
+											 RangeTypeGetDatum(key),
+											 RangeTypeGetDatum(query)));
 	if (negate)
 		retval = !retval;

-	PG_RETURN_BOOL(retval);
+	return retval;
 }

+/*
+ * GiST consistent test on an index leaf page
+ */
 static bool
-range_gist_consistent_leaf(FunctionCallInfo fcinfo, StrategyNumber strategy,
+range_gist_consistent_leaf(FmgrInfo *flinfo, StrategyNumber strategy,
 						   RangeType * key, RangeType * query)
 {
-	Oid			proc;
-	RangeBound	lower1,
-				lower2;
-	RangeBound	upper1,
-				upper2;
-	bool		empty1,
-				empty2;
-
-	range_deserialize(fcinfo, key, &lower1, &upper1, &empty1);
-	range_deserialize(fcinfo, query, &lower2, &upper2, &empty2);
+	PGFunction	proc;

 	switch (strategy)
 	{
 		case RANGESTRAT_EQ:
-			proc = F_RANGE_EQ;
+			proc = range_eq;
 			break;
 		case RANGESTRAT_NE:
-			proc = F_RANGE_NE;
+			proc = range_ne;
 			break;
 		case RANGESTRAT_OVERLAPS:
-			proc = F_RANGE_OVERLAPS;
+			proc = range_overlaps;
 			break;
 		case RANGESTRAT_CONTAINS_ELEM:
 		case RANGESTRAT_CONTAINS:
-			proc = F_RANGE_CONTAINS;
+			proc = range_contains;
 			break;
 		case RANGESTRAT_ELEM_CONTAINED_BY:
 		case RANGESTRAT_CONTAINED_BY:
-			proc = F_RANGE_CONTAINED_BY;
+			proc = range_contained_by;
 			break;
 		case RANGESTRAT_BEFORE:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			proc = F_RANGE_BEFORE;
+			proc = range_before;
 			break;
 		case RANGESTRAT_AFTER:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			proc = F_RANGE_AFTER;
+			proc = range_after;
 			break;
 		case RANGESTRAT_OVERLEFT:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			proc = F_RANGE_OVERLEFT;
+			proc = range_overleft;
 			break;
 		case RANGESTRAT_OVERRIGHT:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			proc = F_RANGE_OVERRIGHT;
+			proc = range_overright;
 			break;
 		case RANGESTRAT_ADJACENT:
-			if (empty1 || empty2)
+			if (RangeIsEmpty(key) || RangeIsEmpty(query))
 				return false;
-			proc = F_RANGE_ADJACENT;
+			proc = range_adjacent;
 			break;
 		default:
 			elog(ERROR, "unrecognized range strategy: %d", strategy);
-			proc = InvalidOid;
+			proc = NULL;		/* keep compiler quiet */
 			break;
 	}

-	return DatumGetBool(OidFunctionCall2(proc, RangeTypeGetDatum(key),
-										 RangeTypeGetDatum(query)));
+	return DatumGetBool(TrickFunctionCall2(proc, flinfo,
+										   RangeTypeGetDatum(key),
+										   RangeTypeGetDatum(query)));
 }

 /*
@@ -545,17 +610,17 @@ sort_item_cmp(const void *a, const void *b)
 	PickSplitSortItem *i2 = (PickSplitSortItem *) b;
 	RangeType  *r1 = i1->data;
 	RangeType  *r2 = i2->data;
+	TypeCacheEntry *typcache = i1->typcache;
 	RangeBound	lower1,
 				lower2;
 	RangeBound	upper1,
 				upper2;
 	bool		empty1,
 				empty2;
-	FunctionCallInfo fcinfo = i1->fcinfo;
 	int			cmp;

-	range_deserialize(fcinfo, r1, &lower1, &upper1, &empty1);
-	range_deserialize(fcinfo, r2, &lower2, &upper2, &empty2);
+	range_deserialize(typcache, r1, &lower1, &upper1, &empty1);
+	range_deserialize(typcache, r2, &lower2, &upper2, &empty2);

 	if (empty1 || empty2)
 	{
@@ -580,13 +645,13 @@ sort_item_cmp(const void *a, const void *b)
 		lower2.infinite || upper2.infinite)
 	{
 		if (lower1.infinite && lower2.infinite)
-			return range_cmp_bounds(fcinfo, &upper1, &upper2);
+			return range_cmp_bounds(typcache, &upper1, &upper2);
 		else if (lower1.infinite)
 			return -1;
 		else if (lower2.infinite)
 			return 1;
 		else if (upper1.infinite && upper2.infinite)
-			return -1 * range_cmp_bounds(fcinfo, &lower1, &lower2);
+			return -1 * range_cmp_bounds(typcache, &lower1, &lower2);
 		else if (upper1.infinite)
 			return 1;
 		else if (upper2.infinite)
@@ -595,8 +660,8 @@ sort_item_cmp(const void *a, const void *b)
 			Assert(false);
 	}

-	if ((cmp = range_cmp_bounds(fcinfo, &lower1, &lower2)) != 0)
+	if ((cmp = range_cmp_bounds(typcache, &lower1, &lower2)) != 0)
 		return cmp;

-	return range_cmp_bounds(fcinfo, &upper1, &upper2);
+	return range_cmp_bounds(typcache, &upper1, &upper2);
 }
--- a/src/backend/utils/cache/typcache.c
+++ b/src/backend/utils/cache/typcache.c
@@ -51,6 +51,7 @@
 #include "catalog/indexing.h"
 #include "catalog/pg_enum.h"
 #include "catalog/pg_operator.h"
+#include "catalog/pg_range.h"
 #include "catalog/pg_type.h"
 #include "commands/defrem.h"
 #include "utils/builtins.h"
@@ -120,6 +121,7 @@ static int32 RecordCacheArrayLen = 0;	/* allocated length of array */
 static int32 NextRecordTypmod = 0;		/* number of entries used */

 static void load_typcache_tupdesc(TypeCacheEntry *typentry);
+static void load_rangetype_info(TypeCacheEntry *typentry);
 static bool array_element_has_equality(TypeCacheEntry *typentry);
 static bool array_element_has_compare(TypeCacheEntry *typentry);
 static bool array_element_has_hashing(TypeCacheEntry *typentry);
@@ -205,6 +207,7 @@ lookup_type_cache(Oid type_id, int flags)
 		typentry->typlen = typtup->typlen;
 		typentry->typbyval = typtup->typbyval;
 		typentry->typalign = typtup->typalign;
+		typentry->typstorage = typtup->typstorage;
 		typentry->typtype = typtup->typtype;
 		typentry->typrelid = typtup->typrelid;

@@ -448,6 +451,16 @@ lookup_type_cache(Oid type_id, int flags)
 		load_typcache_tupdesc(typentry);
 	}

+	/*
+	 * If requested, get information about a range type
+	 */
+	if ((flags & TYPECACHE_RANGE_INFO) &&
+		typentry->rngelemtype == NULL &&
+		typentry->typtype == TYPTYPE_RANGE)
+	{
+		load_rangetype_info(typentry);
+	}
+
 	return typentry;
 }

@@ -479,6 +492,62 @@ load_typcache_tupdesc(TypeCacheEntry *typentry)
 	relation_close(rel, AccessShareLock);
 }

+/*
+ * load_rangetype_info --- helper routine to set up range type information
+ */
+static void
+load_rangetype_info(TypeCacheEntry *typentry)
+{
+	Form_pg_range pg_range;
+	HeapTuple	tup;
+	Oid			subtypeOid;
+	Oid			opclassOid;
+	Oid			canonicalOid;
+	Oid			subdiffOid;
+	Oid			opfamilyOid;
+	Oid			opcintype;
+	Oid			cmpFnOid;
+
+	/* get information from pg_range */
+	tup = SearchSysCache1(RANGETYPE, ObjectIdGetDatum(typentry->type_id));
+	/* should not fail, since we already checked typtype ... */
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for range type %u",
+			 typentry->type_id);
+	pg_range = (Form_pg_range) GETSTRUCT(tup);
+
+	subtypeOid = pg_range->rngsubtype;
+	typentry->rng_collation = pg_range->rngcollation;
+	opclassOid = pg_range->rngsubopc;
+	canonicalOid = pg_range->rngcanonical;
+	subdiffOid = pg_range->rngsubdiff;
+
+	ReleaseSysCache(tup);
+
+	/* get opclass properties and look up the comparison function */
+	opfamilyOid = get_opclass_family(opclassOid);
+	opcintype = get_opclass_input_type(opclassOid);
+
+	cmpFnOid = get_opfamily_proc(opfamilyOid, opcintype, opcintype,
+								 BTORDER_PROC);
+	if (!RegProcedureIsValid(cmpFnOid))
+		elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
+			 BTORDER_PROC, opcintype, opcintype, opfamilyOid);
+
+	/* set up cached fmgrinfo structs */
+	fmgr_info_cxt(cmpFnOid, &typentry->rng_cmp_proc_finfo,
+				  CacheMemoryContext);
+	if (OidIsValid(canonicalOid))
+		fmgr_info_cxt(canonicalOid, &typentry->rng_canonical_finfo,
+					  CacheMemoryContext);
+	if (OidIsValid(subdiffOid))
+		fmgr_info_cxt(subdiffOid, &typentry->rng_subdiff_finfo,
+					  CacheMemoryContext);
+
+	/* Lastly, set up link to the element type --- this marks data valid */
+	typentry->rngelemtype = lookup_type_cache(subtypeOid, 0);
+}
+

 /*
 * array_element_has_equality and friends are helper routines to check