utils/adt/rangetypes_gist.c

/*-------------------------------------------------------------------------
 *
 * rangetypes_gist.c
 *	  GiST support for range types.
 *
 * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/utils/adt/rangetypes_gist.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/gist.h"
#include "access/stratnum.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/rangetypes.h"


/*
 * Range class properties used to segregate different classes of ranges in
 * GiST.  Each unique combination of properties is a class.  CLS_EMPTY cannot
 * be combined with anything else.
 */
#define CLS_NORMAL			0	/* Ordinary finite range (no bits set) */
#define CLS_LOWER_INF		1	/* Lower bound is infinity */
#define CLS_UPPER_INF		2	/* Upper bound is infinity */
#define CLS_CONTAIN_EMPTY	4	/* Contains underlying empty ranges */
#define CLS_EMPTY			8	/* Special class for empty ranges */

#define CLS_COUNT			9	/* # of classes; includes all combinations of
								 * properties. CLS_EMPTY doesn't combine with
								 * anything else, so it's only 2^3 + 1. */

/*
 * Minimum accepted ratio of split for items of the same class.  If the items
 * are of different classes, we will separate along those lines regardless of
 * the ratio.
 */
#define LIMIT_RATIO  0.3

/* Constants for fixed penalty values */
#define INFINITE_BOUND_PENALTY	2.0
#define CONTAIN_EMPTY_PENALTY  1.0
#define DEFAULT_SUBTYPE_DIFF_PENALTY  1.0

/*
 * Per-item data for range_gist_single_sorting_split.
 */
typedef struct
{
	int			index;
	RangeBound	bound;
} SingleBoundSortItem;

/* place on left or right side of split? */
typedef enum
{
	SPLIT_LEFT = 0,				/* makes initialization to SPLIT_LEFT easier */
	SPLIT_RIGHT
} SplitLR;

/*
 * Context for range_gist_consider_split.
 */
typedef struct
{
	TypeCacheEntry *typcache;	/* typcache for range type */
	bool		has_subtype_diff;		/* does it have subtype_diff? */
	int			entries_count;	/* total number of entries being split */

	/* Information about currently selected split follows */

	bool		first;			/* true if no split was selected yet */

	RangeBound *left_upper;		/* upper bound of left interval */
	RangeBound *right_lower;	/* lower bound of right interval */

	float4		ratio;			/* split ratio */
	float4		overlap;		/* overlap between left and right predicate */
	int			common_left;	/* # common entries destined for each side */
	int			common_right;
} ConsiderSplitContext;

/*
 * Bounds extracted from a non-empty range, for use in
 * range_gist_double_sorting_split.
 */
typedef struct
{
	RangeBound	lower;
	RangeBound	upper;
} NonEmptyRange;

/*
 * Represents information about an entry that can be placed in either group
 * without affecting overlap over selected axis ("common entry").
 */
typedef struct
{
	/* Index of entry in the initial array */
	int			index;
	/* Delta between closeness of range to each of the two groups */
	double		delta;
} CommonEntry;

/* Helper macros to place an entry in the left or right group during split */
/* Note direct access to variables v, typcache, left_range, right_range */
#define PLACE_LEFT(range, off)					\
	do {										\
		if (v->spl_nleft > 0)					\
			left_range = range_super_union(typcache, left_range, range); \
		else									\
			left_range = (range);				\
		v->spl_left[v->spl_nleft++] = (off);	\
	} while(0)

#define PLACE_RIGHT(range, off)					\
	do {										\
		if (v->spl_nright > 0)					\
			right_range = range_super_union(typcache, right_range, range); \
		else									\
			right_range = (range);				\
		v->spl_right[v->spl_nright++] = (off);	\
	} while(0)

/* Copy a RangeType datum (hardwires typbyval and typlen for ranges...) */
#define rangeCopy(r) \
	((RangeType *) DatumGetPointer(datumCopy(PointerGetDatum(r), \
											 false, -1)))

static RangeType *range_super_union(TypeCacheEntry *typcache, RangeType *r1,
				  RangeType *r2);
static bool range_gist_consistent_int(TypeCacheEntry *typcache,
						  StrategyNumber strategy, RangeType *key,
						  Datum query);
static bool range_gist_consistent_leaf(TypeCacheEntry *typcache,
						   StrategyNumber strategy, RangeType *key,
						   Datum query);
static void range_gist_fallback_split(TypeCacheEntry *typcache,
						  GistEntryVector *entryvec,
						  GIST_SPLITVEC *v);
static void range_gist_class_split(TypeCacheEntry *typcache,
					   GistEntryVector *entryvec,
					   GIST_SPLITVEC *v,
					   SplitLR *classes_groups);
static void range_gist_single_sorting_split(TypeCacheEntry *typcache,
								GistEntryVector *entryvec,
								GIST_SPLITVEC *v,
								bool use_upper_bound);
static void range_gist_double_sorting_split(TypeCacheEntry *typcache,
								GistEntryVector *entryvec,
								GIST_SPLITVEC *v);
static void range_gist_consider_split(ConsiderSplitContext *context,
						  RangeBound *right_lower, int min_left_count,
						  RangeBound *left_upper, int max_left_count);
static int	get_gist_range_class(RangeType *range);
static int	single_bound_cmp(const void *a, const void *b, void *arg);
static int	interval_cmp_lower(const void *a, const void *b, void *arg);
static int	interval_cmp_upper(const void *a, const void *b, void *arg);
static int	common_entry_cmp(const void *i1, const void *i2);
static float8 call_subtype_diff(TypeCacheEntry *typcache,
				  Datum val1, Datum val2);


/* GiST query consistency check */
Datum
range_gist_consistent(PG_FUNCTION_ARGS)
{
	GISTENTRY  *entry = (GISTENTRY *) PG_GETARG_POINTER(0);
	Datum		query = PG_GETARG_DATUM(1);
	StrategyNumber strategy = (StrategyNumber) PG_GETARG_UINT16(2);

	/* Oid subtype = PG_GETARG_OID(3); */
	bool	   *recheck = (bool *) PG_GETARG_POINTER(4);
	RangeType  *key = DatumGetRangeType(entry->key);
	TypeCacheEntry *typcache;

	/* All operators served by this function are exact */
	*recheck = false;

	typcache = range_get_typcache(fcinfo, RangeTypeGetOid(key));

	if (GIST_LEAF(entry))
		PG_RETURN_BOOL(range_gist_consistent_leaf(typcache, strategy,
												  key, query));
	else
		PG_RETURN_BOOL(range_gist_consistent_int(typcache, strategy,
												 key, query));
}

/* form union range */
Datum
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(typcache, result_range,
										 DatumGetRangeType(ent[i].key));
	}

	PG_RETURN_RANGE(result_range);
}

/* compress, decompress, fetch are no-ops */
Datum
range_gist_compress(PG_FUNCTION_ARGS)
{
	GISTENTRY  *entry = (GISTENTRY *) PG_GETARG_POINTER(0);

	PG_RETURN_POINTER(entry);
}

Datum
range_gist_decompress(PG_FUNCTION_ARGS)
{
	GISTENTRY  *entry = (GISTENTRY *) PG_GETARG_POINTER(0);

	PG_RETURN_POINTER(entry);
}

Datum
range_gist_fetch(PG_FUNCTION_ARGS)
{
	GISTENTRY  *entry = (GISTENTRY *) PG_GETARG_POINTER(0);

	PG_RETURN_POINTER(entry);
}

/*
 * GiST page split penalty function.
 *
 * The penalty function has the following goals (in order from most to least
 * important):
 * - Keep normal ranges separate
 * - Avoid broadening the class of the original predicate
 * - Avoid broadening (as determined by subtype_diff) the original predicate
 * - Favor adding ranges to narrower original predicates
 */
Datum
range_gist_penalty(PG_FUNCTION_ARGS)
{
	GISTENTRY  *origentry = (GISTENTRY *) PG_GETARG_POINTER(0);
	GISTENTRY  *newentry = (GISTENTRY *) PG_GETARG_POINTER(1);
	float	   *penalty = (float *) PG_GETARG_POINTER(2);
	RangeType  *orig = DatumGetRangeType(origentry->key);
	RangeType  *new = DatumGetRangeType(newentry->key);
	TypeCacheEntry *typcache;
	bool		has_subtype_diff;
	RangeBound	orig_lower,
				new_lower,
				orig_upper,
				new_upper;
	bool		orig_empty,
				new_empty;

	if (RangeTypeGetOid(orig) != RangeTypeGetOid(new))
		elog(ERROR, "range types do not match");

	typcache = range_get_typcache(fcinfo, RangeTypeGetOid(orig));

	has_subtype_diff = OidIsValid(typcache->rng_subdiff_finfo.fn_oid);

	range_deserialize(typcache, orig, &orig_lower, &orig_upper, &orig_empty);
	range_deserialize(typcache, new, &new_lower, &new_upper, &new_empty);

	/*
	 * Distinct branches for handling distinct classes of ranges.  Note that
	 * penalty values only need to be commensurate within the same class of
	 * new range.
	 */
	if (new_empty)
	{
		/* Handle insertion of empty range */
		if (orig_empty)
		{
			/*
			 * The best case is to insert it to empty original range.
			 * Insertion here means no broadening of original range. Also
			 * original range is the most narrow.
			 */
			*penalty = 0.0;
		}
		else if (RangeIsOrContainsEmpty(orig))
		{
			/*
			 * The second case is to insert empty range into range which
			 * contains at least one underlying empty range.  There is still
			 * no broadening of original range, but original range is not as
			 * narrow as possible.
			 */
			*penalty = CONTAIN_EMPTY_PENALTY;
		}
		else if (orig_lower.infinite && orig_upper.infinite)
		{
			/*
			 * Original range requires broadening.  (-inf; +inf) is most far
			 * from normal range in this case.
			 */
			*penalty = 2 * CONTAIN_EMPTY_PENALTY;
		}
		else if (orig_lower.infinite || orig_upper.infinite)
		{
			/*
			 * (-inf, x) or (x, +inf) original ranges are closer to normal
			 * ranges, so it's worse to mix it with empty ranges.
			 */
			*penalty = 3 * CONTAIN_EMPTY_PENALTY;
		}
		else
		{
			/*
			 * The least preferred case is broadening of normal range.
			 */
			*penalty = 4 * CONTAIN_EMPTY_PENALTY;
		}
	}
	else if (new_lower.infinite && new_upper.infinite)
	{
		/* Handle insertion of (-inf, +inf) range */
		if (orig_lower.infinite && orig_upper.infinite)
		{
			/*
			 * Best case is inserting to (-inf, +inf) original range.
			 */
			*penalty = 0.0;
		}
		else if (orig_lower.infinite || orig_upper.infinite)
		{
			/*
			 * When original range is (-inf, x) or (x, +inf) it requires
			 * broadening of original range (extension of one bound to
			 * infinity).
			 */
			*penalty = INFINITE_BOUND_PENALTY;
		}
		else
		{
			/*
			 * Insertion to normal original range is least preferred.
			 */
			*penalty = 2 * INFINITE_BOUND_PENALTY;
		}

		if (RangeIsOrContainsEmpty(orig))
		{
			/*
			 * Original range is narrower when it doesn't contain empty
			 * ranges. Add additional penalty otherwise.
			 */
			*penalty += CONTAIN_EMPTY_PENALTY;
		}
	}
	else if (new_lower.infinite)
	{
		/* Handle insertion of (-inf, x) range */
		if (!orig_empty && orig_lower.infinite)
		{
			if (orig_upper.infinite)
			{
				/*
				 * (-inf, +inf) range won't be extended by insertion of (-inf,
				 * x) range. It's a less desirable case than insertion to
				 * (-inf, y) original range without extension, because in that
				 * case original range is narrower. But we can't express that
				 * in single float value.
				 */
				*penalty = 0.0;
			}
			else
			{
				if (range_cmp_bounds(typcache, &new_upper, &orig_upper) > 0)
				{
					/*
					 * Get extension of original range using subtype_diff. Use
					 * constant if subtype_diff unavailable.
					 */
					if (has_subtype_diff)
						*penalty = call_subtype_diff(typcache,
													 new_upper.val,
													 orig_upper.val);
					else
						*penalty = DEFAULT_SUBTYPE_DIFF_PENALTY;
				}
				else
				{
					/* No extension of original range */
					*penalty = 0.0;
				}
			}
		}
		else
		{
			/*
			 * If lower bound of original range is not -inf, then extension of
			 * it is infinity.
			 */
			*penalty = get_float4_infinity();
		}
	}
	else if (new_upper.infinite)
	{
		/* Handle insertion of (x, +inf) range */
		if (!orig_empty && orig_upper.infinite)
		{
			if (orig_lower.infinite)
			{
				/*
				 * (-inf, +inf) range won't be extended by insertion of (x,
				 * +inf) range. It's a less desirable case than insertion to
				 * (y, +inf) original range without extension, because in that
				 * case original range is narrower. But we can't express that
				 * in single float value.
				 */
				*penalty = 0.0;
			}
			else
			{
				if (range_cmp_bounds(typcache, &new_lower, &orig_lower) < 0)
				{
					/*
					 * Get extension of original range using subtype_diff. Use
					 * constant if subtype_diff unavailable.
					 */
					if (has_subtype_diff)
						*penalty = call_subtype_diff(typcache,
													 orig_lower.val,
													 new_lower.val);
					else
						*penalty = DEFAULT_SUBTYPE_DIFF_PENALTY;
				}
				else
				{
					/* No extension of original range */
					*penalty = 0.0;
				}
			}
		}
		else
		{
			/*
			 * If upper bound of original range is not +inf, then extension of
			 * it is infinity.
			 */
			*penalty = get_float4_infinity();
		}
	}
	else
	{
		/* Handle insertion of normal (non-empty, non-infinite) range */
		if (orig_empty || orig_lower.infinite || orig_upper.infinite)
		{
			/*
			 * Avoid mixing normal ranges with infinite and empty ranges.
			 */
			*penalty = get_float4_infinity();
		}
		else
		{
			/*
			 * Calculate extension of original range by calling subtype_diff.
			 * Use constant if subtype_diff unavailable.
			 */
			float8		diff = 0.0;

			if (range_cmp_bounds(typcache, &new_lower, &orig_lower) < 0)
			{
				if (has_subtype_diff)
					diff += call_subtype_diff(typcache,
											  orig_lower.val,
											  new_lower.val);
				else
					diff += DEFAULT_SUBTYPE_DIFF_PENALTY;
			}
			if (range_cmp_bounds(typcache, &new_upper, &orig_upper) > 0)
			{
				if (has_subtype_diff)
					diff += call_subtype_diff(typcache,
											  new_upper.val,
											  orig_upper.val);
				else
					diff += DEFAULT_SUBTYPE_DIFF_PENALTY;
			}
			*penalty = diff;
		}
	}

	PG_RETURN_POINTER(penalty);
}

/*
 * The GiST PickSplit method for ranges
 *
 * Primarily, we try to segregate ranges of different classes.  If splitting
 * ranges of the same class, use the appropriate split method for that class.
 */
Datum
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;
	int			nbytes;
	OffsetNumber maxoff;
	int			count_in_classes[CLS_COUNT];
	int			j;
	int			non_empty_classes_count = 0;
	int			biggest_class = -1;
	int			biggest_class_count = 0;
	int			total_count;

	/* 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));

	maxoff = entryvec->n - 1;
	nbytes = (maxoff + 1) * sizeof(OffsetNumber);
	v->spl_left = (OffsetNumber *) palloc(nbytes);
	v->spl_right = (OffsetNumber *) palloc(nbytes);

	/*
	 * Get count distribution of range classes.
	 */
	memset(count_in_classes, 0, sizeof(count_in_classes));
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		RangeType  *range = DatumGetRangeType(entryvec->vector[i].key);

		count_in_classes[get_gist_range_class(range)]++;
	}

	/*
	 * Count non-empty classes and find biggest class.
	 */
	total_count = maxoff;
	for (j = 0; j < CLS_COUNT; j++)
	{
		if (count_in_classes[j] > 0)
		{
			if (count_in_classes[j] > biggest_class_count)
			{
				biggest_class_count = count_in_classes[j];
				biggest_class = j;
			}
			non_empty_classes_count++;
		}
	}

	Assert(non_empty_classes_count > 0);

	if (non_empty_classes_count == 1)
	{
		/* One non-empty class, so split inside class */
		if ((biggest_class & ~CLS_CONTAIN_EMPTY) == CLS_NORMAL)
		{
			/* double sorting split for normal ranges */
			range_gist_double_sorting_split(typcache, entryvec, v);
		}
		else if ((biggest_class & ~CLS_CONTAIN_EMPTY) == CLS_LOWER_INF)
		{
			/* upper bound sorting split for (-inf, x) ranges */
			range_gist_single_sorting_split(typcache, entryvec, v, true);
		}
		else if ((biggest_class & ~CLS_CONTAIN_EMPTY) == CLS_UPPER_INF)
		{
			/* lower bound sorting split for (x, +inf) ranges */
			range_gist_single_sorting_split(typcache, entryvec, v, false);
		}
		else
		{
			/* trivial split for all (-inf, +inf) or all empty ranges */
			range_gist_fallback_split(typcache, entryvec, v);
		}
	}
	else
	{
		/*
		 * Class based split.
		 *
		 * To which side of the split should each class go?  Initialize them
		 * all to go to the left side.
		 */
		SplitLR		classes_groups[CLS_COUNT];

		memset(classes_groups, 0, sizeof(classes_groups));

		if (count_in_classes[CLS_NORMAL] > 0)
		{
			/* separate normal ranges if any */
			classes_groups[CLS_NORMAL] = SPLIT_RIGHT;
		}
		else
		{
			/*----------
			 * Try to split classes in one of two ways:
			 *	1) containing infinities - not containing infinities
			 *	2) containing empty - not containing empty
			 *
			 * Select the way which balances the ranges between left and right
			 * the best. If split in these ways is not possible, there are at
			 * most 3 classes, so just separate biggest class.
			 *----------
			 */
			int			infCount,
						nonInfCount;
			int			emptyCount,
						nonEmptyCount;

			nonInfCount =
				count_in_classes[CLS_NORMAL] +
				count_in_classes[CLS_CONTAIN_EMPTY] +
				count_in_classes[CLS_EMPTY];
			infCount = total_count - nonInfCount;

			nonEmptyCount =
				count_in_classes[CLS_NORMAL] +
				count_in_classes[CLS_LOWER_INF] +
				count_in_classes[CLS_UPPER_INF] +
				count_in_classes[CLS_LOWER_INF | CLS_UPPER_INF];
			emptyCount = total_count - nonEmptyCount;

			if (infCount > 0 && nonInfCount > 0 &&
				(Abs(infCount - nonInfCount) <=
				 Abs(emptyCount - nonEmptyCount)))
			{
				classes_groups[CLS_NORMAL] = SPLIT_RIGHT;
				classes_groups[CLS_CONTAIN_EMPTY] = SPLIT_RIGHT;
				classes_groups[CLS_EMPTY] = SPLIT_RIGHT;
			}
			else if (emptyCount > 0 && nonEmptyCount > 0)
			{
				classes_groups[CLS_NORMAL] = SPLIT_RIGHT;
				classes_groups[CLS_LOWER_INF] = SPLIT_RIGHT;
				classes_groups[CLS_UPPER_INF] = SPLIT_RIGHT;
				classes_groups[CLS_LOWER_INF | CLS_UPPER_INF] = SPLIT_RIGHT;
			}
			else
			{
				/*
				 * Either total_count == emptyCount or total_count ==
				 * infCount.
				 */
				classes_groups[biggest_class] = SPLIT_RIGHT;
			}
		}

		range_gist_class_split(typcache, entryvec, v, classes_groups);
	}

	PG_RETURN_POINTER(v);
}

/* equality comparator for GiST */
Datum
range_gist_same(PG_FUNCTION_ARGS)
{
	RangeType  *r1 = PG_GETARG_RANGE(0);
	RangeType  *r2 = PG_GETARG_RANGE(1);
	bool	   *result = (bool *) PG_GETARG_POINTER(2);

	/*
	 * range_eq will ignore the RANGE_CONTAIN_EMPTY flag, so we have to check
	 * that for ourselves.  More generally, if the entries have been properly
	 * normalized, then unequal flags bytes must mean unequal ranges ... so
	 * let's just test all the flag bits at once.
	 */
	if (range_get_flags(r1) != range_get_flags(r2))
		*result = false;
	else
	{
		TypeCacheEntry *typcache;

		typcache = range_get_typcache(fcinfo, RangeTypeGetOid(r1));

		*result = range_eq_internal(typcache, r1, r2);
	}

	PG_RETURN_POINTER(result);
}

/*
 *----------------------------------------------------------
 * STATIC FUNCTIONS
 *----------------------------------------------------------
 */

/*
 * Return the smallest range that contains r1 and r2
 *
 * This differs from regular range_union in two critical ways:
 * 1. It won't throw an error for non-adjacent r1 and r2, but just absorb
 * the intervening values into the result range.
 * 2. We track whether any empty range has been union'd into the result,
 * so that contained_by searches can be indexed.  Note that this means
 * that *all* unions formed within the GiST index must go through here.
 */
static RangeType *
range_super_union(TypeCacheEntry *typcache, RangeType *r1, RangeType *r2)
{
	RangeType  *result;
	RangeBound	lower1,
				lower2;
	RangeBound	upper1,
				upper2;
	bool		empty1,
				empty2;
	char		flags1,
				flags2;
	RangeBound *result_lower;
	RangeBound *result_upper;

	range_deserialize(typcache, r1, &lower1, &upper1, &empty1);
	range_deserialize(typcache, r2, &lower2, &upper2, &empty2);
	flags1 = range_get_flags(r1);
	flags2 = range_get_flags(r2);

	if (empty1)
	{
		/* We can return r2 as-is if it already is or contains empty */
		if (flags2 & (RANGE_EMPTY | RANGE_CONTAIN_EMPTY))
			return r2;
		/* Else we'd better copy it (modify-in-place isn't safe) */
		r2 = rangeCopy(r2);
		range_set_contain_empty(r2);
		return r2;
	}
	if (empty2)
	{
		/* We can return r1 as-is if it already is or contains empty */
		if (flags1 & (RANGE_EMPTY | RANGE_CONTAIN_EMPTY))
			return r1;
		/* Else we'd better copy it (modify-in-place isn't safe) */
		r1 = rangeCopy(r1);
		range_set_contain_empty(r1);
		return r1;
	}

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

	if (range_cmp_bounds(typcache, &upper1, &upper2) >= 0)
		result_upper = &upper1;
	else
		result_upper = &upper2;

	/* optimization to avoid constructing a new range */
	if (result_lower == &lower1 && result_upper == &upper1 &&
		((flags1 & RANGE_CONTAIN_EMPTY) || !(flags2 & RANGE_CONTAIN_EMPTY)))
		return r1;
	if (result_lower == &lower2 && result_upper == &upper2 &&
		((flags2 & RANGE_CONTAIN_EMPTY) || !(flags1 & RANGE_CONTAIN_EMPTY)))
		return r2;

	result = make_range(typcache, result_lower, result_upper, false);

	if ((flags1 & RANGE_CONTAIN_EMPTY) || (flags2 & RANGE_CONTAIN_EMPTY))
		range_set_contain_empty(result);

	return result;
}

/*
 * GiST consistent test on an index internal page
 */
static bool
range_gist_consistent_int(TypeCacheEntry *typcache, StrategyNumber strategy,
						  RangeType *key, Datum query)
{
	switch (strategy)
	{
		case RANGESTRAT_BEFORE:
			if (RangeIsEmpty(key) || RangeIsEmpty(DatumGetRangeType(query)))
				return false;
			return (!range_overright_internal(typcache, key,
											  DatumGetRangeType(query)));
		case RANGESTRAT_OVERLEFT:
			if (RangeIsEmpty(key) || RangeIsEmpty(DatumGetRangeType(query)))
				return false;
			return (!range_after_internal(typcache, key,
										  DatumGetRangeType(query)));
		case RANGESTRAT_OVERLAPS:
			return range_overlaps_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_OVERRIGHT:
			if (RangeIsEmpty(key) || RangeIsEmpty(DatumGetRangeType(query)))
				return false;
			return (!range_before_internal(typcache, key,
										   DatumGetRangeType(query)));
		case RANGESTRAT_AFTER:
			if (RangeIsEmpty(key) || RangeIsEmpty(DatumGetRangeType(query)))
				return false;
			return (!range_overleft_internal(typcache, key,
											 DatumGetRangeType(query)));
		case RANGESTRAT_ADJACENT:
			if (RangeIsEmpty(key) || RangeIsEmpty(DatumGetRangeType(query)))
				return false;
			if (range_adjacent_internal(typcache, key,
										DatumGetRangeType(query)))
				return true;
			return range_overlaps_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINS:
			return range_contains_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINED_BY:

			/*
			 * Empty ranges are contained by anything, so if key is or
			 * contains any empty ranges, we must descend into it.  Otherwise,
			 * descend only if key overlaps the query.
			 */
			if (RangeIsOrContainsEmpty(key))
				return true;
			return range_overlaps_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINS_ELEM:
			return range_contains_elem_internal(typcache, key, query);
		case RANGESTRAT_EQ:

			/*
			 * If query is empty, descend only if the key is or contains any
			 * empty ranges.  Otherwise, descend if key contains query.
			 */
			if (RangeIsEmpty(DatumGetRangeType(query)))
				return RangeIsOrContainsEmpty(key);
			return range_contains_internal(typcache, key,
										   DatumGetRangeType(query));
		default:
			elog(ERROR, "unrecognized range strategy: %d", strategy);
			return false;		/* keep compiler quiet */
	}
}

/*
 * GiST consistent test on an index leaf page
 */
static bool
range_gist_consistent_leaf(TypeCacheEntry *typcache, StrategyNumber strategy,
						   RangeType *key, Datum query)
{
	switch (strategy)
	{
		case RANGESTRAT_BEFORE:
			return range_before_internal(typcache, key,
										 DatumGetRangeType(query));
		case RANGESTRAT_OVERLEFT:
			return range_overleft_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_OVERLAPS:
			return range_overlaps_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_OVERRIGHT:
			return range_overright_internal(typcache, key,
											DatumGetRangeType(query));
		case RANGESTRAT_AFTER:
			return range_after_internal(typcache, key,
										DatumGetRangeType(query));
		case RANGESTRAT_ADJACENT:
			return range_adjacent_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINS:
			return range_contains_internal(typcache, key,
										   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINED_BY:
			return range_contained_by_internal(typcache, key,
											   DatumGetRangeType(query));
		case RANGESTRAT_CONTAINS_ELEM:
			return range_contains_elem_internal(typcache, key, query);
		case RANGESTRAT_EQ:
			return range_eq_internal(typcache, key, DatumGetRangeType(query));
		default:
			elog(ERROR, "unrecognized range strategy: %d", strategy);
			return false;		/* keep compiler quiet */
	}
}

/*
 * Trivial split: half of entries will be placed on one page
 * and the other half on the other page.
 */
static void
range_gist_fallback_split(TypeCacheEntry *typcache,
						  GistEntryVector *entryvec,
						  GIST_SPLITVEC *v)
{
	RangeType  *left_range = NULL;
	RangeType  *right_range = NULL;
	OffsetNumber i,
				maxoff,
				split_idx;

	maxoff = entryvec->n - 1;
	/* Split entries before this to left page, after to right: */
	split_idx = (maxoff - FirstOffsetNumber) / 2 + FirstOffsetNumber;

	v->spl_nleft = 0;
	v->spl_nright = 0;
	for (i = FirstOffsetNumber; i <= maxoff; i++)
	{
		RangeType  *range = DatumGetRangeType(entryvec->vector[i].key);

		if (i < split_idx)
			PLACE_LEFT(range, i);
		else
			PLACE_RIGHT(range, i);
	}

	v->spl_ldatum = RangeTypeGetDatum(left_range);
	v->spl_rdatum = RangeTypeGetDatum(right_range);
}

/*
 * Split based on classes of ranges.
 *
 * See get_gist_range_class for class definitions.
 * classes_groups is an array of length CLS_COUNT indicating the side of the
 * split to which each class should go.
 */
static void
range_gist_class_split(TypeCacheEntry *typcache,
					   GistEntryVector *entryvec,
					   GIST_SPLITVEC *v,
					   SplitLR *classes_groups)
{
	RangeType  *left_range = NULL;
	RangeType  *right_range = NULL;
	OffsetNumber i,
				maxoff;

	maxoff = entryvec->n - 1;

	v->spl_nleft = 0;
	v->spl_nright = 0;
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		RangeType  *range = DatumGetRangeType(entryvec->vector[i].key);
		int			class;

		/* Get class of range */
		class = get_gist_range_class(range);

		/* Place range to appropriate page */
		if (classes_groups[class] == SPLIT_LEFT)
			PLACE_LEFT(range, i);
		else
		{
			Assert(classes_groups[class] == SPLIT_RIGHT);
			PLACE_RIGHT(range, i);
		}
	}

	v->spl_ldatum = RangeTypeGetDatum(left_range);
	v->spl_rdatum = RangeTypeGetDatum(right_range);
}

/*
 * Sorting based split. First half of entries according to the sort will be
 * placed to one page, and second half of entries will be placed to other
 * page. use_upper_bound parameter indicates whether to use upper or lower
 * bound for sorting.
 */
static void
range_gist_single_sorting_split(TypeCacheEntry *typcache,
								GistEntryVector *entryvec,
								GIST_SPLITVEC *v,
								bool use_upper_bound)
{
	SingleBoundSortItem *sortItems;
	RangeType  *left_range = NULL;
	RangeType  *right_range = NULL;
	OffsetNumber i,
				maxoff,
				split_idx;

	maxoff = entryvec->n - 1;

	sortItems = (SingleBoundSortItem *)
		palloc(maxoff * sizeof(SingleBoundSortItem));

	/*
	 * Prepare auxiliary array and sort the values.
	 */
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		RangeType  *range = DatumGetRangeType(entryvec->vector[i].key);
		RangeBound	bound2;
		bool		empty;

		sortItems[i - 1].index = i;
		/* Put appropriate bound into array */
		if (use_upper_bound)
			range_deserialize(typcache, range, &bound2,
							  &sortItems[i - 1].bound, &empty);
		else
			range_deserialize(typcache, range, &sortItems[i - 1].bound,
							  &bound2, &empty);
		Assert(!empty);
	}

	qsort_arg(sortItems, maxoff, sizeof(SingleBoundSortItem),
			  single_bound_cmp, typcache);

	split_idx = maxoff / 2;

	v->spl_nleft = 0;
	v->spl_nright = 0;

	for (i = 0; i < maxoff; i++)
	{
		int			idx = sortItems[i].index;
		RangeType  *range = DatumGetRangeType(entryvec->vector[idx].key);

		if (i < split_idx)
			PLACE_LEFT(range, idx);
		else
			PLACE_RIGHT(range, idx);
	}

	v->spl_ldatum = RangeTypeGetDatum(left_range);
	v->spl_rdatum = RangeTypeGetDatum(right_range);
}

/*
 * Double sorting split algorithm.
 *
 * The algorithm considers dividing ranges into two groups. The first (left)
 * group contains general left bound. The second (right) group contains
 * general right bound. The challenge is to find upper bound of left group
 * and lower bound of right group so that overlap of groups is minimal and
 * ratio of distribution is acceptable. Algorithm finds for each lower bound of
 * right group minimal upper bound of left group, and for each upper bound of
 * left group maximal lower bound of right group. For each found pair
 * range_gist_consider_split considers replacement of currently selected
 * split with the new one.
 *
 * After that, all the entries are divided into three groups:
 * 1) Entries which should be placed to the left group
 * 2) Entries which should be placed to the right group
 * 3) "Common entries" which can be placed to either group without affecting
 *	  amount of overlap.
 *
 * The common ranges are distributed by difference of distance from lower
 * bound of common range to lower bound of right group and distance from upper
 * bound of common range to upper bound of left group.
 *
 * For details see:
 * "A new double sorting-based node splitting algorithm for R-tree",
 * A. Korotkov
 * http://syrcose.ispras.ru/2011/files/SYRCoSE2011_Proceedings.pdf#page=36
 */
static void
range_gist_double_sorting_split(TypeCacheEntry *typcache,
								GistEntryVector *entryvec,
								GIST_SPLITVEC *v)
{
	ConsiderSplitContext context;
	OffsetNumber i,
				maxoff;
	RangeType  *range,
			   *left_range = NULL,
			   *right_range = NULL;
	int			common_entries_count;
	NonEmptyRange *by_lower,
			   *by_upper;
	CommonEntry *common_entries;
	int			nentries,
				i1,
				i2;
	RangeBound *right_lower,
			   *left_upper;

	memset(&context, 0, sizeof(ConsiderSplitContext));
	context.typcache = typcache;
	context.has_subtype_diff = OidIsValid(typcache->rng_subdiff_finfo.fn_oid);

	maxoff = entryvec->n - 1;
	nentries = context.entries_count = maxoff - FirstOffsetNumber + 1;
	context.first = true;

	/* Allocate arrays for sorted range bounds */
	by_lower = (NonEmptyRange *) palloc(nentries * sizeof(NonEmptyRange));
	by_upper = (NonEmptyRange *) palloc(nentries * sizeof(NonEmptyRange));

	/* Fill arrays of bounds */
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		RangeType  *range = DatumGetRangeType(entryvec->vector[i].key);
		bool		empty;

		range_deserialize(typcache, range,
						  &by_lower[i - FirstOffsetNumber].lower,
						  &by_lower[i - FirstOffsetNumber].upper,
						  &empty);
		Assert(!empty);
	}

	/*
	 * Make two arrays of range bounds: one sorted by lower bound and another
	 * sorted by upper bound.
	 */
	memcpy(by_upper, by_lower, nentries * sizeof(NonEmptyRange));
	qsort_arg(by_lower, nentries, sizeof(NonEmptyRange),
			  interval_cmp_lower, typcache);
	qsort_arg(by_upper, nentries, sizeof(NonEmptyRange),
			  interval_cmp_upper, typcache);

	/*----------
	 * The goal is to form a left and right range, so that every entry
	 * range is contained by either left or right interval (or both).
	 *
	 * For example, with the ranges (0,1), (1,3), (2,3), (2,4):
	 *
	 * 0 1 2 3 4
	 * +-+
	 *	 +---+
	 *	   +-+
	 *	   +---+
	 *
	 * The left and right ranges are of the form (0,a) and (b,4).
	 * We first consider splits where b is the lower bound of an entry.
	 * We iterate through all entries, and for each b, calculate the
	 * smallest possible a. Then we consider splits where a is the
	 * upper bound of an entry, and for each a, calculate the greatest
	 * possible b.
	 *
	 * In the above example, the first loop would consider splits:
	 * b=0: (0,1)-(0,4)
	 * b=1: (0,1)-(1,4)
	 * b=2: (0,3)-(2,4)
	 *
	 * And the second loop:
	 * a=1: (0,1)-(1,4)
	 * a=3: (0,3)-(2,4)
	 * a=4: (0,4)-(2,4)
	 *----------
	 */

	/*
	 * Iterate over lower bound of right group, finding smallest possible
	 * upper bound of left group.
	 */
	i1 = 0;
	i2 = 0;
	right_lower = &by_lower[i1].lower;
	left_upper = &by_upper[i2].lower;
	while (true)
	{
		/*
		 * Find next lower bound of right group.
		 */
		while (i1 < nentries &&
			   range_cmp_bounds(typcache, right_lower,
								&by_lower[i1].lower) == 0)
		{
			if (range_cmp_bounds(typcache, &by_lower[i1].upper,
								 left_upper) > 0)
				left_upper = &by_lower[i1].upper;
			i1++;
		}
		if (i1 >= nentries)
			break;
		right_lower = &by_lower[i1].lower;

		/*
		 * Find count of ranges which anyway should be placed to the left
		 * group.
		 */
		while (i2 < nentries &&
			   range_cmp_bounds(typcache, &by_upper[i2].upper,
								left_upper) <= 0)
			i2++;

		/*
		 * Consider found split to see if it's better than what we had.
		 */
		range_gist_consider_split(&context, right_lower, i1, left_upper, i2);
	}

	/*
	 * Iterate over upper bound of left group finding greatest possible lower
	 * bound of right group.
	 */
	i1 = nentries - 1;
	i2 = nentries - 1;
	right_lower = &by_lower[i1].upper;
	left_upper = &by_upper[i2].upper;
	while (true)
	{
		/*
		 * Find next upper bound of left group.
		 */
		while (i2 >= 0 &&
			   range_cmp_bounds(typcache, left_upper,
								&by_upper[i2].upper) == 0)
		{
			if (range_cmp_bounds(typcache, &by_upper[i2].lower,
								 right_lower) < 0)
				right_lower = &by_upper[i2].lower;
			i2--;
		}
		if (i2 < 0)
			break;
		left_upper = &by_upper[i2].upper;

		/*
		 * Find count of intervals which anyway should be placed to the right
		 * group.
		 */
		while (i1 >= 0 &&
			   range_cmp_bounds(typcache, &by_lower[i1].lower,
								right_lower) >= 0)
			i1--;

		/*
		 * Consider found split to see if it's better than what we had.
		 */
		range_gist_consider_split(&context, right_lower, i1 + 1,
								  left_upper, i2 + 1);
	}

	/*
	 * If we failed to find any acceptable splits, use trivial split.
	 */
	if (context.first)
	{
		range_gist_fallback_split(typcache, entryvec, v);
		return;
	}

	/*
	 * Ok, we have now selected bounds of the groups. Now we have to
	 * distribute entries themselves. At first we distribute entries which can
	 * be placed unambiguously and collect "common entries" to array.
	 */

	/* Allocate vectors for results */
	v->spl_left = (OffsetNumber *) palloc(nentries * sizeof(OffsetNumber));
	v->spl_right = (OffsetNumber *) palloc(nentries * sizeof(OffsetNumber));
	v->spl_nleft = 0;
	v->spl_nright = 0;

	/*
	 * Allocate an array for "common entries" - entries which can be placed to
	 * either group without affecting overlap along selected axis.
	 */
	common_entries_count = 0;
	common_entries = (CommonEntry *) palloc(nentries * sizeof(CommonEntry));

	/*
	 * Distribute entries which can be distributed unambiguously, and collect
	 * common entries.
	 */
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		RangeBound	lower,
					upper;
		bool		empty;

		/*
		 * Get upper and lower bounds along selected axis.
		 */
		range = DatumGetRangeType(entryvec->vector[i].key);

		range_deserialize(typcache, range, &lower, &upper, &empty);

		if (range_cmp_bounds(typcache, &upper, context.left_upper) <= 0)
		{
			/* Fits in the left group */
			if (range_cmp_bounds(typcache, &lower, context.right_lower) >= 0)
			{
				/* Fits also in the right group, so "common entry" */
				common_entries[common_entries_count].index = i;
				if (context.has_subtype_diff)
				{
					/*
					 * delta = (lower - context.right_lower) -
					 * (context.left_upper - upper)
					 */
					common_entries[common_entries_count].delta =
						call_subtype_diff(typcache,
										  lower.val,
										  context.right_lower->val) -
						call_subtype_diff(typcache,
										  context.left_upper->val,
										  upper.val);
				}
				else
				{
					/* Without subtype_diff, take all deltas as zero */
					common_entries[common_entries_count].delta = 0;
				}
				common_entries_count++;
			}
			else
			{
				/* Doesn't fit to the right group, so join to the left group */
				PLACE_LEFT(range, i);
			}
		}
		else
		{
			/*
			 * Each entry should fit on either left or right group. Since this
			 * entry didn't fit in the left group, it better fit in the right
			 * group.
			 */
			Assert(range_cmp_bounds(typcache, &lower,
									context.right_lower) >= 0);
			PLACE_RIGHT(range, i);
		}
	}

	/*
	 * Distribute "common entries", if any.
	 */
	if (common_entries_count > 0)
	{
		/*
		 * Sort "common entries" by calculated deltas in order to distribute
		 * the most ambiguous entries first.
		 */
		qsort(common_entries, common_entries_count, sizeof(CommonEntry),
			  common_entry_cmp);

		/*
		 * Distribute "common entries" between groups according to sorting.
		 */
		for (i = 0; i < common_entries_count; i++)
		{
			int			idx = common_entries[i].index;

			range = DatumGetRangeType(entryvec->vector[idx].key);

			/*
			 * Check if we have to place this entry in either group to achieve
			 * LIMIT_RATIO.
			 */
			if (i < context.common_left)
				PLACE_LEFT(range, idx);
			else
				PLACE_RIGHT(range, idx);
		}
	}

	v->spl_ldatum = PointerGetDatum(left_range);
	v->spl_rdatum = PointerGetDatum(right_range);
}

/*
 * Consider replacement of currently selected split with a better one
 * during range_gist_double_sorting_split.
 */
static void
range_gist_consider_split(ConsiderSplitContext *context,
						  RangeBound *right_lower, int min_left_count,
						  RangeBound *left_upper, int max_left_count)
{
	int			left_count,
				right_count;
	float4		ratio,
				overlap;

	/*
	 * Calculate entries distribution ratio assuming most uniform distribution
	 * of common entries.
	 */
	if (min_left_count >= (context->entries_count + 1) / 2)
		left_count = min_left_count;
	else if (max_left_count <= context->entries_count / 2)
		left_count = max_left_count;
	else
		left_count = context->entries_count / 2;
	right_count = context->entries_count - left_count;

	/*
	 * Ratio of split: quotient between size of smaller group and total
	 * entries count.  This is necessarily 0.5 or less; if it's less than
	 * LIMIT_RATIO then we will never accept the new split.
	 */
	ratio = ((float4) Min(left_count, right_count)) /
		((float4) context->entries_count);

	if (ratio > LIMIT_RATIO)
	{
		bool		selectthis = false;

		/*
		 * The ratio is acceptable, so compare current split with previously
		 * selected one. We search for minimal overlap (allowing negative
		 * values) and minimal ratio secondarily.  If subtype_diff is
		 * available, it's used for overlap measure.  Without subtype_diff we
		 * use number of "common entries" as an overlap measure.
		 */
		if (context->has_subtype_diff)
			overlap = call_subtype_diff(context->typcache,
										left_upper->val,
										right_lower->val);
		else
			overlap = max_left_count - min_left_count;

		/* If there is no previous selection, select this split */
		if (context->first)
			selectthis = true;
		else
		{
			/*
			 * Choose the new split if it has a smaller overlap, or same
			 * overlap but better ratio.
			 */
			if (overlap < context->overlap ||
				(overlap == context->overlap && ratio > context->ratio))
				selectthis = true;
		}

		if (selectthis)
		{
			/* save information about selected split */
			context->first = false;
			context->ratio = ratio;
			context->overlap = overlap;
			context->right_lower = right_lower;
			context->left_upper = left_upper;
			context->common_left = max_left_count - left_count;
			context->common_right = left_count - min_left_count;
		}
	}
}

/*
 * Find class number for range.
 *
 * The class number is a valid combination of the properties of the
 * range.  Note: the highest possible number is 8, because CLS_EMPTY
 * can't be combined with anything else.
 */
static int
get_gist_range_class(RangeType *range)
{
	int			classNumber;
	char		flags;

	flags = range_get_flags(range);
	if (flags & RANGE_EMPTY)
	{
		classNumber = CLS_EMPTY;
	}
	else
	{
		classNumber = 0;
		if (flags & RANGE_LB_INF)
			classNumber |= CLS_LOWER_INF;
		if (flags & RANGE_UB_INF)
			classNumber |= CLS_UPPER_INF;
		if (flags & RANGE_CONTAIN_EMPTY)
			classNumber |= CLS_CONTAIN_EMPTY;
	}
	return classNumber;
}

/*
 * Comparison function for range_gist_single_sorting_split.
 */
static int
single_bound_cmp(const void *a, const void *b, void *arg)
{
	SingleBoundSortItem *i1 = (SingleBoundSortItem *) a;
	SingleBoundSortItem *i2 = (SingleBoundSortItem *) b;
	TypeCacheEntry *typcache = (TypeCacheEntry *) arg;

	return range_cmp_bounds(typcache, &i1->bound, &i2->bound);
}

/*
 * Compare NonEmptyRanges by lower bound.
 */
static int
interval_cmp_lower(const void *a, const void *b, void *arg)
{
	NonEmptyRange *i1 = (NonEmptyRange *) a;
	NonEmptyRange *i2 = (NonEmptyRange *) b;
	TypeCacheEntry *typcache = (TypeCacheEntry *) arg;

	return range_cmp_bounds(typcache, &i1->lower, &i2->lower);
}

/*
 * Compare NonEmptyRanges by upper bound.
 */
static int
interval_cmp_upper(const void *a, const void *b, void *arg)
{
	NonEmptyRange *i1 = (NonEmptyRange *) a;
	NonEmptyRange *i2 = (NonEmptyRange *) b;
	TypeCacheEntry *typcache = (TypeCacheEntry *) arg;

	return range_cmp_bounds(typcache, &i1->upper, &i2->upper);
}

/*
 * Compare CommonEntrys by their deltas.
 */
static int
common_entry_cmp(const void *i1, const void *i2)
{
	double		delta1 = ((CommonEntry *) i1)->delta;
	double		delta2 = ((CommonEntry *) i2)->delta;

	if (delta1 < delta2)
		return -1;
	else if (delta1 > delta2)
		return 1;
	else
		return 0;
}

/*
 * Convenience function to invoke type-specific subtype_diff function.
 * Caller must have already checked that there is one for the range type.
 */
static float8
call_subtype_diff(TypeCacheEntry *typcache, Datum val1, Datum val2)
{
	float8		value;

	value = DatumGetFloat8(FunctionCall2Coll(&typcache->rng_subdiff_finfo,
											 typcache->rng_collation,
											 val1, val2));
	/* Cope with buggy subtype_diff function by returning zero */
	if (value >= 0.0)
		return value;
	return 0.0;
}