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.

(previously committed as 46a0cd4cefb, reverted by 46a0cd4cefb; the new
part is this:)

Because 'empty' && 'empty' is false, the temporal PK/UQ constraint
allowed duplicates, which is confusing to users and breaks internal
expectations.  For instance, when GROUP BY checks functional
dependencies on the PK, it allows selecting other columns from the
table, but in the presence of duplicate keys you could get the value
from any of their rows.  So we need to forbid empties.

This all means that at the moment we can only support ranges and
multiranges for temporal PK/UQs, unlike the original patch (above).
Documentation and tests for this are added.  But this could
conceivably be extended by introducing some more general support for
the notion of "empty" for other types.

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

15 files changed, 382 insertions, 43 deletions

diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 8686735f234..d2d0b36d4ea 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -1069,3 +1069,32 @@ gist_stratnum_identity(PG_FUNCTION_ARGS)
 
 	PG_RETURN_UINT16(strat);
 }
+
+/*
+ * Returns the opclass's private stratnum used for the given strategy.
+ *
+ * Calls the opclass's GIST_STRATNUM_PROC support function, if any,
+ * and returns the result.
+ * Returns InvalidStrategy if the function is not defined.
+ */
+StrategyNumber
+GistTranslateStratnum(Oid opclass, StrategyNumber strat)
+{
+	Oid			opfamily;
+	Oid			opcintype;
+	Oid			funcid;
+	Datum		result;
+
+	/* Look up the opclass family and input datatype. */
+	if (!get_opclass_opfamily_and_input_type(opclass, &opfamily, &opcintype))
+		return InvalidStrategy;
+
+	/* Check whether the function is provided. */
+	funcid = get_opfamily_proc(opfamily, opcintype, opcintype, GIST_STRATNUM_PROC);
+	if (!OidIsValid(funcid))
+		return InvalidStrategy;
+
+	/* Ask the translation function */
+	result = OidFunctionCall1Coll(funcid, InvalidOid, UInt16GetDatum(strat));
+	return DatumGetUInt16(result);
+}
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 01b43cc6a84..78e59384d1c 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2163,6 +2163,7 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr,
 							  is_local, /* conislocal */
 							  inhcount, /* coninhcount */
 							  is_no_inherit,	/* connoinherit */
+							  false,	/* conperiod */
 							  is_internal); /* internally constructed? */
 
 	pfree(ccbin);
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 33759056e37..b2b3ecb5244 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -1394,7 +1394,8 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId,
 							oldInfo->ii_NullsNotDistinct,
 							false,	/* not ready for inserts */
 							true,
-							indexRelation->rd_indam->amsummarizing);
+							indexRelation->rd_indam->amsummarizing,
+							oldInfo->ii_WithoutOverlaps);
 
 	/*
 	 * Extract the list of column names and the column numbers for the new
@@ -1874,6 +1875,7 @@ index_concurrently_set_dead(Oid heapId, Oid indexId)
  *		INDEX_CONSTR_CREATE_UPDATE_INDEX: update the pg_index row
  *		INDEX_CONSTR_CREATE_REMOVE_OLD_DEPS: remove existing dependencies
  *			of index on table's columns
+ *		INDEX_CONSTR_CREATE_WITHOUT_OVERLAPS: constraint uses WITHOUT OVERLAPS
  * allow_system_table_mods: allow table to be a system catalog
  * is_internal: index is constructed due to internal process
  */
@@ -1897,11 +1899,13 @@ index_constraint_create(Relation heapRelation,
 	bool		mark_as_primary;
 	bool		islocal;
 	bool		noinherit;
+	bool		is_without_overlaps;
 	int			inhcount;
 
 	deferrable = (constr_flags & INDEX_CONSTR_CREATE_DEFERRABLE) != 0;
 	initdeferred = (constr_flags & INDEX_CONSTR_CREATE_INIT_DEFERRED) != 0;
 	mark_as_primary = (constr_flags & INDEX_CONSTR_CREATE_MARK_AS_PRIMARY) != 0;
+	is_without_overlaps = (constr_flags & INDEX_CONSTR_CREATE_WITHOUT_OVERLAPS) != 0;
 
 	/* constraint creation support doesn't work while bootstrapping */
 	Assert(!IsBootstrapProcessingMode());
@@ -1978,6 +1982,7 @@ index_constraint_create(Relation heapRelation,
 								   islocal,
 								   inhcount,
 								   noinherit,
+								   is_without_overlaps,
 								   is_internal);
 
 	/*
@@ -2427,7 +2432,8 @@ BuildIndexInfo(Relation index)
 					   indexStruct->indnullsnotdistinct,
 					   indexStruct->indisready,
 					   false,
-					   index->rd_indam->amsummarizing);
+					   index->rd_indam->amsummarizing,
+					   indexStruct->indisexclusion && indexStruct->indisunique);
 
 	/* fill in attribute numbers */
 	for (i = 0; i < numAtts; i++)
@@ -2486,7 +2492,8 @@ BuildDummyIndexInfo(Relation index)
 					   indexStruct->indnullsnotdistinct,
 					   indexStruct->indisready,
 					   false,
-					   index->rd_indam->amsummarizing);
+					   index->rd_indam->amsummarizing,
+					   indexStruct->indisexclusion && indexStruct->indisunique);
 
 	/* fill in attribute numbers */
 	for (i = 0; i < numAtts; i++)
@@ -3224,7 +3231,6 @@ IndexCheckExclusion(Relation heapRelation,
 	indexInfo->ii_PredicateState = NULL;
 }
 
-
 /*
  * validate_index - support code for concurrent index builds
  *
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 3baf9231ed0..9be050ccee8 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -75,6 +75,7 @@ CreateConstraintEntry(const char *constraintName,
 					  bool conIsLocal,
 					  int conInhCount,
 					  bool conNoInherit,
+					  bool conPeriod,
 					  bool is_internal)
 {
 	Relation	conDesc;
@@ -190,6 +191,7 @@ CreateConstraintEntry(const char *constraintName,
 	values[Anum_pg_constraint_conislocal - 1] = BoolGetDatum(conIsLocal);
 	values[Anum_pg_constraint_coninhcount - 1] = Int16GetDatum(conInhCount);
 	values[Anum_pg_constraint_connoinherit - 1] = BoolGetDatum(conNoInherit);
+	values[Anum_pg_constraint_conperiod - 1] = BoolGetDatum(conPeriod);
 
 	if (conkeyArray)
 		values[Anum_pg_constraint_conkey - 1] = PointerGetDatum(conkeyArray);
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index b987e023849..d7b71b81d3b 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"
@@ -88,6 +89,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);
@@ -146,6 +148,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
@@ -175,7 +178,8 @@ bool
 CheckIndexCompatible(Oid oldId,
 					 const char *accessMethodName,
 					 const List *attributeList,
-					 const List *exclusionOpNames)
+					 const List *exclusionOpNames,
+					 bool isWithoutOverlaps)
 {
 	bool		isconstraint;
 	Oid		   *typeIds;
@@ -239,7 +243,7 @@ CheckIndexCompatible(Oid oldId,
 	 */
 	indexInfo = makeIndexInfo(numberOfAttributes, numberOfAttributes,
 							  accessMethodId, NIL, NIL, false, false,
-							  false, false, amsummarizing);
+							  false, false, amsummarizing, isWithoutOverlaps);
 	typeIds = palloc_array(Oid, numberOfAttributes);
 	collationIds = palloc_array(Oid, numberOfAttributes);
 	opclassIds = palloc_array(Oid, numberOfAttributes);
@@ -250,8 +254,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));
@@ -561,6 +565,7 @@ DefineIndex(Oid tableId,
 	bool		amcanorder;
 	bool		amissummarizing;
 	amoptions_function amoptions;
+	bool		exclusion;
 	bool		partitioned;
 	bool		safe_index;
 	Datum		reloptions;
@@ -681,6 +686,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)
 	{
@@ -849,7 +860,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",
@@ -864,11 +875,16 @@ 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",
 						accessMethodName)));
+	if (stmt->iswithoutoverlaps && strcmp(accessMethodName, "gist") != 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("access method \"%s\" does not support WITHOUT OVERLAPS constraints",
+						accessMethodName)));
 
 	amcanorder = amRoutine->amcanorder;
 	amoptions = amRoutine->amoptions;
@@ -905,7 +921,8 @@ DefineIndex(Oid tableId,
 							  stmt->nulls_not_distinct,
 							  !concurrent,
 							  concurrent,
-							  amissummarizing);
+							  amissummarizing,
+							  stmt->iswithoutoverlaps);
 
 	typeIds = palloc_array(Oid, numberOfAttributes);
 	collationIds = palloc_array(Oid, numberOfAttributes);
@@ -917,8 +934,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.
@@ -936,7 +954,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;
@@ -990,10 +1008,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\"",
@@ -1032,12 +1050,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);
 
@@ -1046,7 +1064,7 @@ DefineIndex(Oid tableId,
 							found = true;
 							break;
 						}
-						else if (stmt->excludeOpNames)
+						else if (exclusion)
 						{
 							/*
 							 * We found a match, but it's not an equality
@@ -1190,6 +1208,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,
@@ -1856,6 +1876,7 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 				  Oid accessMethodId,
 				  bool amcanorder,
 				  bool isconstraint,
+				  bool iswithoutoverlaps,
 				  Oid ddl_userid,
 				  int ddl_sec_context,
 				  int *ddl_save_nestlevel)
@@ -1879,6 +1900,23 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 	else
 		nextExclOp = NULL;
 
+	/*
+	 * If this is a WITHOUT OVERLAPS constraint, we need space for exclusion
+	 * ops, but we don't need to parse anything, so we can let nextExclOp be
+	 * NULL. Note that for partitions/inheriting/LIKE, exclusionOpNames will
+	 * be set, so we already allocated above.
+	 */
+	if (iswithoutoverlaps)
+	{
+		if (exclusionOpNames == NIL)
+		{
+			indexInfo->ii_ExclusionOps = palloc_array(Oid, nkeycols);
+			indexInfo->ii_ExclusionProcs = palloc_array(Oid, nkeycols);
+			indexInfo->ii_ExclusionStrats = palloc_array(uint16, nkeycols);
+		}
+		nextExclOp = NULL;
+	}
+
 	if (OidIsValid(ddl_userid))
 		GetUserIdAndSecContext(&save_userid, &save_sec_context);
 
@@ -2158,6 +2196,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
@@ -2389,6 +2442,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,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index b3cc6f8f690..818ed5702cf 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -10100,6 +10100,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 									  conislocal,	/* islocal */
 									  coninhcount,	/* inhcount */
 									  connoinherit, /* conNoInherit */
+									  false,	/* conPeriod */
 									  false);	/* is_internal */
 
 	ObjectAddressSet(address, ConstraintRelationId, constrOid);
@@ -10398,6 +10399,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
 									  false,
 									  1,
 									  false,
+									  false,	/* conPeriod */
 									  false);
 
 			/*
@@ -10920,6 +10922,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 								  false,	/* islocal */
 								  1,	/* inhcount */
 								  false,	/* conNoInherit */
+								  false,	/* conPeriod */
 								  true);
 
 		/* Set up partition dependencies for the new constraint */
@@ -14100,7 +14103,8 @@ TryReuseIndex(Oid oldId, IndexStmt *stmt)
 	if (CheckIndexCompatible(oldId,
 							 stmt->accessMethod,
 							 stmt->indexParams,
-							 stmt->excludeOpNames))
+							 stmt->excludeOpNames,
+							 stmt->iswithoutoverlaps))
 	{
 		Relation	irel = index_open(oldId, NoLock);
 
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index 170360edda8..29d30bfb6f7 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -834,6 +834,7 @@ CreateTriggerFiringOn(CreateTrigStmt *stmt, const char *queryString,
 											  true, /* islocal */
 											  0,	/* inhcount */
 											  true, /* noinherit */
+											  false,	/* conperiod */
 											  isInternal);	/* is_internal */
 	}
 
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 2a1e7133356..2a6550de907 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -3621,6 +3621,7 @@ domainAddCheckConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
 							  true, /* is local */
 							  0,	/* inhcount */
 							  false,	/* connoinherit */
+							  false,	/* conperiod */
 							  false);	/* is_internal */
 	if (constrAddr)
 		ObjectAddressSet(*constrAddr, ConstraintRelationId, ccoid);
@@ -3727,6 +3728,7 @@ domainAddNotNullConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
 							  true, /* is local */
 							  0,	/* inhcount */
 							  false,	/* connoinherit */
+							  false,	/* conperiod */
 							  false);	/* is_internal */
 
 	if (constrAddr)
diff --git a/src/backend/executor/execIndexing.c b/src/backend/executor/execIndexing.c
index 403a3f40551..f9a2fac79e4 100644
--- a/src/backend/executor/execIndexing.c
+++ b/src/backend/executor/execIndexing.c
@@ -114,6 +114,8 @@
 #include "executor/executor.h"
 #include "nodes/nodeFuncs.h"
 #include "storage/lmgr.h"
+#include "utils/multirangetypes.h"
+#include "utils/rangetypes.h"
 #include "utils/snapmgr.h"
 
 /* waitMode argument to check_exclusion_or_unique_constraint() */
@@ -141,6 +143,8 @@ static bool index_unchanged_by_update(ResultRelInfo *resultRelInfo,
 									  Relation indexRelation);
 static bool index_expression_changed_walker(Node *node,
 											Bitmapset *allUpdatedCols);
+static void ExecWithoutOverlapsNotEmpty(Relation rel, NameData attname, Datum attval,
+										char typtype, Oid atttypid);
 
 /* ----------------------------------------------------------------
  *		ExecOpenIndices
@@ -211,7 +215,7 @@ ExecOpenIndices(ResultRelInfo *resultRelInfo, bool speculative)
 		 * detection in logical replication, add extra information required by
 		 * unique index entries.
 		 */
-		if (speculative && ii->ii_Unique)
+		if (speculative && ii->ii_Unique && !indexDesc->rd_index->indisexclusion)
 			BuildSpeculativeIndexInfo(indexDesc, ii);
 
 		relationDescs[i] = indexDesc;
@@ -726,6 +730,32 @@ check_exclusion_or_unique_constraint(Relation heap, Relation index,
 	}
 
 	/*
+	 * If this is a WITHOUT OVERLAPS constraint, we must also forbid empty
+	 * ranges/multiranges. This must happen before we look for NULLs below, or
+	 * a UNIQUE constraint could insert an empty range along with a NULL
+	 * scalar part.
+	 */
+	if (indexInfo->ii_WithoutOverlaps)
+	{
+		/*
+		 * Look up the type from the heap tuple, but check the Datum from the
+		 * index tuple.
+		 */
+		AttrNumber	attno = indexInfo->ii_IndexAttrNumbers[indnkeyatts - 1];
+
+		if (!isnull[indnkeyatts - 1])
+		{
+			TupleDesc	tupdesc = RelationGetDescr(heap);
+			Form_pg_attribute att = TupleDescAttr(tupdesc, attno - 1);
+			TypeCacheEntry *typcache = lookup_type_cache(att->atttypid, 0);
+
+			ExecWithoutOverlapsNotEmpty(heap, att->attname,
+										values[indnkeyatts - 1],
+										typcache->typtype, att->atttypid);
+		}
+	}
+
+	/*
 	 * If any of the input values are NULL, and the index uses the default
 	 * nulls-are-distinct mode, the constraint check is assumed to pass (i.e.,
 	 * we assume the operators are strict).  Otherwise, we interpret the
@@ -1102,3 +1132,37 @@ index_expression_changed_walker(Node *node, Bitmapset *allUpdatedCols)
 	return expression_tree_walker(node, index_expression_changed_walker,
 								  (void *) allUpdatedCols);
 }
+
+/*
+ * ExecWithoutOverlapsNotEmpty - raise an error if the tuple has an empty
+ * range or multirange in the given attribute.
+ */
+static void
+ExecWithoutOverlapsNotEmpty(Relation rel, NameData attname, Datum attval, char typtype, Oid atttypid)
+{
+	bool		isempty;
+	RangeType  *r;
+	MultirangeType *mr;
+
+	switch (typtype)
+	{
+		case TYPTYPE_RANGE:
+			r = DatumGetRangeTypeP(attval);
+			isempty = RangeIsEmpty(r);
+			break;
+		case TYPTYPE_MULTIRANGE:
+			mr = DatumGetMultirangeTypeP(attval);
+			isempty = MultirangeIsEmpty(mr);
+			break;
+		default:
+			elog(ERROR, "WITHOUT OVERLAPS column \"%s\" is not a range or multirange",
+				 NameStr(attname));
+	}
+
+	/* Report a CHECK_VIOLATION */
+	if (isempty)
+		ereport(ERROR,
+				(errcode(ERRCODE_CHECK_VIOLATION),
+				 errmsg("empty WITHOUT OVERLAPS value found in column \"%s\" in relation \"%s\"",
+						NameStr(attname), RelationGetRelationName(rel))));
+}
diff --git a/src/backend/nodes/makefuncs.c b/src/backend/nodes/makefuncs.c
index 61ac172a857..9cac3c1c27b 100644
--- a/src/backend/nodes/makefuncs.c
+++ b/src/backend/nodes/makefuncs.c
@@ -760,7 +760,8 @@ make_ands_implicit(Expr *clause)
 IndexInfo *
 makeIndexInfo(int numattrs, int numkeyattrs, Oid amoid, List *expressions,
 			  List *predicates, bool unique, bool nulls_not_distinct,
-			  bool isready, bool concurrent, bool summarizing)
+			  bool isready, bool concurrent, bool summarizing,
+			  bool withoutoverlaps)
 {
 	IndexInfo  *n = makeNode(IndexInfo);
 
@@ -775,6 +776,7 @@ makeIndexInfo(int numattrs, int numkeyattrs, Oid amoid, List *expressions,
 	n->ii_IndexUnchanged = false;
 	n->ii_Concurrent = concurrent;
 	n->ii_Summarizing = summarizing;
+	n->ii_WithoutOverlaps = withoutoverlaps;
 
 	/* summarizing indexes cannot contain non-key attributes */
 	Assert(!summarizing || (numkeyattrs == numattrs));
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 82f031f4cfe..b913f91ff03 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -828,7 +828,7 @@ infer_arbiter_indexes(PlannerInfo *root)
 		 */
 		if (indexOidFromConstraint == idxForm->indexrelid)
 		{
-			if (!idxForm->indisunique && onconflict->action == ONCONFLICT_UPDATE)
+			if (idxForm->indisexclusion && onconflict->action == ONCONFLICT_UPDATE)
 				ereport(ERROR,
 						(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 						 errmsg("ON CONFLICT DO UPDATE not supported with exclusion constraints")));
@@ -853,6 +853,13 @@ infer_arbiter_indexes(PlannerInfo *root)
 		if (!idxForm->indisunique)
 			goto next;
 
+		/*
+		 * So-called unique constraints with WITHOUT OVERLAPS are really
+		 * exclusion constraints, so skip those too.
+		 */
+		if (idxForm->indisexclusion)
+			goto next;
+
 		/* Build BMS representation of plain (non expression) index attrs */
 		indexedAttrs = NULL;
 		for (natt = 0; natt < idxForm->indnkeyatts; natt++)
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 84cef57a707..c8b4e8dde4c 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -531,7 +531,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				columnref in_expr having_clause func_table xmltable array_expr
 				OptWhereClause operator_def_arg
 %type <list>	rowsfrom_item rowsfrom_list opt_col_def_list
-%type <boolean> opt_ordinality
+%type <boolean> opt_ordinality opt_without_overlaps
 %type <list>	ExclusionConstraintList ExclusionConstraintElem
 %type <list>	func_arg_list func_arg_list_opt
 %type <node>	func_arg_expr
@@ -4141,7 +4141,7 @@ ConstraintElem:
 					n->initially_valid = !n->skip_validation;
 					$$ = (Node *) n;
 				}
-			| UNIQUE opt_unique_null_treatment '(' columnList ')' opt_c_include opt_definition OptConsTableSpace
+			| UNIQUE opt_unique_null_treatment '(' columnList opt_without_overlaps ')' opt_c_include opt_definition OptConsTableSpace
 				ConstraintAttributeSpec
 				{
 					Constraint *n = makeNode(Constraint);
@@ -4150,11 +4150,12 @@ ConstraintElem:
 					n->location = @1;
 					n->nulls_not_distinct = !$2;
 					n->keys = $4;
-					n->including = $6;
-					n->options = $7;
+					n->without_overlaps = $5;
+					n->including = $7;
+					n->options = $8;
 					n->indexname = NULL;
-					n->indexspace = $8;
-					processCASbits($9, @9, "UNIQUE",
+					n->indexspace = $9;
+					processCASbits($10, @10, "UNIQUE",
 								   &n->deferrable, &n->initdeferred, NULL,
 								   NULL, yyscanner);
 					$$ = (Node *) n;
@@ -4175,7 +4176,7 @@ ConstraintElem:
 								   NULL, yyscanner);
 					$$ = (Node *) n;
 				}
-			| PRIMARY KEY '(' columnList ')' opt_c_include opt_definition OptConsTableSpace
+			| PRIMARY KEY '(' columnList opt_without_overlaps ')' opt_c_include opt_definition OptConsTableSpace
 				ConstraintAttributeSpec
 				{
 					Constraint *n = makeNode(Constraint);
@@ -4183,11 +4184,12 @@ ConstraintElem:
 					n->contype = CONSTR_PRIMARY;
 					n->location = @1;
 					n->keys = $4;
-					n->including = $6;
-					n->options = $7;
+					n->without_overlaps = $5;
+					n->including = $7;
+					n->options = $8;
 					n->indexname = NULL;
-					n->indexspace = $8;
-					processCASbits($9, @9, "PRIMARY KEY",
+					n->indexspace = $9;
+					processCASbits($10, @10, "PRIMARY KEY",
 								   &n->deferrable, &n->initdeferred, NULL,
 								   NULL, yyscanner);
 					$$ = (Node *) n;
@@ -4309,6 +4311,11 @@ opt_no_inherit:	NO INHERIT							{  $$ = true; }
 			| /* EMPTY */							{  $$ = false; }
 		;
 
+opt_without_overlaps:
+			WITHOUT OVERLAPS						{ $$ = true; }
+			| /*EMPTY*/								{ $$ = false; }
+	;
+
 opt_column_list:
 			'(' columnList ')'						{ $$ = $2; }
 			| /*EMPTY*/								{ $$ = NIL; }
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 79cad4ab30c..15274607542 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1555,6 +1555,7 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 	index->unique = idxrec->indisunique;
 	index->nulls_not_distinct = idxrec->indnullsnotdistinct;
 	index->primary = idxrec->indisprimary;
+	index->iswithoutoverlaps = (idxrec->indisprimary || idxrec->indisunique) && idxrec->indisexclusion;
 	index->transformed = true;	/* don't need transformIndexStmt */
 	index->concurrent = false;
 	index->if_not_exists = false;
@@ -1604,7 +1605,9 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 				int			nElems;
 				int			i;
 
-				Assert(conrec->contype == CONSTRAINT_EXCLUSION);
+				Assert(conrec->contype == CONSTRAINT_EXCLUSION ||
+					   (index->iswithoutoverlaps &&
+						(conrec->contype == CONSTRAINT_PRIMARY || conrec->contype == CONSTRAINT_UNIQUE)));
 				/* Extract operator OIDs from the pg_constraint tuple */
 				datum = SysCacheGetAttrNotNull(CONSTROID, ht_constr,
 											   Anum_pg_constraint_conexclop);
@@ -2157,6 +2160,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 	}
 	index->nulls_not_distinct = constraint->nulls_not_distinct;
 	index->isconstraint = true;
+	index->iswithoutoverlaps = constraint->without_overlaps;
 	index->deferrable = constraint->deferrable;
 	index->initdeferred = constraint->initdeferred;
 
@@ -2249,6 +2253,11 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 					 errmsg("index \"%s\" is not valid", index_name),
 					 parser_errposition(cxt->pstate, constraint->location)));
 
+		/*
+		 * Today we forbid non-unique indexes, but we could permit GiST
+		 * indexes whose last entry is a range type and use that to create a
+		 * WITHOUT OVERLAPS constraint (i.e. a temporal constraint).
+		 */
 		if (!index_form->indisunique)
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
@@ -2385,7 +2394,8 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 	 * For UNIQUE and PRIMARY KEY, we just have a list of column names.
 	 *
 	 * Make sure referenced keys exist.  If we are making a PRIMARY KEY index,
-	 * also make sure they are NOT NULL.
+	 * also make sure they are NOT NULL.  For WITHOUT OVERLAPS constraints, we
+	 * make sure the last part is a range or multirange.
 	 */
 	else
 	{
@@ -2397,6 +2407,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 			ColumnDef  *column = NULL;
 			ListCell   *columns;
 			IndexElem  *iparam;
+			Oid			typid = InvalidOid;
 
 			/* Make sure referenced column exists. */
 			foreach(columns, cxt->columns)
@@ -2408,6 +2419,9 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 					break;
 				}
 			}
+			if (!found)
+				column = NULL;
+
 			if (found)
 			{
 				/*
@@ -2463,6 +2477,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 						if (strcmp(key, inhname) == 0)
 						{
 							found = true;
+							typid = inhattr->atttypid;
 
 							/*
 							 * It's tempting to set forced_not_null if the
@@ -2512,6 +2527,50 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 				}
 			}
 
+			/*
+			 * The WITHOUT OVERLAPS part (if any) must be a range or
+			 * multirange type.
+			 */
+			if (constraint->without_overlaps && lc == list_last_cell(constraint->keys))
+			{
+				if (!found && cxt->isalter)
+				{
+					/*
+					 * Look up the column type on existing table. If we can't
+					 * find it, let things fail in DefineIndex.
+					 */
+					Relation	rel = cxt->rel;
+
+					for (int i = 0; i < rel->rd_att->natts; i++)
+					{
+						Form_pg_attribute attr = TupleDescAttr(rel->rd_att, i);
+						const char *attname;
+
+						if (attr->attisdropped)
+							break;
+
+						attname = NameStr(attr->attname);
+						if (strcmp(attname, key) == 0)
+						{
+							found = true;
+							typid = attr->atttypid;
+							break;
+						}
+					}
+				}
+				if (found)
+				{
+					if (!OidIsValid(typid) && column)
+						typid = typenameTypeId(NULL, column->typeName);
+
+					if (!OidIsValid(typid) || !(type_is_range(typid) || type_is_multirange(typid)))
+						ereport(ERROR,
+								(errcode(ERRCODE_DATATYPE_MISMATCH),
+								 errmsg("column \"%s\" in WITHOUT OVERLAPS is not a range or multirange type", key),
+								 parser_errposition(cxt->pstate, constraint->location)));
+				}
+			}
+
 			/* OK, add it to the index definition */
 			iparam = makeNode(IndexElem);
 			iparam->name = pstrdup(key);
@@ -2537,6 +2596,23 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 				notnullcmds = lappend(notnullcmds, notnullcmd);
 			}
 		}
+
+		if (constraint->without_overlaps)
+		{
+			/*
+			 * This enforces that there is at least one equality column
+			 * besides the WITHOUT OVERLAPS columns.  This is per SQL
+			 * standard.  XXX Do we need this?
+			 */
+			if (list_length(constraint->keys) < 2)
+				ereport(ERROR,
+						errcode(ERRCODE_SYNTAX_ERROR),
+						errmsg("constraint using WITHOUT OVERLAPS needs at least two columns"));
+
+			/* WITHOUT OVERLAPS requires a GiST index */
+			index->accessMethod = "gist";
+		}
+
 	}
 
 	/*
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index badbf111ee0..c05d41ce023 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -2403,6 +2403,8 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand,
 											 Anum_pg_constraint_conkey);
 
 				keyatts = decompile_column_index_array(val, conForm->conrelid, &buf);
+				if (conForm->conperiod)
+					appendStringInfoString(&buf, " WITHOUT OVERLAPS");
 
 				appendStringInfoChar(&buf, ')');
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 63efc55f09e..5b6b7b809c0 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -5581,11 +5581,14 @@ RelationGetIdentityKeyBitmap(Relation relation)
 /*
  * RelationGetExclusionInfo -- get info about index's exclusion constraint
  *
- * This should be called only for an index that is known to have an
- * associated exclusion constraint.  It returns arrays (palloc'd in caller's
- * context) of the exclusion operator OIDs, their underlying functions'
- * OIDs, and their strategy numbers in the index's opclasses.  We cache
- * all this information since it requires a fair amount of work to get.
+ * This should be called only for an index that is known to have an associated
+ * exclusion constraint or primary key/unique constraint using WITHOUT
+ * OVERLAPS.
+
+ * It returns arrays (palloc'd in caller's context) of the exclusion operator
+ * OIDs, their underlying functions' OIDs, and their strategy numbers in the
+ * index's opclasses.  We cache all this information since it requires a fair
+ * amount of work to get.
  */
 void
 RelationGetExclusionInfo(Relation indexRelation,
@@ -5649,7 +5652,10 @@ RelationGetExclusionInfo(Relation indexRelation,
 		int			nelem;
 
 		/* We want the exclusion constraint owning the index */
-		if (conform->contype != CONSTRAINT_EXCLUSION ||
+		if ((conform->contype != CONSTRAINT_EXCLUSION &&
+			 !(conform->conperiod && (
+									  conform->contype == CONSTRAINT_PRIMARY
+									  || conform->contype == CONSTRAINT_UNIQUE))) ||
 			conform->conindid != RelationGetRelid(indexRelation))
 			continue;