access/brin/brin_inclusion.c

/*
 * brin_inclusion.c
 *		Implementation of inclusion opclasses for BRIN
 *
 * This module provides framework BRIN support functions for the "inclusion"
 * operator classes.  A few SQL-level support functions are also required for
 * each opclass.
 *
 * The "inclusion" BRIN strategy is useful for types that support R-Tree
 * operations.  This implementation is a straight mapping of those operations
 * to the block-range nature of BRIN, with two exceptions: (a) we explicitly
 * support "empty" elements: at least with range types, we need to consider
 * emptiness separately from regular R-Tree strategies; and (b) we need to
 * consider "unmergeable" elements, that is, a set of elements for whose union
 * no representation exists.  The only case where that happens as of this
 * writing is the INET type, where IPv6 values cannot be merged with IPv4
 * values.
 *
 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/brin/brin_inclusion.c
 */
#include "postgres.h"

#include "access/brin_internal.h"
#include "access/brin_tuple.h"
#include "access/genam.h"
#include "access/skey.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_type.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"


/*
 * Additional SQL level support functions
 *
 * Procedure numbers must not use values reserved for BRIN itself; see
 * brin_internal.h.
 */
#define		INCLUSION_MAX_PROCNUMS	4	/* maximum support procs we need */
#define		PROCNUM_MERGE			11	/* required */
#define		PROCNUM_MERGEABLE		12	/* optional */
#define		PROCNUM_CONTAINS		13	/* optional */
#define		PROCNUM_EMPTY			14	/* optional */


/*
 * Subtract this from procnum to obtain index in InclusionOpaque arrays
 * (Must be equal to minimum of private procnums).
 */
#define		PROCNUM_BASE			11

/*-
 * The values stored in the bv_values arrays correspond to:
 *
 * INCLUSION_UNION
 *		the union of the values in the block range
 * INCLUSION_UNMERGEABLE
 *		whether the values in the block range cannot be merged
 *		(e.g. an IPv6 address amidst IPv4 addresses)
 * INCLUSION_CONTAINS_EMPTY
 *		whether an empty value is present in any tuple
 *		in the block range
 */
#define INCLUSION_UNION				0
#define INCLUSION_UNMERGEABLE		1
#define INCLUSION_CONTAINS_EMPTY	2


typedef struct InclusionOpaque
{
	FmgrInfo	extra_procinfos[INCLUSION_MAX_PROCNUMS];
	bool		extra_proc_missing[INCLUSION_MAX_PROCNUMS];
	Oid			cached_subtype;
	FmgrInfo	strategy_procinfos[RTMaxStrategyNumber];
} InclusionOpaque;

static FmgrInfo *inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno,
					   uint16 procnum);
static FmgrInfo *inclusion_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno,
								Oid subtype, uint16 strategynum);


/*
 * BRIN inclusion OpcInfo function
 */
Datum
brin_inclusion_opcinfo(PG_FUNCTION_ARGS)
{
	Oid			typoid = PG_GETARG_OID(0);
	BrinOpcInfo *result;
	TypeCacheEntry *bool_typcache = lookup_type_cache(BOOLOID, 0);

	/*
	 * All members of opaque are initialized lazily; both procinfo arrays
	 * start out as non-initialized by having fn_oid be InvalidOid, and
	 * "missing" to false, by zeroing here.  strategy_procinfos elements can
	 * be invalidated when cached_subtype changes by zeroing fn_oid.
	 * extra_procinfo entries are never invalidated, but if a lookup fails
	 * (which is expected), extra_proc_missing is set to true, indicating not
	 * to look it up again.
	 */
	result = palloc0(MAXALIGN(SizeofBrinOpcInfo(3)) + sizeof(InclusionOpaque));
	result->oi_nstored = 3;
	result->oi_opaque = (InclusionOpaque *)
		MAXALIGN((char *) result + SizeofBrinOpcInfo(3));

	/* the union */
	result->oi_typcache[INCLUSION_UNION] =
		lookup_type_cache(typoid, 0);

	/* includes elements that are not mergeable */
	result->oi_typcache[INCLUSION_UNMERGEABLE] = bool_typcache;

	/* includes the empty element */
	result->oi_typcache[INCLUSION_CONTAINS_EMPTY] = bool_typcache;

	PG_RETURN_POINTER(result);
}

/*
 * BRIN inclusion add value function
 *
 * Examine the given index tuple (which contains partial status of a certain
 * page range) by comparing it to the given value that comes from another heap
 * tuple.  If the new value is outside the union specified by the existing
 * tuple values, update the index tuple and return true.  Otherwise, return
 * false and do not modify in this case.
 */
Datum
brin_inclusion_add_value(PG_FUNCTION_ARGS)
{
	BrinDesc   *bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1);
	Datum		newval = PG_GETARG_DATUM(2);
	bool		isnull = PG_GETARG_BOOL(3);
	Oid			colloid = PG_GET_COLLATION();
	FmgrInfo   *finfo;
	Datum		result;
	bool		new = false;
	AttrNumber	attno;
	Form_pg_attribute attr;

	/*
	 * If the new value is null, we record that we saw it if it's the first
	 * one; otherwise, there's nothing to do.
	 */
	if (isnull)
	{
		if (column->bv_hasnulls)
			PG_RETURN_BOOL(false);

		column->bv_hasnulls = true;
		PG_RETURN_BOOL(true);
	}

	attno = column->bv_attno;
	attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1);

	/*
	 * If the recorded value is null, copy the new value (which we know to be
	 * not null), and we're almost done.
	 */
	if (column->bv_allnulls)
	{
		column->bv_values[INCLUSION_UNION] =
			datumCopy(newval, attr->attbyval, attr->attlen);
		column->bv_values[INCLUSION_UNMERGEABLE] = BoolGetDatum(false);
		column->bv_values[INCLUSION_CONTAINS_EMPTY] = BoolGetDatum(false);
		column->bv_allnulls = false;
		new = true;
	}

	/*
	 * No need for further processing if the block range is marked as
	 * containing unmergeable values.
	 */
	if (DatumGetBool(column->bv_values[INCLUSION_UNMERGEABLE]))
		PG_RETURN_BOOL(false);

	/*
	 * If the opclass supports the concept of empty values, test the passed
	 * new value for emptiness; if it returns true, we need to set the
	 * "contains empty" flag in the element (unless already set).
	 */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_EMPTY);
	if (finfo != NULL && DatumGetBool(FunctionCall1Coll(finfo, colloid, newval)))
	{
		if (!DatumGetBool(column->bv_values[INCLUSION_CONTAINS_EMPTY]))
		{
			column->bv_values[INCLUSION_CONTAINS_EMPTY] = BoolGetDatum(true);
			PG_RETURN_BOOL(true);
		}

		PG_RETURN_BOOL(false);
	}

	if (new)
		PG_RETURN_BOOL(true);

	/* Check if the new value is already contained. */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_CONTAINS);
	if (finfo != NULL &&
		DatumGetBool(FunctionCall2Coll(finfo, colloid,
									   column->bv_values[INCLUSION_UNION],
									   newval)))
		PG_RETURN_BOOL(false);

	/*
	 * Check if the new value is mergeable to the existing union.  If it is
	 * not, mark the value as containing unmergeable elements and get out.
	 *
	 * Note: at this point we could remove the value from the union, since
	 * it's not going to be used any longer.  However, the BRIN framework
	 * doesn't allow for the value not being present.  Improve someday.
	 */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE);
	if (finfo != NULL &&
		!DatumGetBool(FunctionCall2Coll(finfo, colloid,
										column->bv_values[INCLUSION_UNION],
										newval)))
	{
		column->bv_values[INCLUSION_UNMERGEABLE] = BoolGetDatum(true);
		PG_RETURN_BOOL(true);
	}

	/* Finally, merge the new value to the existing union. */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE);
	Assert(finfo != NULL);
	result = FunctionCall2Coll(finfo, colloid,
							   column->bv_values[INCLUSION_UNION], newval);
	if (!attr->attbyval &&
		DatumGetPointer(result) != DatumGetPointer(column->bv_values[INCLUSION_UNION]))
	{
		pfree(DatumGetPointer(column->bv_values[INCLUSION_UNION]));

		if (result == newval)
			result = datumCopy(result, attr->attbyval, attr->attlen);
	}
	column->bv_values[INCLUSION_UNION] = result;

	PG_RETURN_BOOL(true);
}

/*
 * BRIN inclusion consistent function
 *
 * All of the strategies are optional.
 */
Datum
brin_inclusion_consistent(PG_FUNCTION_ARGS)
{
	BrinDesc   *bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1);
	ScanKey		key = (ScanKey) PG_GETARG_POINTER(2);
	Oid			colloid = PG_GET_COLLATION(),
				subtype;
	Datum		unionval;
	AttrNumber	attno;
	Datum		query;
	FmgrInfo   *finfo;
	Datum		result;

	Assert(key->sk_attno == column->bv_attno);

	/* Handle IS NULL/IS NOT NULL tests. */
	if (key->sk_flags & SK_ISNULL)
	{
		if (key->sk_flags & SK_SEARCHNULL)
		{
			if (column->bv_allnulls || column->bv_hasnulls)
				PG_RETURN_BOOL(true);
			PG_RETURN_BOOL(false);
		}

		/*
		 * For IS NOT NULL, we can only skip ranges that are known to have
		 * only nulls.
		 */
		if (key->sk_flags & SK_SEARCHNOTNULL)
			PG_RETURN_BOOL(!column->bv_allnulls);

		/*
		 * Neither IS NULL nor IS NOT NULL was used; assume all indexable
		 * operators are strict and return false.
		 */
		PG_RETURN_BOOL(false);
	}

	/* If it is all nulls, it cannot possibly be consistent. */
	if (column->bv_allnulls)
		PG_RETURN_BOOL(false);

	/* It has to be checked, if it contains elements that are not mergeable. */
	if (DatumGetBool(column->bv_values[INCLUSION_UNMERGEABLE]))
		PG_RETURN_BOOL(true);

	attno = key->sk_attno;
	subtype = key->sk_subtype;
	query = key->sk_argument;
	unionval = column->bv_values[INCLUSION_UNION];
	switch (key->sk_strategy)
	{
			/*
			 * Placement strategies
			 *
			 * These are implemented by logically negating the result of the
			 * converse placement operator; for this to work, the converse
			 * operator must be part of the opclass.  An error will be thrown
			 * by inclusion_get_strategy_procinfo() if the required strategy
			 * is not part of the opclass.
			 *
			 * These all return false if either argument is empty, so there is
			 * no need to check for empty elements.
			 */

		case RTLeftStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverRightStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTOverLeftStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTRightStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTOverRightStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTLeftStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTRightStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverLeftStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTBelowStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverAboveStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTOverBelowStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTAboveStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTOverAboveStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTBelowStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		case RTAboveStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverBelowStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

			/*
			 * Overlap and contains strategies
			 *
			 * These strategies are simple enough that we can simply call the
			 * operator and return its result.  Empty elements don't change
			 * the result.
			 */

		case RTOverlapStrategyNumber:
		case RTContainsStrategyNumber:
		case RTOldContainsStrategyNumber:
		case RTContainsElemStrategyNumber:
		case RTSubStrategyNumber:
		case RTSubEqualStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													key->sk_strategy);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_DATUM(result);

			/*
			 * Contained by strategies
			 *
			 * We cannot just call the original operator for the contained by
			 * strategies because some elements can be contained even though
			 * the union is not; instead we use the overlap operator.
			 *
			 * We check for empty elements separately as they are not merged
			 * to the union but contained by everything.
			 */

		case RTContainedByStrategyNumber:
		case RTOldContainedByStrategyNumber:
		case RTSuperStrategyNumber:
		case RTSuperEqualStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverlapStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			if (DatumGetBool(result))
				PG_RETURN_BOOL(true);

			PG_RETURN_DATUM(column->bv_values[INCLUSION_CONTAINS_EMPTY]);

			/*
			 * Adjacent strategy
			 *
			 * We test for overlap first but to be safe we need to call the
			 * actual adjacent operator also.
			 *
			 * An empty element cannot be adjacent to any other, so there is
			 * no need to check for it.
			 */

		case RTAdjacentStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTOverlapStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			if (DatumGetBool(result))
				PG_RETURN_BOOL(true);

			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTAdjacentStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_DATUM(result);

			/*
			 * Basic comparison strategies
			 *
			 * It is straightforward to support the equality strategies with
			 * the contains operator.  Generally, inequality strategies do not
			 * make much sense for the types which will be used with the
			 * inclusion BRIN family of opclasses, but is possible to
			 * implement them with logical negation of the left-of and
			 * right-of operators.
			 *
			 * NB: These strategies cannot be used with geometric datatypes
			 * that use comparison of areas!  The only exception is the "same"
			 * strategy.
			 *
			 * Empty elements are considered to be less than the others.  We
			 * cannot use the empty support function to check the query is an
			 * empty element, because the query can be another data type than
			 * the empty support function argument.  So we will return true,
			 * if there is a possibility that empty elements will change the
			 * result.
			 */

		case RTLessStrategyNumber:
		case RTLessEqualStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTRightStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			if (!DatumGetBool(result))
				PG_RETURN_BOOL(true);

			PG_RETURN_DATUM(column->bv_values[INCLUSION_CONTAINS_EMPTY]);

		case RTSameStrategyNumber:
		case RTEqualStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTContainsStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			if (DatumGetBool(result))
				PG_RETURN_BOOL(true);

			PG_RETURN_DATUM(column->bv_values[INCLUSION_CONTAINS_EMPTY]);

		case RTGreaterEqualStrategyNumber:
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTLeftStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			if (!DatumGetBool(result))
				PG_RETURN_BOOL(true);

			PG_RETURN_DATUM(column->bv_values[INCLUSION_CONTAINS_EMPTY]);

		case RTGreaterStrategyNumber:
			/* no need to check for empty elements */
			finfo = inclusion_get_strategy_procinfo(bdesc, attno, subtype,
													RTLeftStrategyNumber);
			result = FunctionCall2Coll(finfo, colloid, unionval, query);
			PG_RETURN_BOOL(!DatumGetBool(result));

		default:
			/* shouldn't happen */
			elog(ERROR, "invalid strategy number %d", key->sk_strategy);
			PG_RETURN_BOOL(false);
	}
}

/*
 * BRIN inclusion union function
 *
 * Given two BrinValues, update the first of them as a union of the summary
 * values contained in both.  The second one is untouched.
 */
Datum
brin_inclusion_union(PG_FUNCTION_ARGS)
{
	BrinDesc   *bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *col_a = (BrinValues *) PG_GETARG_POINTER(1);
	BrinValues *col_b = (BrinValues *) PG_GETARG_POINTER(2);
	Oid			colloid = PG_GET_COLLATION();
	AttrNumber	attno;
	Form_pg_attribute attr;
	FmgrInfo   *finfo;
	Datum		result;

	Assert(col_a->bv_attno == col_b->bv_attno);

	/* Adjust "hasnulls". */
	if (!col_a->bv_hasnulls && col_b->bv_hasnulls)
		col_a->bv_hasnulls = true;

	/* If there are no values in B, there's nothing left to do. */
	if (col_b->bv_allnulls)
		PG_RETURN_VOID();

	attno = col_a->bv_attno;
	attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1);

	/*
	 * Adjust "allnulls".  If A doesn't have values, just copy the values from
	 * B into A, and we're done.  We cannot run the operators in this case,
	 * because values in A might contain garbage.  Note we already established
	 * that B contains values.
	 */
	if (col_a->bv_allnulls)
	{
		col_a->bv_allnulls = false;
		col_a->bv_values[INCLUSION_UNION] =
			datumCopy(col_b->bv_values[INCLUSION_UNION],
					  attr->attbyval, attr->attlen);
		col_a->bv_values[INCLUSION_UNMERGEABLE] =
			col_b->bv_values[INCLUSION_UNMERGEABLE];
		col_a->bv_values[INCLUSION_CONTAINS_EMPTY] =
			col_b->bv_values[INCLUSION_CONTAINS_EMPTY];
		PG_RETURN_VOID();
	}

	/* If B includes empty elements, mark A similarly, if needed. */
	if (!DatumGetBool(col_a->bv_values[INCLUSION_CONTAINS_EMPTY]) &&
		DatumGetBool(col_b->bv_values[INCLUSION_CONTAINS_EMPTY]))
		col_a->bv_values[INCLUSION_CONTAINS_EMPTY] = BoolGetDatum(true);

	/* Check if A includes elements that are not mergeable. */
	if (DatumGetBool(col_a->bv_values[INCLUSION_UNMERGEABLE]))
		PG_RETURN_VOID();

	/* If B includes elements that are not mergeable, mark A similarly. */
	if (DatumGetBool(col_b->bv_values[INCLUSION_UNMERGEABLE]))
	{
		col_a->bv_values[INCLUSION_UNMERGEABLE] = BoolGetDatum(true);
		PG_RETURN_VOID();
	}

	/* Check if A and B are mergeable; if not, mark A unmergeable. */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGEABLE);
	if (finfo != NULL &&
		!DatumGetBool(FunctionCall2Coll(finfo, colloid,
										col_a->bv_values[INCLUSION_UNION],
										col_b->bv_values[INCLUSION_UNION])))
	{
		col_a->bv_values[INCLUSION_UNMERGEABLE] = BoolGetDatum(true);
		PG_RETURN_VOID();
	}

	/* Finally, merge B to A. */
	finfo = inclusion_get_procinfo(bdesc, attno, PROCNUM_MERGE);
	Assert(finfo != NULL);
	result = FunctionCall2Coll(finfo, colloid,
							   col_a->bv_values[INCLUSION_UNION],
							   col_b->bv_values[INCLUSION_UNION]);
	if (!attr->attbyval &&
		DatumGetPointer(result) != DatumGetPointer(col_a->bv_values[INCLUSION_UNION]))
	{
		pfree(DatumGetPointer(col_a->bv_values[INCLUSION_UNION]));

		if (result == col_b->bv_values[INCLUSION_UNION])
			result = datumCopy(result, attr->attbyval, attr->attlen);
	}
	col_a->bv_values[INCLUSION_UNION] = result;

	PG_RETURN_VOID();
}

/*
 * Cache and return inclusion opclass support procedure
 *
 * Return the procedure corresponding to the given function support number
 * or null if it is not exists.
 */
static FmgrInfo *
inclusion_get_procinfo(BrinDesc *bdesc, uint16 attno, uint16 procnum)
{
	InclusionOpaque *opaque;
	uint16		basenum = procnum - PROCNUM_BASE;

	/*
	 * We cache these in the opaque struct, to avoid repetitive syscache
	 * lookups.
	 */
	opaque = (InclusionOpaque *) bdesc->bd_info[attno - 1]->oi_opaque;

	/*
	 * If we already searched for this proc and didn't find it, don't bother
	 * searching again.
	 */
	if (opaque->extra_proc_missing[basenum])
		return NULL;

	if (opaque->extra_procinfos[basenum].fn_oid == InvalidOid)
	{
		if (RegProcedureIsValid(index_getprocid(bdesc->bd_index, attno,
												procnum)))
		{
			fmgr_info_copy(&opaque->extra_procinfos[basenum],
						   index_getprocinfo(bdesc->bd_index, attno, procnum),
						   bdesc->bd_context);
		}
		else
		{
			opaque->extra_proc_missing[basenum] = true;
			return NULL;
		}
	}

	return &opaque->extra_procinfos[basenum];
}

/*
 * Cache and return the procedure of the given strategy
 *
 * Return the procedure corresponding to the given sub-type and strategy
 * number.  The data type of the index will be used as the left hand side of
 * the operator and the given sub-type will be used as the right hand side.
 * Throws an error if the pg_amop row does not exist, but that should not
 * happen with a properly configured opclass.
 *
 * It always throws an error when the data type of the opclass is different
 * from the data type of the column or the expression.  That happens when the
 * column data type has implicit cast to the opclass data type.  We don't
 * bother casting types, because this situation can easily be avoided by
 * setting storage data type to that of the opclass.  The same problem does not
 * apply to the data type of the right hand side, because the type in the
 * ScanKey always matches the opclass' one.
 *
 * Note: this function mirrors minmax_get_strategy_procinfo; if changes are
 * made here, see that function too.
 */
static FmgrInfo *
inclusion_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno, Oid subtype,
								uint16 strategynum)
{
	InclusionOpaque *opaque;

	Assert(strategynum >= 1 &&
		   strategynum <= RTMaxStrategyNumber);

	opaque = (InclusionOpaque *) bdesc->bd_info[attno - 1]->oi_opaque;

	/*
	 * We cache the procedures for the last sub-type in the opaque struct, to
	 * avoid repetitive syscache lookups.  If the sub-type is changed,
	 * invalidate all the cached entries.
	 */
	if (opaque->cached_subtype != subtype)
	{
		uint16		i;

		for (i = 1; i <= RTMaxStrategyNumber; i++)
			opaque->strategy_procinfos[i - 1].fn_oid = InvalidOid;
		opaque->cached_subtype = subtype;
	}

	if (opaque->strategy_procinfos[strategynum - 1].fn_oid == InvalidOid)
	{
		Form_pg_attribute attr;
		HeapTuple	tuple;
		Oid			opfamily,
					oprid;
		bool		isNull;

		opfamily = bdesc->bd_index->rd_opfamily[attno - 1];
		attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1);
		tuple = SearchSysCache4(AMOPSTRATEGY, ObjectIdGetDatum(opfamily),
								ObjectIdGetDatum(attr->atttypid),
								ObjectIdGetDatum(subtype),
								Int16GetDatum(strategynum));

		if (!HeapTupleIsValid(tuple))
			elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
				 strategynum, attr->atttypid, subtype, opfamily);

		oprid = DatumGetObjectId(SysCacheGetAttr(AMOPSTRATEGY, tuple,
												 Anum_pg_amop_amopopr, &isNull));
		ReleaseSysCache(tuple);
		Assert(!isNull && RegProcedureIsValid(oprid));

		fmgr_info_cxt(get_opcode(oprid),
					  &opaque->strategy_procinfos[strategynum - 1],
					  bdesc->bd_context);
	}

	return &opaque->strategy_procinfos[strategynum - 1];
}