Support deferrable uniqueness constraints.

The current implementation fires an AFTER ROW trigger for each tuple that looks like it might be non-unique according to the index contents at the time of insertion. This works well as long as there aren't many conflicts, but won't scale to massive unique-key reassignments. Improving that case is a TODO item. Dean Rasheed
2025-08-25 20:23:07 +03:00 · 2009-07-29 20:56:21 +00:00
parent 8504905793
commit 25d9bf2e3e
51 changed files with 1241 additions and 245 deletions
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -4,7 +4,7 @@
 #    Makefile for backend/commands
 #
 # IDENTIFICATION
-#    $PostgreSQL: pgsql/src/backend/commands/Makefile,v 1.39 2008/12/19 16:25:17 petere Exp $
+#    $PostgreSQL: pgsql/src/backend/commands/Makefile,v 1.40 2009/07/29 20:56:18 tgl Exp $
 #
 #-------------------------------------------------------------------------

@@ -13,7 +13,7 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global

 OBJS = aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o  \
-	conversioncmds.o copy.o \
+	constraint.o conversioncmds.o copy.o \
 	dbcommands.o define.o discard.o explain.o foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o operatorcmds.o opclasscmds.o \
 	portalcmds.o prepare.o proclang.o \
--- a/src/backend/commands/constraint.c
+++ b/src/backend/commands/constraint.c
@@ -0,0 +1,166 @@
+/*-------------------------------------------------------------------------
+ *
+ * constraint.c
+ *	  PostgreSQL CONSTRAINT support code.
+ *
+ * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
+ * 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 $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "catalog/index.h"
+#include "commands/trigger.h"
+#include "executor/executor.h"
+#include "utils/builtins.h"
+#include "utils/tqual.h"
+
+
+/*
+ * unique_key_recheck - trigger function to do a deferred uniqueness check.
+ *
+ * This is invoked as an AFTER ROW trigger for both INSERT and UPDATE,
+ * for any rows recorded as potentially violating a deferrable unique
+ * constraint.
+ *
+ * This may be an end-of-statement check, a commit-time check, or a
+ * check triggered by a SET CONSTRAINTS command.
+ */
+Datum
+unique_key_recheck(PG_FUNCTION_ARGS)
+{
+	TriggerData *trigdata = (TriggerData *) fcinfo->context;
+	const char *funcname = "unique_key_recheck";
+	HeapTuple	new_row;
+	ItemPointerData tmptid;
+	Relation	indexRel;
+	IndexInfo  *indexInfo;
+	EState	   *estate;
+	ExprContext *econtext;
+	TupleTableSlot *slot;
+	Datum		values[INDEX_MAX_KEYS];
+	bool		isnull[INDEX_MAX_KEYS];
+
+	/*
+	 * Make sure this is being called as an AFTER ROW trigger.  Note:
+	 * translatable error strings are shared with ri_triggers.c, so
+	 * resist the temptation to fold the function name into them.
+	 */
+	if (!CALLED_AS_TRIGGER(fcinfo))
+		ereport(ERROR,
+				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+				 errmsg("function \"%s\" was not called by trigger manager",
+						funcname)));
+
+	if (!TRIGGER_FIRED_AFTER(trigdata->tg_event) ||
+		!TRIGGER_FIRED_FOR_ROW(trigdata->tg_event))
+		ereport(ERROR,
+				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+				 errmsg("function \"%s\" must be fired AFTER ROW",
+						funcname)));
+
+	/*
+	 * Get the new data that was inserted/updated.
+	 */
+	if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+		new_row = trigdata->tg_trigtuple;
+	else if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
+		new_row = trigdata->tg_newtuple;
+	else
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
+				 errmsg("function \"%s\" must be fired for INSERT or UPDATE",
+						funcname)));
+		new_row = NULL;			/* keep compiler quiet */
+	}
+
+	/*
+	 * If the new_row is now dead (ie, inserted and then deleted within our
+	 * transaction), we can skip the check.  However, we have to be careful,
+	 * 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
+	 * heap_hot_search, not just HeapTupleSatisfiesVisibility as is done in
+	 * the comparable test in RI_FKey_check.
+	 *
+	 * This might look like just an optimization, because the index AM will
+	 * make this identical test before throwing an error.  But it's actually
+	 * needed for correctness, because the index AM will also throw an error
+	 * if it doesn't find the index entry for the row.  If the row's dead then
+	 * it's possible the index entry has also been marked dead, and even
+	 * removed.
+	 */
+	tmptid = new_row->t_self;
+	if (!heap_hot_search(&tmptid, trigdata->tg_relation, SnapshotSelf, NULL))
+	{
+		/*
+		 * All rows in the HOT chain are dead, so skip the check.
+		 */
+		return PointerGetDatum(NULL);
+	}
+
+	/*
+	 * Open the index, acquiring a RowExclusiveLock, just as if we were
+	 * going to update it.  (This protects against possible changes of the
+	 * index schema, not against concurrent updates.)
+	 */
+	indexRel = index_open(trigdata->tg_trigger->tgconstrindid,
+						  RowExclusiveLock);
+	indexInfo = BuildIndexInfo(indexRel);
+
+	/*
+	 * The heap tuple must be put into a slot for FormIndexDatum.
+	 */
+	slot = MakeSingleTupleTableSlot(RelationGetDescr(trigdata->tg_relation));
+
+	ExecStoreTuple(new_row, slot, InvalidBuffer, false);
+
+	/*
+	 * Typically the index won't have expressions, but if it does we need
+	 * an EState to evaluate them.
+	 */
+	if (indexInfo->ii_Expressions != NIL)
+	{
+		estate = CreateExecutorState();
+		econtext = GetPerTupleExprContext(estate);
+		econtext->ecxt_scantuple = slot;
+	}
+	else
+		estate = NULL;
+
+	/*
+	 * Form the index values and isnull flags for the index entry that
+	 * we need to check.
+	 *
+	 * 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.
+	 */
+	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.
+	 */
+	index_insert(indexRel, values, isnull, &(new_row->t_self),
+				 trigdata->tg_relation, UNIQUE_CHECK_EXISTING);
+
+	/*
+	 * If that worked, then this index entry is unique, and we are done.
+	 */
+	if (estate != NULL)
+		FreeExecutorState(estate);
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	index_close(indexRel, RowExclusiveLock);
+
+	return PointerGetDatum(NULL);
+}
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/copy.c,v 1.315 2009/07/25 17:04:19 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/copy.c,v 1.316 2009/07/29 20:56:18 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -2130,6 +2130,8 @@ CopyFrom(CopyState cstate)

 		if (!skip_tuple)
 		{
+			List *recheckIndexes = NIL;
+
 			/* Place tuple in tuple slot */
 			ExecStoreTuple(tuple, slot, InvalidBuffer, false);

@@ -2141,10 +2143,12 @@ CopyFrom(CopyState cstate)
 			heap_insert(cstate->rel, tuple, mycid, hi_options, bistate);

 			if (resultRelInfo->ri_NumIndices > 0)
-				ExecInsertIndexTuples(slot, &(tuple->t_self), estate, false);
+				recheckIndexes = ExecInsertIndexTuples(slot, &(tuple->t_self),
+													   estate, false);

 			/* AFTER ROW INSERT Triggers */
-			ExecARInsertTriggers(estate, resultRelInfo, tuple);
+			ExecARInsertTriggers(estate, resultRelInfo, tuple,
+								 recheckIndexes);

 			/*
 			 * We count only tuples not suppressed by a BEFORE INSERT trigger;
--- 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.186 2009/07/16 06:33:42 petere Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/indexcmds.c,v 1.187 2009/07/29 20:56:18 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -87,6 +87,8 @@ static bool relationHasPrimaryKey(Relation rel);
 * 'primary': mark the index as a primary key in the catalogs.
 * 'isconstraint': index is for a PRIMARY KEY or UNIQUE constraint,
 *		so build a pg_constraint entry for it.
+ * 'deferrable': constraint is DEFERRABLE.
+ * 'initdeferred': constraint is INITIALLY DEFERRED.
 * 'is_alter_table': this is due to an ALTER rather than a CREATE operation.
 * 'check_rights': check for CREATE rights in the namespace.  (This should
 *		be true except when ALTER is deleting/recreating an index.)
@@ -107,6 +109,8 @@ DefineIndex(RangeVar *heapRelation,
 			bool unique,
 			bool primary,
 			bool isconstraint,
+			bool deferrable,
+			bool initdeferred,
 			bool is_alter_table,
 			bool check_rights,
 			bool skip_build,
@@ -447,7 +451,8 @@ DefineIndex(RangeVar *heapRelation,
 		indexRelationId =
 			index_create(relationId, indexRelationName, indexRelationId,
 					  indexInfo, accessMethodId, tablespaceId, classObjectId,
-						 coloptions, reloptions, primary, isconstraint,
+						 coloptions, reloptions, primary,
+						 isconstraint, deferrable, initdeferred,
 						 allowSystemTableMods, skip_build, concurrent);

 		return;					/* We're done, in the standard case */
@@ -465,7 +470,8 @@ DefineIndex(RangeVar *heapRelation,
 	indexRelationId =
 		index_create(relationId, indexRelationName, indexRelationId,
 					 indexInfo, accessMethodId, tablespaceId, classObjectId,
-					 coloptions, reloptions, primary, isconstraint,
+					 coloptions, reloptions, primary,
+					 isconstraint, deferrable, initdeferred,
 					 allowSystemTableMods, true, concurrent);

 	/*
--- 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.292 2009/07/28 02:56:30 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.293 2009/07/29 20:56:18 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -4391,6 +4391,8 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 				stmt->unique,
 				stmt->primary,
 				stmt->isconstraint,
+				stmt->deferrable,
+				stmt->initdeferred,
 				true,			/* is_alter_table */
 				check_rights,
 				skip_build,
@@ -4955,6 +4957,17 @@ transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
 		indexStruct = (Form_pg_index) GETSTRUCT(indexTuple);
 		if (indexStruct->indisprimary)
 		{
+			/*
+			 * Refuse to use a deferrable primary key.  This is per SQL spec,
+			 * and there would be a lot of interesting semantic problems if
+			 * we tried to allow it.
+			 */
+			if (!indexStruct->indimmediate)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						 errmsg("cannot use a deferrable primary key for referenced table \"%s\"",
+								RelationGetRelationName(pkrel))));
+
 			*indexOid = indexoid;
 			break;
 		}
@@ -5040,11 +5053,12 @@ transformFkeyCheckAttrs(Relation pkrel,
 		indexStruct = (Form_pg_index) GETSTRUCT(indexTuple);

 		/*
-		 * Must have the right number of columns; must be unique and not a
-		 * partial index; forget it if there are any expressions, too
+		 * Must have the right number of columns; must be unique (non
+		 * deferrable) and not a partial index; forget it if there are any
+		 * expressions, too
 		 */
 		if (indexStruct->indnatts == numattrs &&
-			indexStruct->indisunique &&
+			indexStruct->indisunique && indexStruct->indimmediate &&
 			heap_attisnull(indexTuple, Anum_pg_index_indpred) &&
 			heap_attisnull(indexTuple, Anum_pg_index_indexprs))
 		{
@@ -5243,7 +5257,8 @@ CreateFKCheckTrigger(RangeVar *myRel, FkConstraint *fkconstraint,
 	fk_trigger->constrrel = fkconstraint->pktable;
 	fk_trigger->args = NIL;

-	(void) CreateTrigger(fk_trigger, constraintOid, indexOid, false);
+	(void) CreateTrigger(fk_trigger, constraintOid, indexOid,
+						 "RI_ConstraintTrigger", false);

 	/* Make changes-so-far visible */
 	CommandCounterIncrement();
@@ -5322,7 +5337,8 @@ createForeignKeyTriggers(Relation rel, FkConstraint *fkconstraint,
 	}
 	fk_trigger->args = NIL;

-	(void) CreateTrigger(fk_trigger, constraintOid, indexOid, false);
+	(void) CreateTrigger(fk_trigger, constraintOid, indexOid,
+						 "RI_ConstraintTrigger", false);

 	/* Make changes-so-far visible */
 	CommandCounterIncrement();
@@ -5373,7 +5389,8 @@ createForeignKeyTriggers(Relation rel, FkConstraint *fkconstraint,
 	}
 	fk_trigger->args = NIL;

-	(void) CreateTrigger(fk_trigger, constraintOid, indexOid, false);
+	(void) CreateTrigger(fk_trigger, constraintOid, indexOid,
+						 "RI_ConstraintTrigger", false);
 }

 /*
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -7,7 +7,7 @@
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.249 2009/07/28 02:56:30 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.250 2009/07/29 20:56:18 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -63,7 +63,8 @@ static HeapTuple ExecCallTriggerFunc(TriggerData *trigdata,
 					Instrumentation *instr,
 					MemoryContext per_tuple_context);
 static void AfterTriggerSaveEvent(ResultRelInfo *relinfo, int event,
-					  bool row_trigger, HeapTuple oldtup, HeapTuple newtup);
+					  bool row_trigger, HeapTuple oldtup, HeapTuple newtup,
+					  List *recheckIndexes);


 /*
@@ -77,6 +78,10 @@ static void AfterTriggerSaveEvent(ResultRelInfo *relinfo, int event,
 * indexOid, if nonzero, is the OID of an index associated with the constraint.
 * We do nothing with this except store it into pg_trigger.tgconstrindid.
 *
+ * prefix is NULL for user-created triggers.  For internally generated
+ * constraint triggers, it is a prefix string to use in building the
+ * trigger name.  (stmt->trigname is the constraint name in such cases.)
+ *
 * If checkPermissions is true we require ACL_TRIGGER permissions on the
 * relation.  If not, the caller already checked permissions.  (This is
 * currently redundant with constraintOid being zero, but it's clearer to
@@ -87,7 +92,7 @@ static void AfterTriggerSaveEvent(ResultRelInfo *relinfo, int event,
 */
 Oid
 CreateTrigger(CreateTrigStmt *stmt,
-			  Oid constraintOid, Oid indexOid,
+			  Oid constraintOid, Oid indexOid, const char *prefix,
 			  bool checkPermissions)
 {
 	int16		tgtype;
@@ -216,20 +221,20 @@ CreateTrigger(CreateTrigStmt *stmt,
 	trigoid = GetNewOid(tgrel);

 	/*
-	 * If trigger is for an RI constraint, the passed-in name is the
-	 * constraint name; save that and build a unique trigger name to avoid
-	 * collisions with user-selected trigger names.
+	 * If trigger is for a constraint, stmt->trigname is the constraint
+	 * name; save that and build a unique trigger name based on the supplied
+	 * prefix, to avoid collisions with user-selected trigger names.
 	 */
-	if (OidIsValid(constraintOid))
+	if (prefix != NULL)
 	{
 		snprintf(constrtrigname, sizeof(constrtrigname),
-				 "RI_ConstraintTrigger_%u", trigoid);
+				 "%s_%u", prefix, trigoid);
 		trigname = constrtrigname;
 		constrname = stmt->trigname;
 	}
 	else if (stmt->isconstraint)
 	{
-		/* constraint trigger: trigger name is also constraint name */
+		/* user constraint trigger: trigger name is also constraint name */
 		trigname = stmt->trigname;
 		constrname = stmt->trigname;
 	}
@@ -1650,7 +1655,7 @@ ExecASInsertTriggers(EState *estate, ResultRelInfo *relinfo)

 	if (trigdesc && trigdesc->n_after_statement[TRIGGER_EVENT_INSERT] > 0)
 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_INSERT,
-							  false, NULL, NULL);
+							  false, NULL, NULL, NIL);
 }

 HeapTuple
@@ -1706,13 +1711,13 @@ ExecBRInsertTriggers(EState *estate, ResultRelInfo *relinfo,

 void
 ExecARInsertTriggers(EState *estate, ResultRelInfo *relinfo,
-					 HeapTuple trigtuple)
+					 HeapTuple trigtuple, List *recheckIndexes)
 {
 	TriggerDesc *trigdesc = relinfo->ri_TrigDesc;

 	if (trigdesc && trigdesc->n_after_row[TRIGGER_EVENT_INSERT] > 0)
 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_INSERT,
-							  true, NULL, trigtuple);
+							  true, NULL, trigtuple, recheckIndexes);
 }

 void
@@ -1781,7 +1786,7 @@ ExecASDeleteTriggers(EState *estate, ResultRelInfo *relinfo)

 	if (trigdesc && trigdesc->n_after_statement[TRIGGER_EVENT_DELETE] > 0)
 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_DELETE,
-							  false, NULL, NULL);
+							  false, NULL, NULL, NIL);
 }

 bool
@@ -1858,7 +1863,7 @@ ExecARDeleteTriggers(EState *estate, ResultRelInfo *relinfo,
 												   tupleid, NULL);

 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_DELETE,
-							  true, trigtuple, NULL);
+							  true, trigtuple, NULL, NIL);
 		heap_freetuple(trigtuple);
 	}
 }
@@ -1929,7 +1934,7 @@ ExecASUpdateTriggers(EState *estate, ResultRelInfo *relinfo)

 	if (trigdesc && trigdesc->n_after_statement[TRIGGER_EVENT_UPDATE] > 0)
 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_UPDATE,
-							  false, NULL, NULL);
+							  false, NULL, NULL, NIL);
 }

 HeapTuple
@@ -1999,7 +2004,8 @@ ExecBRUpdateTriggers(EState *estate, ResultRelInfo *relinfo,

 void
 ExecARUpdateTriggers(EState *estate, ResultRelInfo *relinfo,
-					 ItemPointer tupleid, HeapTuple newtuple)
+					 ItemPointer tupleid, HeapTuple newtuple,
+					 List *recheckIndexes)
 {
 	TriggerDesc *trigdesc = relinfo->ri_TrigDesc;

@@ -2009,7 +2015,7 @@ ExecARUpdateTriggers(EState *estate, ResultRelInfo *relinfo,
 												   tupleid, NULL);

 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_UPDATE,
-							  true, trigtuple, newtuple);
+							  true, trigtuple, newtuple, recheckIndexes);
 		heap_freetuple(trigtuple);
 	}
 }
@@ -2080,7 +2086,7 @@ ExecASTruncateTriggers(EState *estate, ResultRelInfo *relinfo)

 	if (trigdesc && trigdesc->n_after_statement[TRIGGER_EVENT_TRUNCATE] > 0)
 		AfterTriggerSaveEvent(relinfo, TRIGGER_EVENT_TRUNCATE,
-							  false, NULL, NULL);
+							  false, NULL, NULL, NIL);
 }


@@ -3793,15 +3799,18 @@ AfterTriggerPendingOnRel(Oid relid)
 /* ----------
 * AfterTriggerSaveEvent()
 *
- *	Called by ExecA[RS]...Triggers() to add the event to the queue.
+ *	Called by ExecA[RS]...Triggers() to queue up the triggers that should
+ *	be fired for an event.
 *
- *	NOTE: should be called only if we've determined that an event must
- *	be added to the queue.
+ *	NOTE: this is called whenever there are any triggers associated with
+ *	the event (even if they are disabled).  This function decides which
+ *	triggers actually need to be queued.
 * ----------
 */
 static void
 AfterTriggerSaveEvent(ResultRelInfo *relinfo, int event, bool row_trigger,
-					  HeapTuple oldtup, HeapTuple newtup)
+					  HeapTuple oldtup, HeapTuple newtup,
+					  List *recheckIndexes)
 {
 	Relation	rel = relinfo->ri_RelationDesc;
 	TriggerDesc *trigdesc = relinfo->ri_TrigDesc;
@@ -3961,6 +3970,17 @@ AfterTriggerSaveEvent(ResultRelInfo *relinfo, int event, bool row_trigger,
 			}
 		}

+		/*
+		 * If the trigger is a deferred unique constraint check trigger,
+		 * only queue it if the unique constraint was potentially violated,
+		 * which we know from index insertion time.
+		 */
+		if (trigger->tgfoid == F_UNIQUE_KEY_RECHECK)
+		{
+			if (!list_member_oid(recheckIndexes, trigger->tgconstrindid))
+				continue;		/* Uniqueness definitely not violated */
+		}
+
 		/*
 		 * Fill in event structure and add it to the current query's queue.
 		 */