Add exclusion constraints, which generalize the concept of uniqueness to

support any indexable commutative operator, not just equality.  Two rows
violate the exclusion constraint if "row1.col OP row2.col" is TRUE for
each of the columns in the constraint.

Jeff Davis, reviewed by Robert Haas

5 files changed, 193 insertions, 25 deletions

diff --git a/src/backend/commands/constraint.c b/src/backend/commands/constraint.c
index 42d4d4e1f9..41adf87147 100644
--- a/src/backend/commands/constraint.c
+++ b/src/backend/commands/constraint.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/constraint.c,v 1.1 2009/07/29 20:56:18 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/constraint.c,v 1.2 2009/12/07 05:22:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -23,9 +23,12 @@
 /*
  * unique_key_recheck - trigger function to do a deferred uniqueness check.
  *
+ * This now also does deferred exclusion-constraint checks, so the name is
+ * somewhat historical.
+ *
  * This is invoked as an AFTER ROW trigger for both INSERT and UPDATE,
  * for any rows recorded as potentially violating a deferrable unique
- * constraint.
+ * or exclusion constraint.
  *
  * This may be an end-of-statement check, a commit-time check, or a
  * check triggered by a SET CONSTRAINTS command.
@@ -85,7 +88,7 @@ unique_key_recheck(PG_FUNCTION_ARGS)
 	 * because this trigger gets queued only in response to index insertions;
 	 * which means it does not get queued for HOT updates.  The row we are
 	 * called for might now be dead, but have a live HOT child, in which case
-	 * we still need to make the uniqueness check.  Therefore we have to use
+	 * we still need to make the check.  Therefore we have to use
 	 * heap_hot_search, not just HeapTupleSatisfiesVisibility as is done in
 	 * the comparable test in RI_FKey_check.
 	 *
@@ -123,9 +126,11 @@ unique_key_recheck(PG_FUNCTION_ARGS)
 
 	/*
 	 * Typically the index won't have expressions, but if it does we need
-	 * an EState to evaluate them.
+	 * an EState to evaluate them.  We need it for exclusion constraints
+	 * too, even if they are just on simple columns.
 	 */
-	if (indexInfo->ii_Expressions != NIL)
+	if (indexInfo->ii_Expressions != NIL ||
+		indexInfo->ii_ExclusionOps != NULL)
 	{
 		estate = CreateExecutorState();
 		econtext = GetPerTupleExprContext(estate);
@@ -141,19 +146,37 @@ unique_key_recheck(PG_FUNCTION_ARGS)
 	 * Note: if the index uses functions that are not as immutable as they
 	 * are supposed to be, this could produce an index tuple different from
 	 * the original.  The index AM can catch such errors by verifying that
-	 * it finds a matching index entry with the tuple's TID.
+	 * it finds a matching index entry with the tuple's TID.  For exclusion
+	 * constraints we check this in check_exclusion_constraint().
 	 */
 	FormIndexDatum(indexInfo, slot, estate, values, isnull);
 
 	/*
-	 * Now do the uniqueness check. This is not a real insert; it is a
-	 * check that the index entry that has already been inserted is unique.
+	 * Now do the appropriate check.
 	 */
-	index_insert(indexRel, values, isnull, &(new_row->t_self),
-				 trigdata->tg_relation, UNIQUE_CHECK_EXISTING);
+	if (indexInfo->ii_ExclusionOps == NULL)
+	{
+		/*
+		 * Note: this is not a real insert; it is a check that the index entry
+		 * that has already been inserted is unique.
+		 */
+		index_insert(indexRel, values, isnull, &(new_row->t_self),
+					 trigdata->tg_relation, UNIQUE_CHECK_EXISTING);
+	}
+	else
+	{
+		/*
+		 * For exclusion constraints we just do the normal check, but now
+		 * it's okay to throw error.
+		 */
+		check_exclusion_constraint(trigdata->tg_relation, indexRel, indexInfo,
+								   &(new_row->t_self), values, isnull,
+								   estate, false, false);
+	}
 
 	/*
-	 * If that worked, then this index entry is unique, and we are done.
+	 * If that worked, then this index entry is unique or non-excluded,
+	 * and we are done.
 	 */
 	if (estate != NULL)
 		FreeExecutorState(estate);
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 96272ab998..00786442b3 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/indexcmds.c,v 1.187 2009/07/29 20:56:18 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/indexcmds.c,v 1.188 2009/12/07 05:22:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -25,6 +25,7 @@
 #include "catalog/index.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_opclass.h"
+#include "catalog/pg_opfamily.h"
 #include "catalog/pg_tablespace.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
@@ -36,6 +37,7 @@
 #include "optimizer/clauses.h"
 #include "parser/parse_coerce.h"
 #include "parser/parse_func.h"
+#include "parser/parse_oper.h"
 #include "parser/parsetree.h"
 #include "storage/lmgr.h"
 #include "storage/proc.h"
@@ -58,6 +60,7 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo,
 				  Oid *classOidP,
 				  int16 *colOptionP,
 				  List *attList,
+				  List *exclusionOpNames,
 				  Oid relId,
 				  char *accessMethodName, Oid accessMethodId,
 				  bool amcanorder,
@@ -83,6 +86,8 @@ static bool relationHasPrimaryKey(Relation rel);
  *		to index on.
  * 'predicate': the partial-index condition, or NULL if none.
  * 'options': reloptions from WITH (in list-of-DefElem form).
+ * 'exclusionOpNames': list of names of exclusion-constraint operators,
+ *		or NIL if not an exclusion constraint.
  * 'unique': make the index enforce uniqueness.
  * 'primary': mark the index as a primary key in the catalogs.
  * 'isconstraint': index is for a PRIMARY KEY or UNIQUE constraint,
@@ -106,6 +111,7 @@ DefineIndex(RangeVar *heapRelation,
 			List *attributeList,
 			Expr *predicate,
 			List *options,
+			List *exclusionOpNames,
 			bool unique,
 			bool primary,
 			bool isconstraint,
@@ -247,10 +253,21 @@ DefineIndex(RangeVar *heapRelation,
 	if (indexRelationName == NULL)
 	{
 		if (primary)
+		{
 			indexRelationName = ChooseRelationName(RelationGetRelationName(rel),
 												   NULL,
 												   "pkey",
 												   namespaceId);
+		}
+		else if (exclusionOpNames != NIL)
+		{
+			IndexElem  *iparam = (IndexElem *) linitial(attributeList);
+
+			indexRelationName = ChooseRelationName(RelationGetRelationName(rel),
+												   iparam->name,
+												   "exclusion",
+												   namespaceId);
+		}
 		else
 		{
 			IndexElem  *iparam = (IndexElem *) linitial(attributeList);
@@ -303,6 +320,11 @@ DefineIndex(RangeVar *heapRelation,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 		  errmsg("access method \"%s\" does not support multicolumn indexes",
 				 accessMethodName)));
+	if (exclusionOpNames != NIL && !OidIsValid(accessMethodForm->amgettuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("access method \"%s\" does not support exclusion constraints",
+						accessMethodName)));
 
 	amcanorder = accessMethodForm->amcanorder;
 	amoptions = accessMethodForm->amoptions;
@@ -418,6 +440,9 @@ DefineIndex(RangeVar *heapRelation,
 	indexInfo->ii_ExpressionsState = NIL;
 	indexInfo->ii_Predicate = make_ands_implicit(predicate);
 	indexInfo->ii_PredicateState = NIL;
+	indexInfo->ii_ExclusionOps = NULL;
+	indexInfo->ii_ExclusionProcs = NULL;
+	indexInfo->ii_ExclusionStrats = NULL;
 	indexInfo->ii_Unique = unique;
 	/* In a concurrent build, mark it not-ready-for-inserts */
 	indexInfo->ii_ReadyForInserts = !concurrent;
@@ -427,7 +452,8 @@ DefineIndex(RangeVar *heapRelation,
 	classObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
 	coloptions = (int16 *) palloc(numberOfAttributes * sizeof(int16));
 	ComputeIndexAttrs(indexInfo, classObjectId, coloptions, attributeList,
-					  relationId, accessMethodName, accessMethodId,
+					  exclusionOpNames, relationId,
+					  accessMethodName, accessMethodId,
 					  amcanorder, isconstraint);
 
 	/*
@@ -435,11 +461,27 @@ DefineIndex(RangeVar *heapRelation,
 	 * error checks)
 	 */
 	if (isconstraint && !quiet)
+	{
+		const char *constraint_type;
+
+		if (primary)
+			constraint_type = "PRIMARY KEY";
+		else if (unique)
+			constraint_type = "UNIQUE";
+		else if (exclusionOpNames != NIL)
+			constraint_type = "EXCLUDE";
+		else
+		{
+			elog(ERROR, "unknown constraint type");
+			constraint_type = NULL;	/* keep compiler quiet */
+		}
+
 		ereport(NOTICE,
 		  (errmsg("%s %s will create implicit index \"%s\" for table \"%s\"",
 				  is_alter_table ? "ALTER TABLE / ADD" : "CREATE TABLE /",
-				  primary ? "PRIMARY KEY" : "UNIQUE",
+				  constraint_type,
 				  indexRelationName, RelationGetRelationName(rel))));
+	}
 
 	/* save lockrelid and locktag for below, then close rel */
 	heaprelid = rel->rd_lockInfo.lockRelId;
@@ -799,21 +841,38 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 				  Oid *classOidP,
 				  int16 *colOptionP,
 				  List *attList,	/* list of IndexElem's */
+				  List *exclusionOpNames,
 				  Oid relId,
 				  char *accessMethodName,
 				  Oid accessMethodId,
 				  bool amcanorder,
 				  bool isconstraint)
 {
-	ListCell   *rest;
-	int			attn = 0;
+	ListCell   *nextExclOp;
+	ListCell   *lc;
+	int			attn;
+
+	/* Allocate space for exclusion operator info, if needed */
+	if (exclusionOpNames)
+	{
+		int		ncols = list_length(attList);
+
+		Assert(list_length(exclusionOpNames) == ncols);
+		indexInfo->ii_ExclusionOps = (Oid *) palloc(sizeof(Oid) * ncols);
+		indexInfo->ii_ExclusionProcs = (Oid *) palloc(sizeof(Oid) * ncols);
+		indexInfo->ii_ExclusionStrats = (uint16 *) palloc(sizeof(uint16) * ncols);
+		nextExclOp = list_head(exclusionOpNames);
+	}
+	else
+		nextExclOp = NULL;
 
 	/*
 	 * process attributeList
 	 */
-	foreach(rest, attList)
+	attn = 0;
+	foreach(lc, attList)
 	{
-		IndexElem  *attribute = (IndexElem *) lfirst(rest);
+		IndexElem  *attribute = (IndexElem *) lfirst(lc);
 		Oid			atttype;
 
 		/*
@@ -898,6 +957,71 @@ ComputeIndexAttrs(IndexInfo *indexInfo,
 										  accessMethodId);
 
 		/*
+		 * Identify the exclusion operator, if any.
+		 */
+		if (nextExclOp)
+		{
+			List   *opname = (List *) lfirst(nextExclOp);
+			Oid		opid;
+			Oid		opfamily;
+			int		strat;
+
+			/*
+			 * Find the operator --- it must accept the column datatype
+			 * without runtime coercion (but binary compatibility is OK)
+			 */
+			opid = compatible_oper_opid(opname, atttype, atttype, false);
+
+			/*
+			 * Only allow commutative operators to be used in exclusion
+			 * constraints. If X conflicts with Y, but Y does not conflict
+			 * with X, bad things will happen.
+			 */
+			if (get_commutator(opid) != opid)
+				ereport(ERROR,
+						(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+						 errmsg("operator %s is not commutative",
+								format_operator(opid)),
+						 errdetail("Only commutative operators can be used in exclusion constraints.")));
+
+			/*
+			 * Operator must be a member of the right opfamily, too
+			 */
+			opfamily = get_opclass_family(classOidP[attn]);
+			strat = get_op_opfamily_strategy(opid, opfamily);
+			if (strat == 0)
+			{
+				HeapTuple opftuple;
+				Form_pg_opfamily opfform;
+
+				/*
+				 * attribute->opclass might not explicitly name the opfamily,
+				 * so fetch the name of the selected opfamily for use in the
+				 * error message.
+				 */
+				opftuple = SearchSysCache(OPFAMILYOID,
+										  ObjectIdGetDatum(opfamily),
+										  0, 0, 0);
+				if (!HeapTupleIsValid(opftuple))
+					elog(ERROR, "cache lookup failed for opfamily %u",
+						 opfamily);
+				opfform = (Form_pg_opfamily) GETSTRUCT(opftuple);
+
+				ereport(ERROR,
+						(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+						 errmsg("operator %s is not a member of operator family \"%s\"",
+								format_operator(opid),
+								NameStr(opfform->opfname)),
+						 errdetail("The exclusion operator must be related to the index operator class for the constraint.")));
+			}
+
+			indexInfo->ii_ExclusionOps[attn] = opid;
+			indexInfo->ii_ExclusionProcs[attn] = get_opcode(opid);
+			indexInfo->ii_ExclusionStrats[attn] = strat;
+			nextExclOp = lnext(nextExclOp);
+		}
+
+		/*
 		 * Set up the per-column options (indoption field).  For now, this is
 		 * zero for any un-ordered index, while ordered indexes have DESC and
 		 * NULLS FIRST/LAST options.
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1de1950453..d10a7490b7 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.306 2009/11/20 20:38:10 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.307 2009/12/07 05:22:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -4603,6 +4603,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 				stmt->indexParams,		/* parameters */
 				(Expr *) stmt->whereClause,
 				stmt->options,
+				stmt->excludeOpNames,
 				stmt->unique,
 				stmt->primary,
 				stmt->isconstraint,
@@ -5035,6 +5036,7 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
 									  fkconstraint->fk_upd_action,
 									  fkconstraint->fk_del_action,
 									  fkconstraint->fk_matchtype,
+									  NULL,		/* no exclusion constraint */
 									  NULL,		/* no check constraint */
 									  NULL,
 									  NULL,
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 4f997a0dce..54d8220cb3 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.138 2009/10/08 02:39:19 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.139 2009/12/07 05:22:21 tgl Exp $
  *
  * DESCRIPTION
  *	  The "DefineFoo" routines take the parse tree and pick out the
@@ -984,6 +984,12 @@ DefineDomain(CreateDomainStmt *stmt)
 				errmsg("primary key constraints not possible for domains")));
 				break;
 
+			case CONSTR_EXCLUSION:
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("exclusion constraints not possible for domains")));
+				break;
+
 			case CONSTR_FOREIGN:
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
@@ -1868,6 +1874,12 @@ AlterDomainAddConstraint(List *names, Node *newConstraint)
 				errmsg("primary key constraints not possible for domains")));
 			break;
 
+		case CONSTR_EXCLUSION:
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("exclusion constraints not possible for domains")));
+			break;
+
 		case CONSTR_FOREIGN:
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -2297,9 +2309,10 @@ domainAddConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
 						  ' ',
 						  ' ',
 						  ' ',
-						  expr, /* Tree form check constraint */
-						  ccbin,	/* Binary form check constraint */
-						  ccsrc,	/* Source form check constraint */
+						  NULL,	/* not an exclusion constraint */
+						  expr, /* Tree form of check constraint */
+						  ccbin,	/* Binary form of check constraint */
+						  ccsrc,	/* Source form of check constraint */
 						  true, /* is local */
 						  0);	/* inhcount */
 
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index eeee7654e0..83b8c18e8e 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -13,7 +13,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/vacuum.c,v 1.396 2009/11/16 21:32:06 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/vacuum.c,v 1.397 2009/12/07 05:22:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -763,7 +763,8 @@ vac_update_relstats(Relation relation,
 
 	/*
 	 * If we have discovered that there are no indexes, then there's no
-	 * primary key either.	This could be done more thoroughly...
+	 * primary key either, nor any exclusion constraints.  This could be done
+	 * more thoroughly...
 	 */
 	if (!hasindex)
 	{
@@ -772,6 +773,11 @@ vac_update_relstats(Relation relation,
 			pgcform->relhaspkey = false;
 			dirty = true;
 		}
+		if (pgcform->relhasexclusion && pgcform->relkind != RELKIND_INDEX)
+		{
+			pgcform->relhasexclusion = false;
+			dirty = true;
+		}
 	}
 
 	/* We also clear relhasrules and relhastriggers if needed */