ri_triggers.c

来自「postgresql8.3.4源码,开源数据库」· C语言 代码 · 共 2,476 行 · 第 1/5 页

/* ----------
 * ri_triggers.c
 *
 *	Generic trigger procedures for referential integrity constraint
 *	checks.
 *
 *	Note about memory management: the private hashtables kept here live
 *	across query and transaction boundaries, in fact they live as long as
 *	the backend does.  This works because the hashtable structures
 *	themselves are allocated by dynahash.c in its permanent DynaHashCxt,
 *	and the SPI plans they point to are saved using SPI_saveplan().
 *	There is not currently any provision for throwing away a no-longer-needed
 *	plan --- consider improving this someday.
 *
 *
 * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
 *
 * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.103.2.3 2008/09/15 23:37:49 tgl Exp $
 *
 * ----------
 */


/* ----------
 * Internal TODO:
 *
 *		Add MATCH PARTIAL logic.
 * ----------
 */

#include "postgres.h"

#include "catalog/pg_constraint.h"
#include "catalog/pg_operator.h"
#include "commands/trigger.h"
#include "executor/spi.h"
#include "parser/parse_coerce.h"
#include "parser/parse_relation.h"
#include "miscadmin.h"
#include "utils/acl.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"


/* ----------
 * Local definitions
 * ----------
 */

#define RI_MAX_NUMKEYS					INDEX_MAX_KEYS

#define RI_INIT_QUERYHASHSIZE			128

#define RI_KEYS_ALL_NULL				0
#define RI_KEYS_SOME_NULL				1
#define RI_KEYS_NONE_NULL				2

/* queryno values must be distinct for the convenience of ri_PerformCheck */
#define RI_PLAN_CHECK_LOOKUPPK_NOCOLS	1
#define RI_PLAN_CHECK_LOOKUPPK			2
#define RI_PLAN_CASCADE_DEL_DODELETE	3
#define RI_PLAN_CASCADE_UPD_DOUPDATE	4
#define RI_PLAN_NOACTION_DEL_CHECKREF	5
#define RI_PLAN_NOACTION_UPD_CHECKREF	6
#define RI_PLAN_RESTRICT_DEL_CHECKREF	7
#define RI_PLAN_RESTRICT_UPD_CHECKREF	8
#define RI_PLAN_SETNULL_DEL_DOUPDATE	9
#define RI_PLAN_SETNULL_UPD_DOUPDATE	10

#define MAX_QUOTED_NAME_LEN  (NAMEDATALEN*2+3)
#define MAX_QUOTED_REL_NAME_LEN  (MAX_QUOTED_NAME_LEN*2)

#define RIAttName(rel, attnum)	NameStr(*attnumAttName(rel, attnum))
#define RIAttType(rel, attnum)	attnumTypeId(rel, attnum)

#define RI_TRIGTYPE_INSERT 1
#define RI_TRIGTYPE_UPDATE 2
#define RI_TRIGTYPE_INUP   3
#define RI_TRIGTYPE_DELETE 4

#define RI_KEYPAIR_FK_IDX	0
#define RI_KEYPAIR_PK_IDX	1


/* ----------
 * RI_ConstraintInfo
 *
 *	Information extracted from an FK pg_constraint entry.
 * ----------
 */
typedef struct RI_ConstraintInfo
{
	Oid			constraint_id;	/* OID of pg_constraint entry */
	NameData	conname;		/* name of the FK constraint */
	Oid			pk_relid;		/* referenced relation */
	Oid			fk_relid;		/* referencing relation */
	char		confupdtype;	/* foreign key's ON UPDATE action */
	char		confdeltype;	/* foreign key's ON DELETE action */
	char		confmatchtype;	/* foreign key's match type */
	int			nkeys;			/* number of key columns */
	int16		pk_attnums[RI_MAX_NUMKEYS];		/* attnums of referenced cols */
	int16		fk_attnums[RI_MAX_NUMKEYS];		/* attnums of referencing cols */
	Oid			pf_eq_oprs[RI_MAX_NUMKEYS];		/* equality operators (PK =
												 * FK) */
	Oid			pp_eq_oprs[RI_MAX_NUMKEYS];		/* equality operators (PK =
												 * PK) */
	Oid			ff_eq_oprs[RI_MAX_NUMKEYS];		/* equality operators (FK =
												 * FK) */
} RI_ConstraintInfo;


/* ----------
 * RI_QueryKey
 *
 *	The key identifying a prepared SPI plan in our query hashtable
 * ----------
 */
typedef struct RI_QueryKey
{
	char		constr_type;
	Oid			constr_id;
	int32		constr_queryno;
	Oid			fk_relid;
	Oid			pk_relid;
	int32		nkeypairs;
	int16		keypair[RI_MAX_NUMKEYS][2];
} RI_QueryKey;


/* ----------
 * RI_QueryHashEntry
 * ----------
 */
typedef struct RI_QueryHashEntry
{
	RI_QueryKey key;
	SPIPlanPtr	plan;
} RI_QueryHashEntry;


/* ----------
 * RI_CompareKey
 *
 *	The key identifying an entry showing how to compare two values
 * ----------
 */
typedef struct RI_CompareKey
{
	Oid			eq_opr;			/* the equality operator to apply */
	Oid			typeid;			/* the data type to apply it to */
} RI_CompareKey;


/* ----------
 * RI_CompareHashEntry
 * ----------
 */
typedef struct RI_CompareHashEntry
{
	RI_CompareKey key;
	bool		valid;			/* successfully initialized? */
	FmgrInfo	eq_opr_finfo;	/* call info for equality fn */
	FmgrInfo	cast_func_finfo;	/* in case we must coerce input */
} RI_CompareHashEntry;


/* ----------
 * Local data
 * ----------
 */
static HTAB *ri_query_cache = NULL;
static HTAB *ri_compare_cache = NULL;


/* ----------
 * Local function prototypes
 * ----------
 */
static void quoteOneName(char *buffer, const char *name);
static void quoteRelationName(char *buffer, Relation rel);
static void ri_GenerateQual(StringInfo buf,
				const char *sep,
				const char *leftop, Oid leftoptype,
				Oid opoid,
				const char *rightop, Oid rightoptype);
static void ri_add_cast_to(StringInfo buf, Oid typid);
static int ri_NullCheck(Relation rel, HeapTuple tup,
			 RI_QueryKey *key, int pairidx);
static void ri_BuildQueryKeyFull(RI_QueryKey *key,
					 const RI_ConstraintInfo *riinfo,
					 int32 constr_queryno);
static void ri_BuildQueryKeyPkCheck(RI_QueryKey *key,
						const RI_ConstraintInfo *riinfo,
						int32 constr_queryno);
static bool ri_KeysEqual(Relation rel, HeapTuple oldtup, HeapTuple newtup,
			 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
static bool ri_AllKeysUnequal(Relation rel, HeapTuple oldtup, HeapTuple newtup,
				  const RI_ConstraintInfo *riinfo, bool rel_is_pk);
static bool ri_OneKeyEqual(Relation rel, int column,
			   HeapTuple oldtup, HeapTuple newtup,
			   const RI_ConstraintInfo *riinfo, bool rel_is_pk);
static bool ri_AttributesEqual(Oid eq_opr, Oid typeid,
				   Datum oldvalue, Datum newvalue);
static bool ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
				  HeapTuple old_row,
				  const RI_ConstraintInfo *riinfo);

static void ri_InitHashTables(void);
static SPIPlanPtr ri_FetchPreparedPlan(RI_QueryKey *key);
static void ri_HashPreparedPlan(RI_QueryKey *key, SPIPlanPtr plan);
static RI_CompareHashEntry *ri_HashCompareOp(Oid eq_opr, Oid typeid);

static void ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname,
				int tgkind);
static void ri_FetchConstraintInfo(RI_ConstraintInfo *riinfo,
					   Trigger *trigger, Relation trig_rel, bool rel_is_pk);
static SPIPlanPtr ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
			 RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel,
			 bool cache_plan);
static bool ri_PerformCheck(RI_QueryKey *qkey, SPIPlanPtr qplan,
				Relation fk_rel, Relation pk_rel,
				HeapTuple old_tuple, HeapTuple new_tuple,
				bool detectNewRows,
				int expect_OK, const char *constrname);
static void ri_ExtractValues(RI_QueryKey *qkey, int key_idx,
				 Relation rel, HeapTuple tuple,
				 Datum *vals, char *nulls);
static void ri_ReportViolation(RI_QueryKey *qkey, const char *constrname,
				   Relation pk_rel, Relation fk_rel,
				   HeapTuple violator, TupleDesc tupdesc,
				   bool spi_err);


/* ----------
 * RI_FKey_check -
 *
 *	Check foreign key existence (combined for INSERT and UPDATE).
 * ----------
 */
static Datum
RI_FKey_check(PG_FUNCTION_ARGS)
{
	TriggerData *trigdata = (TriggerData *) fcinfo->context;
	RI_ConstraintInfo riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	HeapTuple	new_row;
	HeapTuple	old_row;
	Buffer		new_row_buf;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;
	int			i;

	/*
	 * Check that this is a valid trigger call on the right time and event.
	 */
	ri_CheckTrigger(fcinfo, "RI_FKey_check", RI_TRIGTYPE_INUP);

	/*
	 * Get arguments.
	 */
	ri_FetchConstraintInfo(&riinfo,
						 trigdata->tg_trigger, trigdata->tg_relation, false);

	if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
	{
		old_row = trigdata->tg_trigtuple;
		new_row = trigdata->tg_newtuple;
		new_row_buf = trigdata->tg_newtuplebuf;
	}
	else
	{
		old_row = NULL;
		new_row = trigdata->tg_trigtuple;
		new_row_buf = trigdata->tg_trigtuplebuf;
	}

	/*
	 * We should not even consider checking the row if it is no longer valid,
	 * since it was either deleted (so the deferred check should be skipped)
	 * or updated (in which case only the latest version of the row should be
	 * checked).  Test its liveness according to SnapshotSelf.
	 *
	 * NOTE: The normal coding rule is that one must acquire the buffer
	 * content lock to call HeapTupleSatisfiesVisibility.  We can skip that
	 * here because we know that AfterTriggerExecute just fetched the tuple
	 * successfully, so there cannot be a VACUUM compaction in progress on the
	 * page (either heap_fetch would have waited for the VACUUM, or the
	 * VACUUM's LockBufferForCleanup would be waiting for us to drop pin). And
	 * since this is a row inserted by our open transaction, no one else can
	 * be entitled to change its xmin/xmax.
	 */
	Assert(new_row_buf != InvalidBuffer);
	if (!HeapTupleSatisfiesVisibility(new_row, SnapshotSelf, new_row_buf))
		return PointerGetDatum(NULL);

	/*
	 * Get the relation descriptors of the FK and PK tables.
	 *
	 * pk_rel is opened in RowShareLock mode since that's what our eventual
	 * SELECT FOR SHARE will get on it.
	 */
	fk_rel = trigdata->tg_relation;
	pk_rel = heap_open(riinfo.pk_relid, RowShareLock);

	/* ----------
	 * SQL3 11.9 <referential constraint definition>
	 *	General rules 2) a):
	 *		If Rf and Rt are empty (no columns to compare given)
	 *		constraint is true if 0 < (SELECT COUNT(*) FROM T)
	 *
	 *	Note: The special case that no columns are given cannot
	 *		occur up to now in Postgres, it's just there for
	 *		future enhancements.
	 * ----------
	 */
	if (riinfo.nkeys == 0)
	{
		ri_BuildQueryKeyFull(&qkey, &riinfo, RI_PLAN_CHECK_LOOKUPPK_NOCOLS);

		if (SPI_connect() != SPI_OK_CONNECT)
			elog(ERROR, "SPI_connect failed");

		if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
		{
			char		querystr[MAX_QUOTED_REL_NAME_LEN + 100];
			char		pkrelname[MAX_QUOTED_REL_NAME_LEN];

			/* ---------
			 * The query string built is
			 *	SELECT 1 FROM ONLY <pktable>
			 * ----------
			 */
			quoteRelationName(pkrelname, pk_rel);
			snprintf(querystr, sizeof(querystr),
					 "SELECT 1 FROM ONLY %s x FOR SHARE OF x",
					 pkrelname);

			/* Prepare and save the plan */
			qplan = ri_PlanCheck(querystr, 0, NULL,
								 &qkey, fk_rel, pk_rel, true);
		}

		/*
		 * Execute the plan
		 */
		ri_PerformCheck(&qkey, qplan,
						fk_rel, pk_rel,
						NULL, NULL,
						false,
						SPI_OK_SELECT,
						NameStr(riinfo.conname));

		if (SPI_finish() != SPI_OK_FINISH)
			elog(ERROR, "SPI_finish failed");

		heap_close(pk_rel, RowShareLock);

		return PointerGetDatum(NULL);
	}

	if (riinfo.confmatchtype == FKCONSTR_MATCH_PARTIAL)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("MATCH PARTIAL not yet implemented")));

	ri_BuildQueryKeyFull(&qkey, &riinfo, RI_PLAN_CHECK_LOOKUPPK);

	switch (ri_NullCheck(fk_rel, new_row, &qkey, RI_KEYPAIR_FK_IDX))
	{
		case RI_KEYS_ALL_NULL:

			/*
			 * No check - if NULLs are allowed at all is already checked by
			 * NOT NULL constraint.
			 *
			 * This is true for MATCH FULL, MATCH PARTIAL, and MATCH
			 * <unspecified>
			 */
			heap_close(pk_rel, RowShareLock);
			return PointerGetDatum(NULL);

		case RI_KEYS_SOME_NULL:

			/*
			 * This is the only case that differs between the three kinds of
			 * MATCH.
			 */
			switch (riinfo.confmatchtype)
			{
				case FKCONSTR_MATCH_FULL:

					/*
					 * Not allowed - MATCH FULL says either all or none of the
					 * attributes can be NULLs
					 */
					ereport(ERROR,
							(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
							 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
							  RelationGetRelationName(trigdata->tg_relation),
									NameStr(riinfo.conname)),
							 errdetail("MATCH FULL does not allow mixing of null and nonnull key values.")));
					heap_close(pk_rel, RowShareLock);
					return PointerGetDatum(NULL);

				case FKCONSTR_MATCH_UNSPECIFIED:

					/*
					 * MATCH <unspecified> - if ANY column is null, we have a
					 * match.
					 */
					heap_close(pk_rel, RowShareLock);
					return PointerGetDatum(NULL);

				case FKCONSTR_MATCH_PARTIAL:

					/*
					 * MATCH PARTIAL - all non-null columns must match. (not
					 * implemented, can be done by modifying the query below
					 * to only include non-null columns, or by writing a
					 * special version here)
					 */
					ereport(ERROR,
							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
							 errmsg("MATCH PARTIAL not yet implemented")));
					heap_close(pk_rel, RowShareLock);
					return PointerGetDatum(NULL);
			}

		case RI_KEYS_NONE_NULL:

			/*
			 * Have a full qualified key - continue below for all three kinds
			 * of MATCH.
			 */
			break;
	}

	if (SPI_connect() != SPI_OK_CONNECT)
		elog(ERROR, "SPI_connect failed");

	/*
	 * Fetch or prepare a saved plan for the real check
	 */
	if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
	{
		StringInfoData querybuf;
		char		pkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		attname[MAX_QUOTED_NAME_LEN];
		char		paramname[16];
		const char *querysep;
		Oid			queryoids[RI_MAX_NUMKEYS];

		/* ----------
		 * The query string built is
		 *	SELECT 1 FROM ONLY <pktable> WHERE pkatt1 = $1 [AND ...] FOR SHARE
		 * The type id's for the $ parameters are those of the
		 * corresponding FK attributes.
		 * ----------
		 */
		initStringInfo(&querybuf);
		quoteRelationName(pkrelname, pk_rel);
		appendStringInfo(&querybuf, "SELECT 1 FROM ONLY %s x", pkrelname);
		querysep = "WHERE";
		for (i = 0; i < riinfo.nkeys; i++)
		{
			Oid			pk_type = RIAttType(pk_rel, riinfo.pk_attnums[i]);
			Oid			fk_type = RIAttType(fk_rel, riinfo.fk_attnums[i]);

			quoteOneName(attname,
						 RIAttName(pk_rel, riinfo.pk_attnums[i]));
			sprintf(paramname, "$%d", i + 1);
			ri_GenerateQual(&querybuf, querysep,
							attname, pk_type,
							riinfo.pf_eq_oprs[i],
							paramname, fk_type);
			querysep = "AND";
			queryoids[i] = fk_type;
		}
		appendStringInfo(&querybuf, " FOR SHARE OF x");

		/* Prepare and save the plan */
		qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
							 &qkey, fk_rel, pk_rel, true);
	}

	/*
	 * Now check that foreign key exists in PK table
	 */
	ri_PerformCheck(&qkey, qplan,
					fk_rel, pk_rel,
					NULL, new_row,
					false,
					SPI_OK_SELECT,
					NameStr(riinfo.conname));