Add BRIN infrastructure for "inclusion" opclasses

This lets BRIN be used with R-Tree-like indexing strategies.

Also provided are operator classes for range types, box and inet/cidr.
The infrastructure provided here should be sufficient to create operator
classes for similar datatypes; for instance, opclasses for PostGIS
geometries should be doable, though we didn't try to implement one.

(A box/point opclass was also submitted, but we ripped it out before
commit because the handling of floating point comparisons in existing
code is inconsistent and would generate corrupt indexes.)

Author: Emre Hasegeli.  Cosmetic changes by me
Review: Andreas Karlsson

4 files changed, 721 insertions, 74 deletions

diff --git a/src/backend/access/brin/Makefile b/src/backend/access/brin/Makefile
index ac44fcdee39..f4572d80a89 100644
--- a/src/backend/access/brin/Makefile
+++ b/src/backend/access/brin/Makefile
@@ -13,6 +13,6 @@ top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = brin.o brin_pageops.o brin_revmap.o brin_tuple.o brin_xlog.o \
-	   brin_minmax.o
+       brin_minmax.o brin_inclusion.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index 2b5fb8dce9b..199512551e5 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -105,11 +105,6 @@ brininsert(PG_FUNCTION_ARGS)
 		BrinMemTuple *dtup;
 		BlockNumber heapBlk;
 		int			keyno;
-#ifdef USE_ASSERT_CHECKING
-		BrinTuple  *tmptup;
-		BrinMemTuple *tmpdtup;
-		Size 		tmpsiz;
-#endif
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -137,45 +132,6 @@ brininsert(PG_FUNCTION_ARGS)
 
 		dtup = brin_deform_tuple(bdesc, brtup);
 
-#ifdef USE_ASSERT_CHECKING
-		{
-			/*
-			 * When assertions are enabled, we use this as an opportunity to
-			 * test the "union" method, which would otherwise be used very
-			 * rarely: first create a placeholder tuple, and addValue the
-			 * value we just got into it.  Then union the existing index tuple
-			 * with the updated placeholder tuple.  The tuple resulting from
-			 * that union should be identical to the one resulting from the
-			 * regular operation (straight addValue) below.
-			 *
-			 * Here we create the tuple to compare with; the actual comparison
-			 * is below.
-			 */
-			tmptup = brin_form_placeholder_tuple(bdesc, heapBlk, &tmpsiz);
-			tmpdtup = brin_deform_tuple(bdesc, tmptup);
-			for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
-			{
-				BrinValues *bval;
-				FmgrInfo   *addValue;
-
-				bval = &tmpdtup->bt_columns[keyno];
-				addValue = index_getprocinfo(idxRel, keyno + 1,
-											 BRIN_PROCNUM_ADDVALUE);
-				FunctionCall4Coll(addValue,
-								  idxRel->rd_indcollation[keyno],
-								  PointerGetDatum(bdesc),
-								  PointerGetDatum(bval),
-								  values[keyno],
-								  nulls[keyno]);
-			}
-
-			union_tuples(bdesc, tmpdtup, brtup);
-
-			tmpdtup->bt_placeholder = dtup->bt_placeholder;
-			tmptup = brin_form_tuple(bdesc, heapBlk, tmpdtup, &tmpsiz);
-		}
-#endif
-
 		/*
 		 * Compare the key values of the new tuple to the stored index values;
 		 * our deformed tuple will get updated if the new tuple doesn't fit
@@ -202,20 +158,6 @@ brininsert(PG_FUNCTION_ARGS)
 			need_insert |= DatumGetBool(result);
 		}
 
-#ifdef USE_ASSERT_CHECKING
-		{
-			/*
-			 * Now we can compare the tuple produced by the union function
-			 * with the one from plain addValue.
-			 */
-			BrinTuple  *cmptup;
-			Size		cmpsz;
-
-			cmptup = brin_form_tuple(bdesc, heapBlk, dtup, &cmpsz);
-			Assert(brin_tuples_equal(tmptup, tmpsiz, cmptup, cmpsz));
-		}
-#endif
-
 		if (!need_insert)
 		{
 			/*
@@ -323,8 +265,6 @@ brinbeginscan(PG_FUNCTION_ARGS)
  * If a TID from the revmap is read as InvalidTID, we know that range is
  * unsummarized.  Pages in those ranges need to be returned regardless of scan
  * keys.
- *
- * XXX see _bt_first on what to do about sk_subtype.
  */
 Datum
 bringetbitmap(PG_FUNCTION_ARGS)
@@ -340,7 +280,6 @@ bringetbitmap(PG_FUNCTION_ARGS)
 	BlockNumber nblocks;
 	BlockNumber heapBlk;
 	int			totalpages = 0;
-	int			keyno;
 	FmgrInfo   *consistentFn;
 	MemoryContext oldcxt;
 	MemoryContext perRangeCxt;
@@ -359,18 +298,11 @@ bringetbitmap(PG_FUNCTION_ARGS)
 	heap_close(heapRel, AccessShareLock);
 
 	/*
-	 * Obtain consistent functions for all indexed column.  Maybe it'd be
-	 * possible to do this lazily only the first time we see a scan key that
-	 * involves each particular attribute.
+	 * Make room for the consistent support procedures of indexed columns.  We
+	 * don't look them up here; we do that lazily the first time we see a scan
+	 * key reference each of them.  We rely on zeroing fn_oid to InvalidOid.
 	 */
-	consistentFn = palloc(sizeof(FmgrInfo) * bdesc->bd_tupdesc->natts);
-	for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
-	{
-		FmgrInfo   *tmp;
-
-		tmp = index_getprocinfo(idxRel, keyno + 1, BRIN_PROCNUM_CONSISTENT);
-		fmgr_info_copy(&consistentFn[keyno], tmp, CurrentMemoryContext);
-	}
+	consistentFn = palloc0(sizeof(FmgrInfo) * bdesc->bd_tupdesc->natts);
 
 	/*
 	 * Setup and use a per-range memory context, which is reset every time we
@@ -418,7 +350,6 @@ bringetbitmap(PG_FUNCTION_ARGS)
 		else
 		{
 			BrinMemTuple *dtup;
-			int			keyno;
 
 			dtup = brin_deform_tuple(bdesc, tup);
 			if (dtup->bt_placeholder)
@@ -431,6 +362,8 @@ bringetbitmap(PG_FUNCTION_ARGS)
 			}
 			else
 			{
+				int			keyno;
+
 				/*
 				 * Compare scan keys with summary values stored for the range.
 				 * If scan keys are matched, the page range must be added to
@@ -456,6 +389,17 @@ bringetbitmap(PG_FUNCTION_ARGS)
 						   (key->sk_collation ==
 					   bdesc->bd_tupdesc->attrs[keyattno - 1]->attcollation));
 
+					/* First time this column? look up consistent function */
+					if (consistentFn[keyattno - 1].fn_oid == InvalidOid)
+					{
+						FmgrInfo   *tmp;
+
+						tmp = index_getprocinfo(idxRel, keyattno,
+												BRIN_PROCNUM_CONSISTENT);
+						fmgr_info_copy(&consistentFn[keyattno - 1], tmp,
+									   CurrentMemoryContext);
+					}
+
 					/*
 					 * Check whether the scan key is consistent with the page
 					 * range values; if so, have the pages in the range added
diff --git a/src/backend/access/brin/brin_inclusion.c b/src/backend/access/brin/brin_inclusion.c
new file mode 100644
index 00000000000..1f0bc7fdb1f
--- /dev/null
+++ b/src/backend/access/brin/brin_inclusion.c
@@ -0,0 +1,696 @@
+/*
+ * 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-2015, 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/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:
+ *
+ * 0 - the union of the values in the block range
+ * 1 - whether an empty value is present in any tuple in the block range
+ * 2 - whether the values in the block range cannot be merged (e.g. an IPv6
+ *     address amidst IPv4 addresses).
+ */
+#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;
+
+Datum		brin_inclusion_opcinfo(PG_FUNCTION_ARGS);
+Datum		brin_inclusion_add_value(PG_FUNCTION_ARGS);
+Datum		brin_inclusion_consistent(PG_FUNCTION_ARGS);
+Datum		brin_inclusion_union(PG_FUNCTION_ARGS);
+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 = bdesc->bd_tupdesc->attrs[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)
+		pfree(DatumGetPointer(column->bv_values[INCLUSION_UNION]));
+	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.
+		 */
+		Assert(key->sk_flags & SK_SEARCHNOTNULL);
+		PG_RETURN_BOOL(!column->bv_allnulls);
+	}
+
+	/* 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 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 = bdesc->bd_tupdesc->attrs[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)
+		pfree(DatumGetPointer(col_a->bv_values[INCLUSION_UNION]));
+	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 = bdesc->bd_tupdesc->attrs[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];
+}
diff --git a/src/backend/access/brin/brin_minmax.c b/src/backend/access/brin/brin_minmax.c
index 1175649a6d6..b105f980eca 100644
--- a/src/backend/access/brin/brin_minmax.c
+++ b/src/backend/access/brin/brin_minmax.c
@@ -28,6 +28,10 @@ typedef struct MinmaxOpaque
 	FmgrInfo	strategy_procinfos[BTMaxStrategyNumber];
 } MinmaxOpaque;
 
+Datum		brin_minmax_opcinfo(PG_FUNCTION_ARGS);
+Datum		brin_minmax_add_value(PG_FUNCTION_ARGS);
+Datum		brin_minmax_consistent(PG_FUNCTION_ARGS);
+Datum		brin_minmax_union(PG_FUNCTION_ARGS);
 static FmgrInfo *minmax_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno,
 					Oid subtype, uint16 strategynum);
 
@@ -302,6 +306,9 @@ brin_minmax_union(PG_FUNCTION_ARGS)
 
 /*
  * Cache and return the procedure for the given strategy.
+ *
+ * Note: this function mirrors inclusion_get_strategy_procinfo; see notes
+ * there.  If changes are made here, see that function too.
  */
 static FmgrInfo *
 minmax_get_strategy_procinfo(BrinDesc *bdesc, uint16 attno, Oid subtype,