Add temporal PRIMARY KEY and UNIQUE constraints

Add WITHOUT OVERLAPS clause to PRIMARY KEY and UNIQUE constraints.
These are backed by GiST indexes instead of B-tree indexes, since they
are essentially exclusion constraints with = for the scalar parts of
the key and && for the temporal part.

Author: Paul A. Jungwirth <pj@illuminatedcomputing.com>
Reviewed-by: Peter Eisentraut <peter@eisentraut.org>
Reviewed-by: jian he <jian.universality@gmail.com>
Discussion: https://www.postgresql.org/message-id/flat/CA+renyUApHgSZF9-nd-a0+OPGharLQLO=mDHcY4_qQ0+noCUVg@mail.gmail.com

1 files changed, 129 insertions, 14 deletions

diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 340248a3f29..7a87626f5f0 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -16,6 +16,7 @@
 #include "postgres.h"
 
 #include "access/amapi.h"
+#include "access/gist.h"
 #include "access/heapam.h"
 #include "access/htup_details.h"
 #include "access/reloptions.h"
@@ -86,6 +87,7 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo,
 							  Oid accessMethodId,
 							  bool amcanorder,
 							  bool isconstraint,
+							  bool iswithoutoverlaps,
 							  Oid ddl_userid,
 							  int ddl_sec_context,
 							  int *ddl_save_nestlevel);
@@ -144,6 +146,7 @@ typedef struct ReindexErrorInfo
  *		to index on.
  * 'exclusionOpNames': list of names of exclusion-constraint operators,
  *		or NIL if not an exclusion constraint.
+ * 'isWithoutOverlaps': true iff this index has a WITHOUT OVERLAPS clause.
  *
  * This is tailored to the needs of ALTER TABLE ALTER TYPE, which recreates
  * any indexes that depended on a changing column from their pg_get_indexdef
@@ -173,7 +176,8 @@ bool
 CheckIndexCompatible(Oid oldId,
 					 const char *accessMethodName,
 					 const List *attributeList,
-					 const List *exclusionOpNames)
+					 const List *exclusionOpNames,
+					 bool isWithoutOverlaps)
 {
 	bool		isconstraint;
 	Oid		   *typeIds;
@@ -248,8 +252,8 @@ CheckIndexCompatible(Oid oldId,
 					  coloptions, attributeList,
 					  exclusionOpNames, relationId,
 					  accessMethodName, accessMethodId,
-					  amcanorder, isconstraint, InvalidOid, 0, NULL);
-
+					  amcanorder, isconstraint, isWithoutOverlaps, InvalidOid,
+					  0, NULL);
 
 	/* Get the soon-obsolete pg_index tuple. */
 	tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(oldId));
@@ -559,6 +563,7 @@ DefineIndex(Oid tableId,
 	bool		amcanorder;
 	bool		amissummarizing;
 	amoptions_function amoptions;
+	bool		exclusion;
 	bool		partitioned;
 	bool		safe_index;
 	Datum		reloptions;
@@ -677,6 +682,12 @@ DefineIndex(Oid tableId,
 
 	namespaceId = RelationGetNamespace(rel);
 
+	/*
+	 * It has exclusion constraint behavior if it's an EXCLUDE constraint or a
+	 * temporal PRIMARY KEY/UNIQUE constraint
+	 */
+	exclusion = stmt->excludeOpNames || stmt->iswithoutoverlaps;
+
 	/* Ensure that it makes sense to index this kind of relation */
 	switch (rel->rd_rel->relkind)
 	{
@@ -845,7 +856,7 @@ DefineIndex(Oid tableId,
 	pgstat_progress_update_param(PROGRESS_CREATEIDX_ACCESS_METHOD_OID,
 								 accessMethodId);
 
-	if (stmt->unique && !amRoutine->amcanunique)
+	if (stmt->unique && !stmt->iswithoutoverlaps && !amRoutine->amcanunique)
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("access method \"%s\" does not support unique indexes",
@@ -860,7 +871,7 @@ DefineIndex(Oid tableId,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("access method \"%s\" does not support multicolumn indexes",
 						accessMethodName)));
-	if (stmt->excludeOpNames && amRoutine->amgettuple == NULL)
+	if (exclusion && amRoutine->amgettuple == NULL)
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("access method \"%s\" does not support exclusion constraints",
@@ -913,8 +924,9 @@ DefineIndex(Oid tableId,
 					  coloptions, allIndexParams,
 					  stmt->excludeOpNames, tableId,
 					  accessMethodName, accessMethodId,
-					  amcanorder, stmt->isconstraint, root_save_userid,
-					  root_save_sec_context, &root_save_nestlevel);
+					  amcanorder, stmt->isconstraint, stmt->iswithoutoverlaps,
+					  root_save_userid, root_save_sec_context,
+					  &root_save_nestlevel);
 
 	/*
 	 * Extra checks when creating a PRIMARY KEY index.
@@ -932,7 +944,7 @@ DefineIndex(Oid tableId,
 	 * We could lift this limitation if we had global indexes, but those have
 	 * their own problems, so this is a useful feature combination.
 	 */
-	if (partitioned && (stmt->unique || stmt->excludeOpNames))
+	if (partitioned && (stmt->unique || exclusion))
 	{
 		PartitionKey key = RelationGetPartitionKey(rel);
 		const char *constraint_type;
@@ -986,10 +998,10 @@ DefineIndex(Oid tableId,
 			 * associated with index columns, too.  We know what to do with
 			 * btree opclasses; if there are ever any other index types that
 			 * support unique indexes, this logic will need extension. But if
-			 * we have an exclusion constraint, it already knows the
-			 * operators, so we don't have to infer them.
+			 * we have an exclusion constraint (or a temporal PK), it already
+			 * knows the operators, so we don't have to infer them.
 			 */
-			if (stmt->unique && accessMethodId != BTREE_AM_OID)
+			if (stmt->unique && !stmt->iswithoutoverlaps && accessMethodId != BTREE_AM_OID)
 				ereport(ERROR,
 						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 						 errmsg("cannot match partition key to an index using access method \"%s\"",
@@ -1028,12 +1040,12 @@ DefineIndex(Oid tableId,
 					{
 						Oid			idx_eqop = InvalidOid;
 
-						if (stmt->unique)
+						if (stmt->unique && !stmt->iswithoutoverlaps)
 							idx_eqop = get_opfamily_member(idx_opfamily,
 														   idx_opcintype,
 														   idx_opcintype,
 														   BTEqualStrategyNumber);
-						else if (stmt->excludeOpNames)
+						else if (exclusion)
 							idx_eqop = indexInfo->ii_ExclusionOps[j];
 						Assert(idx_eqop);
 
@@ -1042,7 +1054,7 @@ DefineIndex(Oid tableId,
 							found = true;
 							break;
 						}
-						else if (stmt->excludeOpNames)
+						else if (exclusion)
 						{
 							/*
 							 * We found a match, but it's not an equality
@@ -1186,6 +1198,8 @@ DefineIndex(Oid tableId,
 		constr_flags |= INDEX_CONSTR_CREATE_DEFERRABLE;
 	if (stmt->initdeferred)
 		constr_flags |= INDEX_CONSTR_CREATE_INIT_DEFERRED;
+	if (stmt->iswithoutoverlaps)
+		constr_flags |= INDEX_CONSTR_CREATE_WITHOUT_OVERLAPS;
 
 	indexRelationId =
 		index_create(rel, indexRelationName, indexRelationId, parentIndexId,
@@ -1850,6 +1864,7 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 				  Oid accessMethodId,
 				  bool amcanorder,
 				  bool isconstraint,
+				  bool iswithoutoverlaps,
 				  Oid ddl_userid,
 				  int ddl_sec_context,
 				  int *ddl_save_nestlevel)
@@ -1873,6 +1888,14 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 	else
 		nextExclOp = NULL;
 
+	/* exclusionOpNames can be non-NIL if we are creating a partition */
+	if (iswithoutoverlaps && exclusionOpNames == NIL)
+	{
+		indexInfo->ii_ExclusionOps = palloc_array(Oid, nkeycols);
+		indexInfo->ii_ExclusionProcs = palloc_array(Oid, nkeycols);
+		indexInfo->ii_ExclusionStrats = palloc_array(uint16, nkeycols);
+	}
+
 	if (OidIsValid(ddl_userid))
 		GetUserIdAndSecContext(&save_userid, &save_sec_context);
 
@@ -2149,6 +2172,21 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 			indexInfo->ii_ExclusionStrats[attn] = strat;
 			nextExclOp = lnext(exclusionOpNames, nextExclOp);
 		}
+		else if (iswithoutoverlaps)
+		{
+			StrategyNumber strat;
+			Oid			opid;
+
+			if (attn == nkeycols - 1)
+				strat = RTOverlapStrategyNumber;
+			else
+				strat = RTEqualStrategyNumber;
+			GetOperatorFromWellKnownStrategy(opclassOids[attn], atttype,
+											 &opid, &strat);
+			indexInfo->ii_ExclusionOps[attn] = opid;
+			indexInfo->ii_ExclusionProcs[attn] = get_opcode(opid);
+			indexInfo->ii_ExclusionStrats[attn] = strat;
+		}
 
 		/*
 		 * Set up the per-column options (indoption field).  For now, this is
@@ -2380,6 +2418,83 @@ GetDefaultOpClass(Oid type_id, Oid am_id)
 }
 
 /*
+ * GetOperatorFromWellKnownStrategy
+ *
+ * opclass - the opclass to use
+ * atttype - the type to ask about
+ * opid - holds the operator we found
+ * strat - holds the input and output strategy number
+ *
+ * Finds an operator from a "well-known" strategy number.  This is used for
+ * temporal index constraints (and other temporal features) to look up
+ * equality and overlaps operators, since the strategy numbers for non-btree
+ * indexams need not follow any fixed scheme.  We ask an opclass support
+ * function to translate from the well-known number to the internal value.  If
+ * the function isn't defined or it gives no result, we return
+ * InvalidStrategy.
+ */
+void
+GetOperatorFromWellKnownStrategy(Oid opclass, Oid atttype,
+								 Oid *opid, StrategyNumber *strat)
+{
+	Oid			opfamily;
+	Oid			opcintype;
+	StrategyNumber instrat = *strat;
+
+	Assert(instrat == RTEqualStrategyNumber || instrat == RTOverlapStrategyNumber);
+
+	*opid = InvalidOid;
+
+	if (get_opclass_opfamily_and_input_type(opclass, &opfamily, &opcintype))
+	{
+		/*
+		 * Ask the opclass to translate to its internal stratnum
+		 *
+		 * For now we only need GiST support, but this could support other
+		 * indexams if we wanted.
+		 */
+		*strat = GistTranslateStratnum(opclass, instrat);
+		if (*strat == InvalidStrategy)
+		{
+			HeapTuple	tuple;
+
+			tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclass));
+			if (!HeapTupleIsValid(tuple))
+				elog(ERROR, "cache lookup failed for operator class %u", opclass);
+
+			ereport(ERROR,
+					errcode(ERRCODE_UNDEFINED_OBJECT),
+					instrat == RTEqualStrategyNumber ?
+					errmsg("could not identify an equality operator for type %s", format_type_be(atttype)) :
+					errmsg("could not identify an overlaps operator for type %s", format_type_be(atttype)),
+					errdetail("Could not translate strategy number %d for operator class \"%s\" for access method \"%s\".",
+							  instrat, NameStr(((Form_pg_opclass) GETSTRUCT(tuple))->opcname), "gist"));
+
+			ReleaseSysCache(tuple);
+		}
+
+		*opid = get_opfamily_member(opfamily, opcintype, opcintype, *strat);
+	}
+
+	if (!OidIsValid(*opid))
+	{
+		HeapTuple	tuple;
+
+		tuple = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfamily));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for operator family %u", opfamily);
+
+		ereport(ERROR,
+				errcode(ERRCODE_UNDEFINED_OBJECT),
+				instrat == RTEqualStrategyNumber ?
+				errmsg("could not identify an equality operator for type %s", format_type_be(atttype)) :
+				errmsg("could not identify an overlaps operator for type %s", format_type_be(atttype)),
+				errdetail("There is no suitable operator in operator family \"%s\" for access method \"%s\".",
+						  NameStr(((Form_pg_opfamily) GETSTRUCT(tuple))->opfname), "gist"));
+	}
+}
+
+/*
  *	makeObjectName()
  *
  *	Create a name for an implicitly created index, sequence, constraint,