src/backend/utils/adt/ri_triggers.c - master - githubmirror/postgres

src/backend/utils/adt/ri_triggers.c
4 457 строк134 KB
Amit Langote
Handle nullable referenced key in RI fast-path check
07 авг 2026, 11:45
07 авг 2026, 11:450ec3f04
/*-------------------------------------------------------------------------
 *
 * 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_keepplan().
 *	There is not currently any provision for throwing away a no-longer-needed
 *	plan --- consider improving this someday.
 *
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 *
 * src/backend/utils/adt/ri_triggers.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/amapi.h"
#include "access/genam.h"
#include "access/htup_details.h"
#include "access/skey.h"
#include "access/sysattr.h"
#include "access/table.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/index.h"
#include "catalog/pg_am_d.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_namespace.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/spi.h"
#include "lib/ilist.h"
#include "miscadmin.h"
#include "parser/parse_coerce.h"
#include "parser/parse_relation.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/rls.h"
#include "utils/ruleutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

/*
 * Local definitions
 */

#define RI_MAX_NUMKEYS					INDEX_MAX_KEYS

#define RI_INIT_CONSTRAINTHASHSIZE		64
#define RI_INIT_QUERYHASHSIZE			(RI_INIT_CONSTRAINTHASHSIZE * 4)

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

/* RI query type codes */
/* these queries are executed against the PK (referenced) table: */
#define RI_PLAN_CHECK_LOOKUPPK			1
#define RI_PLAN_CHECK_LOOKUPPK_FROM_PK	2
#define RI_PLAN_LAST_ON_PK				RI_PLAN_CHECK_LOOKUPPK_FROM_PK
/* these queries are executed against the FK (referencing) table: */
#define RI_PLAN_CASCADE_ONDELETE		3
#define RI_PLAN_CASCADE_ONUPDATE		4
#define RI_PLAN_NO_ACTION				5
/* For RESTRICT, the same plan can be used for both ON DELETE and ON UPDATE triggers. */
#define RI_PLAN_RESTRICT				6
#define RI_PLAN_SETNULL_ONDELETE		7
#define RI_PLAN_SETNULL_ONUPDATE		8
#define RI_PLAN_SETDEFAULT_ONDELETE		9
#define RI_PLAN_SETDEFAULT_ONUPDATE		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 RIAttCollation(rel, attnum) attnumCollationId(rel, attnum)

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

typedef struct FastPathMeta FastPathMeta;

/*
 * RI_ConstraintInfo
 *
 * Information extracted from an FK pg_constraint entry.  This is cached in
 * ri_constraint_cache.
 *
 * Note that pf/pp/ff_eq_oprs may hold the overlaps operator instead of equals
 * for the PERIOD part of a temporal foreign key.
 */
typedef struct RI_ConstraintInfo
{
	Oid			constraint_id;	/* OID of pg_constraint entry (hash key) */
	bool		valid;			/* successfully initialized? */
	Oid			constraint_root_id; /* OID of topmost ancestor constraint;
									 * same as constraint_id if not inherited */
	uint32		oidHashValue;	/* hash value of constraint_id */
	uint32		rootHashValue;	/* hash value of constraint_root_id */
	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 */
	int			ndelsetcols;	/* number of columns referenced in ON DELETE
								 * SET clause */
	int16		confdelsetcols[RI_MAX_NUMKEYS]; /* attnums of cols to set on
												 * delete */
	char		confmatchtype;	/* foreign key's match type */
	bool		hasperiod;		/* if the foreign key uses PERIOD */
	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) */
	Oid			period_contained_by_oper;	/* anyrange <@ anyrange (or
											 * multiranges) */
	Oid			agged_period_contained_by_oper; /* fkattr <@ range_agg(pkattr) */
	Oid			period_intersect_oper;	/* anyrange * anyrange (or
										 * multiranges) */
	dlist_node	valid_link;		/* Link in list of valid entries */

	Oid			conindid;
	bool		pk_is_partitioned;
	bool		pk_index_is_btree;	/* is conindid a btree index? */

	FastPathMeta *fpmeta;
} RI_ConstraintInfo;

typedef struct RI_CompareHashEntry RI_CompareHashEntry;

/* Fast-path metadata for RI checks on foreign key referencing tables */
typedef struct FastPathMeta
{
	FmgrInfo	eq_opr_finfo[RI_MAX_NUMKEYS];
	FmgrInfo	cast_func_finfo[RI_MAX_NUMKEYS];
	RegProcedure regops[RI_MAX_NUMKEYS];
	Oid			subtypes[RI_MAX_NUMKEYS];
	int			strats[RI_MAX_NUMKEYS];
	AttrNumber	index_attnos[RI_MAX_NUMKEYS];	/* index column positions */
} FastPathMeta;

/*
 * RI_QueryKey
 *
 * The key identifying a prepared SPI plan in our query hashtable
 */
typedef struct RI_QueryKey
{
	Oid			constr_id;		/* OID of pg_constraint entry */
	int32		constr_queryno; /* query type ID, see RI_PLAN_XXX above */
} 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;

/*
 * Maximum number of FK rows buffered before flushing.
 *
 * Larger batches amortize per-flush overhead and let the SK_SEARCHARRAY
 * path walk more leaf pages in a single sorted traversal.  But each
 * buffered row is a materialized HeapTuple in flush_cxt, and the matched[]
 * scan in ri_FastPathFlushArray() is O(batch_size) per index match.
 * Benchmarking showed little difference between 16 and 64, with 256
 * consistently slower.  64 is a reasonable default.
 */
#define RI_FASTPATH_BATCH_SIZE	64

/*
 * RI_FastPathEntry
 *		Per-constraint cache of resources needed by ri_FastPathBatchFlush().
 *
 * One entry per constraint, keyed by pg_constraint OID.  Created lazily
 * by ri_FastPathGetEntry() on first use within a trigger-firing batch
 * and torn down by ri_FastPathTeardown() at batch end.
 *
 * FK tuples are buffered in batch[] across trigger invocations and
 * flushed when the buffer fills or the batch ends.
 *
 * RI_FastPathEntry is not subject to cache invalidation.  The cached
 * relations are held open with locks for the transaction duration, preventing
 * relcache invalidation.  The entry itself is torn down at batch end by
 * ri_FastPathEndBatch(); on abort, ResourceOwner releases the cached
 * relations and AtEOXact_RI() NULLs the static cache pointer to prevent
 * any subsequent access.
 */
typedef struct RI_FastPathEntry
{
	Oid			conoid;			/* hash key: pg_constraint OID */
	Oid			fk_relid;		/* for ri_FastPathEndBatch() */
	Relation	pk_rel;
	Relation	idx_rel;
	TupleTableSlot *pk_slot;
	TupleTableSlot *fk_slot;
	MemoryContext flush_cxt;	/* short-lived context for per-flush work */

	/*
	 * TODO: batch[] is HeapTuple[] because the AFTER trigger machinery
	 * currently passes tuples as HeapTuples.  Once trigger infrastructure is
	 * slotified, this should use a slot array or whatever batched tuple
	 * storage abstraction exists at that point to be TAM-agnostic.
	 */
	HeapTuple	batch[RI_FASTPATH_BATCH_SIZE];
	int			batch_count;

	/*
	 * true while this entry's batch is being flushed; guards against
	 * re-entrant ri_FastPathBatchAdd from user code run during the flush.
	 */
	bool		flushing;
} RI_FastPathEntry;

/*
 * Local data
 */
static HTAB *ri_constraint_cache = NULL;
static HTAB *ri_query_cache = NULL;
static HTAB *ri_compare_cache = NULL;
static dclist_head ri_constraint_cache_valid_list;

static HTAB *ri_fastpath_cache = NULL;
static bool ri_fastpath_callback_registered = false;
static bool ri_fastpath_flushing = false;

/*
 * Local function prototypes
 */
static bool ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
							  TupleTableSlot *oldslot,
							  const RI_ConstraintInfo *riinfo);
static Datum ri_restrict(TriggerData *trigdata, bool is_no_action);
static Datum ri_set(TriggerData *trigdata, bool is_set_null, int tgkind);
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_GenerateQualCollation(StringInfo buf, Oid collation);
static int	ri_NullCheck(TupleDesc tupDesc, TupleTableSlot *slot,
						 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
static void ri_BuildQueryKey(RI_QueryKey *key,
							 const RI_ConstraintInfo *riinfo,
							 int32 constr_queryno);
static bool ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
						 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
static bool ri_CompareWithCast(Oid eq_opr, Oid typeid, Oid collid,
							   Datum lhs, Datum rhs);

static void ri_InitHashTables(void);
static void InvalidateConstraintCacheCallBack(Datum arg, SysCacheIdentifier cacheid,
											  uint32 hashvalue);
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 RI_ConstraintInfo *ri_FetchConstraintInfo(Trigger *trigger,
												 Relation trig_rel, bool rel_is_pk);
static RI_ConstraintInfo *ri_LoadConstraintInfo(Oid constraintOid);
static Oid	get_ri_constraint_root(Oid constrOid);
static SPIPlanPtr ri_PlanCheck(const char *querystr, int nargs, const Oid *argtypes,
							   RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel);
static bool ri_PerformCheck(const RI_ConstraintInfo *riinfo,
							RI_QueryKey *qkey, SPIPlanPtr qplan,
							Relation fk_rel, Relation pk_rel,
							TupleTableSlot *oldslot, TupleTableSlot *newslot,
							bool is_restrict,
							bool detectNewRows, int expect_OK);
static void ri_FastPathCheck(RI_ConstraintInfo *riinfo,
							 Relation fk_rel, TupleTableSlot *newslot);
static void ri_FastPathBatchAdd(RI_ConstraintInfo *riinfo,
								Relation fk_rel, TupleTableSlot *newslot);
static void ri_FastPathBatchFlush(RI_FastPathEntry *fpentry, Relation fk_rel,
								  RI_ConstraintInfo *riinfo);
static int	ri_FastPathFlushArray(RI_FastPathEntry *fpentry, TupleTableSlot *fk_slot,
								  const RI_ConstraintInfo *riinfo, Relation fk_rel,
								  Snapshot snapshot, IndexScanDesc scandesc);
static int	ri_FastPathFlushLoop(RI_FastPathEntry *fpentry, TupleTableSlot *fk_slot,
								 const RI_ConstraintInfo *riinfo, Relation fk_rel,
								 Snapshot snapshot, IndexScanDesc scandesc);
static bool ri_FastPathProbeOne(Relation pk_rel, Relation idx_rel,
								IndexScanDesc scandesc, TupleTableSlot *slot,
								Snapshot snapshot, const RI_ConstraintInfo *riinfo,
								ScanKeyData *skey, int nkeys);
static bool ri_LockPKTuple(Relation pk_rel, TupleTableSlot *slot, Snapshot snap,
						   bool *concurrently_updated);
static bool ri_fastpath_is_applicable(const RI_ConstraintInfo *riinfo);
static void ri_CheckPermissions(Relation query_rel);
static bool recheck_matched_pk_tuple(Relation idxrel, ScanKeyData *skeys,
									 int nkeys, TupleTableSlot *new_slot);
static void build_index_scankeys(const RI_ConstraintInfo *riinfo,
								 Relation idx_rel, Datum *pk_vals,
								 char *pk_nulls, ScanKey skeys);
static void ri_populate_fastpath_metadata(RI_ConstraintInfo *riinfo,
										  Relation fk_rel, Relation idx_rel);
static void ri_ExtractValues(Relation rel, TupleTableSlot *slot,
							 const RI_ConstraintInfo *riinfo, bool rel_is_pk,
							 Datum *vals, char *nulls);
pg_noreturn static void ri_ReportViolation(const RI_ConstraintInfo *riinfo,
										   Relation pk_rel, Relation fk_rel,
										   TupleTableSlot *violatorslot, TupleDesc tupdesc,
										   int queryno, bool is_restrict, bool partgone);
static RI_FastPathEntry *ri_FastPathGetEntry(const RI_ConstraintInfo *riinfo,
											 Relation fk_rel);
static void ri_FastPathEndBatch(void *arg);
static void ri_FastPathTeardown(void);


/*
 * RI_FKey_check -
 *
 * Check foreign key existence (combined for INSERT and UPDATE).
 */
static Datum
RI_FKey_check(TriggerData *trigdata)
{
	RI_ConstraintInfo *riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	TupleTableSlot *newslot;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;

	riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
									trigdata->tg_relation, false);

	if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
		newslot = trigdata->tg_newslot;
	else
		newslot = trigdata->tg_trigslot;

	/*
	 * 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.  We need pin
	 * and lock on the buffer to call HeapTupleSatisfiesVisibility.  Caller
	 * should be holding pin, but not lock.
	 */
	if (!table_tuple_satisfies_snapshot(trigdata->tg_relation, newslot, SnapshotSelf))
		return PointerGetDatum(NULL);

	fk_rel = trigdata->tg_relation;

	switch (ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false))
	{
		case RI_KEYS_ALL_NULL:

			/*
			 * No further check needed - an all-NULL key passes every type of
			 * foreign key constraint.
			 */
			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(fk_rel),
									NameStr(riinfo->conname)),
							 errdetail("MATCH FULL does not allow mixing of null and nonnull key values."),
							 errtableconstraint(fk_rel,
												NameStr(riinfo->conname))));
					return PointerGetDatum(NULL);

				case FKCONSTR_MATCH_SIMPLE:

					/*
					 * MATCH SIMPLE - if ANY column is null, the key passes
					 * the constraint.
					 */
					return PointerGetDatum(NULL);

#ifdef NOT_USED
				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)
					 */
					break;
#endif
			}

		case RI_KEYS_NONE_NULL:

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

	/*
	 * Fast path: probe the PK unique index directly, bypassing SPI.
	 *
	 * For non-partitioned, non-temporal FKs, we can skip the SPI machinery
	 * (plan cache, executor setup, etc.) and do a direct index scan + tuple
	 * lock.  This is semantically equivalent to the SPI path below but avoids
	 * the per-row executor overhead.
	 *
	 * ri_FastPathBatchAdd() and ri_FastPathCheck() report the violation
	 * themselves if no matching PK row is found, so they only return on
	 * success.
	 */
	if (ri_fastpath_is_applicable(riinfo))
	{
		if (AfterTriggerIsActive() &&
			GetCurrentTransactionNestLevel() == 1 &&
			!ri_fastpath_flushing)
		{
			/* Batched path: buffer and probe in groups */
			ri_FastPathBatchAdd(riinfo, fk_rel, newslot);
		}
		else
		{
			/*
			 * Per-row path, used when batching is not safe or not applicable:
			 *
			 * - ALTER TABLE validation, where no after-trigger firing is
			 * active;
			 *
			 * - any FK check inside a subtransaction, since the batch cache
			 * is confined to the top transaction level (it cannot be cleanly
			 * unwound on subxact abort);
			 *
			 * - a re-entrant check from user cast/operator code running
			 * during a batch flush, since adding a cache entry while
			 * ri_FastPathEndBatch is iterating the cache could leave it
			 * unflushed.
			 */
			ri_FastPathCheck(riinfo, fk_rel, newslot);
		}
		return PointerGetDatum(NULL);
	}

	SPI_connect();

	/*
	 * pk_rel is opened in RowShareLock mode since that's what our eventual
	 * SELECT FOR KEY SHARE will get on it.
	 */
	pk_rel = table_open(riinfo->pk_relid, RowShareLock);

	/* Fetch or prepare a saved plan for the real check */
	ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK);

	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];
		const char *pk_only;

		/* ----------
		 * The query string built is
		 *	SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
		 *		   FOR KEY SHARE OF x
		 * The type id's for the $ parameters are those of the
		 * corresponding FK attributes.
		 *
		 * But for temporal FKs we need to make sure
		 * the FK's range is completely covered.
		 * So we use this query instead:
		 *  SELECT 1
		 *	FROM	(
		 *		SELECT pkperiodatt AS r
		 *		FROM   [ONLY] pktable x
		 *		WHERE  pkatt1 = $1 [AND ...]
		 *		AND    pkperiodatt && $n
		 *		FOR KEY SHARE OF x
		 *	) x1
		 *  HAVING $n <@ range_agg(x1.r)
		 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
		 * we can make this a bit simpler.
		 * ----------
		 */
		initStringInfo(&querybuf);
		pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(pkrelname, pk_rel);
		if (riinfo->hasperiod)
		{
			quoteOneName(attname,
						 RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));

			appendStringInfo(&querybuf,
							 "SELECT 1 FROM (SELECT %s AS r FROM %s%s x",
							 attname, pk_only, pkrelname);
		}
		else
		{
			appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
							 pk_only, pkrelname);
		}
		querysep = "WHERE";
		for (int 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;
		}
		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
		if (riinfo->hasperiod)
		{
			Oid			fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);

			appendStringInfoString(&querybuf, ") x1 HAVING ");
			sprintf(paramname, "$%d", riinfo->nkeys);
			ri_GenerateQual(&querybuf, "",
							paramname, fk_type,
							riinfo->agged_period_contained_by_oper,
							"pg_catalog.range_agg", ANYMULTIRANGEOID);
			appendStringInfoString(&querybuf, "(x1.r)");
		}

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

	/*
	 * Now check that foreign key exists in PK table
	 *
	 * XXX detectNewRows must be true when a partitioned table is on the
	 * referenced side.  The reason is that our snapshot must be fresh in
	 * order for the hack in find_inheritance_children() to work.
	 */
	ri_PerformCheck(riinfo, &qkey, qplan,
					fk_rel, pk_rel,
					NULL, newslot,
					false,
					pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE,
					SPI_OK_SELECT);

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

	table_close(pk_rel, RowShareLock);

	return PointerGetDatum(NULL);
}


/*
 * RI_FKey_check_ins -
 *
 * Check foreign key existence at insert event on FK table.
 */
Datum
RI_FKey_check_ins(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_check_ins", RI_TRIGTYPE_INSERT);

	/* Share code with UPDATE case. */
	return RI_FKey_check((TriggerData *) fcinfo->context);
}


/*
 * RI_FKey_check_upd -
 *
 * Check foreign key existence at update event on FK table.
 */
Datum
RI_FKey_check_upd(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_check_upd", RI_TRIGTYPE_UPDATE);

	/* Share code with INSERT case. */
	return RI_FKey_check((TriggerData *) fcinfo->context);
}


/*
 * ri_Check_Pk_Match
 *
 * Check to see if another PK row has been created that provides the same
 * key values as the "oldslot" that's been modified or deleted in our trigger
 * event.  Returns true if a match is found in the PK table.
 *
 * We assume the caller checked that the oldslot contains no NULL key values,
 * since otherwise a match is impossible.
 */
static bool
ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
				  TupleTableSlot *oldslot,
				  const RI_ConstraintInfo *riinfo)
{
	SPIPlanPtr	qplan;
	RI_QueryKey qkey;
	bool		result;

	/* Only called for non-null rows */
	Assert(ri_NullCheck(RelationGetDescr(pk_rel), oldslot, riinfo, true) == RI_KEYS_NONE_NULL);

	SPI_connect();

	/*
	 * Fetch or prepare a saved plan for checking PK table with values coming
	 * from a PK row
	 */
	ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK_FROM_PK);

	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;
		const char *pk_only;
		Oid			queryoids[RI_MAX_NUMKEYS];

		/* ----------
		 * The query string built is
		 *	SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
		 *		   FOR KEY SHARE OF x
		 * The type id's for the $ parameters are those of the
		 * PK attributes themselves.
		 *
		 * But for temporal FKs we need to make sure
		 * the old PK's range is completely covered.
		 * So we use this query instead:
		 *  SELECT 1
		 *  FROM    (
		 *	  SELECT pkperiodatt AS r
		 *	  FROM   [ONLY] pktable x
		 *	  WHERE  pkatt1 = $1 [AND ...]
		 *	  AND    pkperiodatt && $n
		 *	  FOR KEY SHARE OF x
		 *  ) x1
		 *  HAVING $n <@ range_agg(x1.r)
		 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
		 * we can make this a bit simpler.
		 * ----------
		 */
		initStringInfo(&querybuf);
		pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(pkrelname, pk_rel);
		if (riinfo->hasperiod)
		{
			quoteOneName(attname, RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));

			appendStringInfo(&querybuf,
							 "SELECT 1 FROM (SELECT %s AS r FROM %s%s x",
							 attname, pk_only, pkrelname);
		}
		else
		{
			appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
							 pk_only, pkrelname);
		}
		querysep = "WHERE";
		for (int i = 0; i < riinfo->nkeys; i++)
		{
			Oid			pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);

			quoteOneName(attname,
						 RIAttName(pk_rel, riinfo->pk_attnums[i]));
			sprintf(paramname, "$%d", i + 1);
			ri_GenerateQual(&querybuf, querysep,
							attname, pk_type,
							riinfo->pp_eq_oprs[i],
							paramname, pk_type);
			querysep = "AND";
			queryoids[i] = pk_type;
		}
		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
		if (riinfo->hasperiod)
		{
			Oid			fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);

			appendStringInfoString(&querybuf, ") x1 HAVING ");
			sprintf(paramname, "$%d", riinfo->nkeys);
			ri_GenerateQual(&querybuf, "",
							paramname, fk_type,
							riinfo->agged_period_contained_by_oper,
							"pg_catalog.range_agg", ANYMULTIRANGEOID);
			appendStringInfoString(&querybuf, "(x1.r)");
		}

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

	/*
	 * We have a plan now. Run it.
	 */
	result = ri_PerformCheck(riinfo, &qkey, qplan,
							 fk_rel, pk_rel,
							 oldslot, NULL,
							 false,
							 true,	/* treat like update */
							 SPI_OK_SELECT);

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

	return result;
}


/*
 * RI_FKey_noaction_del -
 *
 * Give an error and roll back the current transaction if the
 * delete has resulted in a violation of the given referential
 * integrity constraint.
 */
Datum
RI_FKey_noaction_del(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_noaction_del", RI_TRIGTYPE_DELETE);

	/* Share code with RESTRICT/UPDATE cases. */
	return ri_restrict((TriggerData *) fcinfo->context, true);
}

/*
 * RI_FKey_restrict_del -
 *
 * Restrict delete from PK table to rows unreferenced by foreign key.
 *
 * The SQL standard intends that this referential action occur exactly when
 * the delete is performed, rather than after.  This appears to be
 * the only difference between "NO ACTION" and "RESTRICT".  In Postgres
 * we still implement this as an AFTER trigger, but it's non-deferrable.
 */
Datum
RI_FKey_restrict_del(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_restrict_del", RI_TRIGTYPE_DELETE);

	/* Share code with NO ACTION/UPDATE cases. */
	return ri_restrict((TriggerData *) fcinfo->context, false);
}

/*
 * RI_FKey_noaction_upd -
 *
 * Give an error and roll back the current transaction if the
 * update has resulted in a violation of the given referential
 * integrity constraint.
 */
Datum
RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_noaction_upd", RI_TRIGTYPE_UPDATE);

	/* Share code with RESTRICT/DELETE cases. */
	return ri_restrict((TriggerData *) fcinfo->context, true);
}

/*
 * RI_FKey_restrict_upd -
 *
 * Restrict update of PK to rows unreferenced by foreign key.
 *
 * The SQL standard intends that this referential action occur exactly when
 * the update is performed, rather than after.  This appears to be
 * the only difference between "NO ACTION" and "RESTRICT".  In Postgres
 * we still implement this as an AFTER trigger, but it's non-deferrable.
 */
Datum
RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_restrict_upd", RI_TRIGTYPE_UPDATE);

	/* Share code with NO ACTION/DELETE cases. */
	return ri_restrict((TriggerData *) fcinfo->context, false);
}

/*
 * ri_restrict -
 *
 * Common code for ON DELETE RESTRICT, ON DELETE NO ACTION,
 * ON UPDATE RESTRICT, and ON UPDATE NO ACTION.
 */
static Datum
ri_restrict(TriggerData *trigdata, bool is_no_action)
{
	const RI_ConstraintInfo *riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	TupleTableSlot *oldslot;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;

	riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
									trigdata->tg_relation, true);

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

	/*
	 * If another PK row now exists providing the old key values, we should
	 * not do anything.  However, this check should only be made in the NO
	 * ACTION case; in RESTRICT cases we don't wish to allow another row to be
	 * substituted.
	 *
	 * If the foreign key has PERIOD, we incorporate looking for replacement
	 * rows in the main SQL query below, so we needn't do it here.
	 */
	if (is_no_action && !riinfo->hasperiod &&
		ri_Check_Pk_Match(pk_rel, fk_rel, oldslot, riinfo))
	{
		table_close(fk_rel, RowShareLock);
		return PointerGetDatum(NULL);
	}

	SPI_connect();

	/*
	 * Fetch or prepare a saved plan for the restrict lookup (it's the same
	 * query for delete and update cases)
	 */
	ri_BuildQueryKey(&qkey, riinfo, is_no_action ? RI_PLAN_NO_ACTION : RI_PLAN_RESTRICT);

	if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
	{
		StringInfoData querybuf;
		char		pkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		attname[MAX_QUOTED_NAME_LEN];
		char		periodattname[MAX_QUOTED_NAME_LEN];
		char		paramname[16];
		const char *querysep;
		Oid			queryoids[RI_MAX_NUMKEYS];
		const char *fk_only;

		/* ----------
		 * The query string built is
		 *	SELECT 1 FROM [ONLY] <fktable> x WHERE $1 = fkatt1 [AND ...]
		 *		   FOR KEY SHARE OF x
		 * The type id's for the $ parameters are those of the
		 * corresponding PK attributes.
		 * ----------
		 */
		initStringInfo(&querybuf);
		fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(fkrelname, fk_rel);
		appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
						 fk_only, fkrelname);
		querysep = "WHERE";
		for (int 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(fk_rel, riinfo->fk_attnums[i]));
			sprintf(paramname, "$%d", i + 1);
			ri_GenerateQual(&querybuf, querysep,
							paramname, pk_type,
							riinfo->pf_eq_oprs[i],
							attname, fk_type);
			querysep = "AND";
			queryoids[i] = pk_type;
		}

		/*----------
		 * For temporal foreign keys, a reference could still be valid if the
		 * referenced range didn't change too much.  Also if a referencing
		 * range extends past the current PK row, we don't want to check that
		 * part: some other PK row should fulfill it.  We only want to check
		 * the part matching the PK record we've changed.  Therefore to find
		 * invalid records we do this:
		 *
		 * SELECT 1 FROM [ONLY] <fktable> x WHERE $1 = x.fkatt1 [AND ...]
		 * -- begin temporal
		 * AND $n && x.fkperiod
		 * AND NOT coalesce((x.fkperiod * $n) <@
		 *  (SELECT range_agg(r)
		 *   FROM (SELECT y.pkperiod r
		 *         FROM [ONLY] <pktable> y
		 *         WHERE $1 = y.pkatt1 [AND ...] AND $n && y.pkperiod
		 *         FOR KEY SHARE OF y) y2), false)
		 * -- end temporal
		 * FOR KEY SHARE OF x
		 *
		 * We need the coalesce in case the first subquery returns no rows.
		 * We need the second subquery because FOR KEY SHARE doesn't support
		 * aggregate queries.
		 */
		if (riinfo->hasperiod && is_no_action)
		{
			Oid			pk_period_type = RIAttType(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]);
			Oid			fk_period_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
			StringInfoData intersectbuf;
			StringInfoData replacementsbuf;
			char	   *pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
				"" : "ONLY ";

			quoteOneName(attname, RIAttName(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]));
			sprintf(paramname, "$%d", riinfo->nkeys);

			appendStringInfoString(&querybuf, " AND NOT coalesce(");

			/* Intersect the fk with the old pk range */
			initStringInfo(&intersectbuf);
			appendStringInfoChar(&intersectbuf, '(');
			ri_GenerateQual(&intersectbuf, "",
							attname, fk_period_type,
							riinfo->period_intersect_oper,
							paramname, pk_period_type);
			appendStringInfoChar(&intersectbuf, ')');

			/* Find the remaining history */
			initStringInfo(&replacementsbuf);
			appendStringInfoString(&replacementsbuf, "(SELECT pg_catalog.range_agg(r) FROM ");

			quoteOneName(periodattname, RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));
			quoteRelationName(pkrelname, pk_rel);
			appendStringInfo(&replacementsbuf, "(SELECT y.%s r FROM %s%s y",
							 periodattname, pk_only, pkrelname);

			/* Restrict pk rows to what matches */
			querysep = "WHERE";
			for (int i = 0; i < riinfo->nkeys; i++)
			{
				Oid			pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);

				quoteOneName(attname,
							 RIAttName(pk_rel, riinfo->pk_attnums[i]));
				sprintf(paramname, "$%d", i + 1);
				ri_GenerateQual(&replacementsbuf, querysep,
								paramname, pk_type,
								riinfo->pp_eq_oprs[i],
								attname, pk_type);
				querysep = "AND";
				queryoids[i] = pk_type;
			}
			appendStringInfoString(&replacementsbuf, " FOR KEY SHARE OF y) y2)");

			ri_GenerateQual(&querybuf, "",
							intersectbuf.data, fk_period_type,
							riinfo->agged_period_contained_by_oper,
							replacementsbuf.data, ANYMULTIRANGEOID);
			/* end of coalesce: */
			appendStringInfoString(&querybuf, ", false)");
		}

		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");

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

	/*
	 * We have a plan now. Run it to check for existing references.
	 */
	ri_PerformCheck(riinfo, &qkey, qplan,
					fk_rel, pk_rel,
					oldslot, NULL,
					!is_no_action,
					true,		/* must detect new rows */
					SPI_OK_SELECT);

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

	table_close(fk_rel, RowShareLock);

	return PointerGetDatum(NULL);
}


/*
 * RI_FKey_cascade_del -
 *
 * Cascaded delete foreign key references at delete event on PK table.
 */
Datum
RI_FKey_cascade_del(PG_FUNCTION_ARGS)
{
	TriggerData *trigdata = (TriggerData *) fcinfo->context;
	const RI_ConstraintInfo *riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	TupleTableSlot *oldslot;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;

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

	riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
									trigdata->tg_relation, true);

	/*
	 * Get the relation descriptors of the FK and PK tables and the old tuple.
	 *
	 * fk_rel is opened in RowExclusiveLock mode since that's what our
	 * eventual DELETE will get on it.
	 */
	fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
	pk_rel = trigdata->tg_relation;
	oldslot = trigdata->tg_trigslot;

	SPI_connect();

	/* Fetch or prepare a saved plan for the cascaded delete */
	ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_ONDELETE);

	if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
	{
		StringInfoData querybuf;
		char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		attname[MAX_QUOTED_NAME_LEN];
		char		paramname[16];
		const char *querysep;
		Oid			queryoids[RI_MAX_NUMKEYS];
		const char *fk_only;

		/* ----------
		 * The query string built is
		 *	DELETE FROM [ONLY] <fktable> WHERE $1 = fkatt1 [AND ...]
		 * The type id's for the $ parameters are those of the
		 * corresponding PK attributes.
		 * ----------
		 */
		initStringInfo(&querybuf);
		fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(fkrelname, fk_rel);
		appendStringInfo(&querybuf, "DELETE FROM %s%s",
						 fk_only, fkrelname);
		querysep = "WHERE";
		for (int 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(fk_rel, riinfo->fk_attnums[i]));
			sprintf(paramname, "$%d", i + 1);
			ri_GenerateQual(&querybuf, querysep,
							paramname, pk_type,
							riinfo->pf_eq_oprs[i],
							attname, fk_type);
			querysep = "AND";
			queryoids[i] = pk_type;
		}

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

	/*
	 * We have a plan now. Build up the arguments from the key values in the
	 * deleted PK tuple and delete the referencing rows
	 */
	ri_PerformCheck(riinfo, &qkey, qplan,
					fk_rel, pk_rel,
					oldslot, NULL,
					false,
					true,		/* must detect new rows */
					SPI_OK_DELETE);

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

	table_close(fk_rel, RowExclusiveLock);

	return PointerGetDatum(NULL);
}


/*
 * RI_FKey_cascade_upd -
 *
 * Cascaded update foreign key references at update event on PK table.
 */
Datum
RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
{
	TriggerData *trigdata = (TriggerData *) fcinfo->context;
	const RI_ConstraintInfo *riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	TupleTableSlot *newslot;
	TupleTableSlot *oldslot;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;

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

	riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
									trigdata->tg_relation, true);

	/*
	 * Get the relation descriptors of the FK and PK tables and the new and
	 * old tuple.
	 *
	 * fk_rel is opened in RowExclusiveLock mode since that's what our
	 * eventual UPDATE will get on it.
	 */
	fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
	pk_rel = trigdata->tg_relation;
	newslot = trigdata->tg_newslot;
	oldslot = trigdata->tg_trigslot;

	SPI_connect();

	/* Fetch or prepare a saved plan for the cascaded update */
	ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_ONUPDATE);

	if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
	{
		StringInfoData querybuf;
		StringInfoData qualbuf;
		char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		attname[MAX_QUOTED_NAME_LEN];
		char		paramname[16];
		const char *querysep;
		const char *qualsep;
		Oid			queryoids[RI_MAX_NUMKEYS * 2];
		const char *fk_only;

		/* ----------
		 * The query string built is
		 *	UPDATE [ONLY] <fktable> SET fkatt1 = $1 [, ...]
		 *			WHERE $n = fkatt1 [AND ...]
		 * The type id's for the $ parameters are those of the
		 * corresponding PK attributes.  Note that we are assuming
		 * there is an assignment cast from the PK to the FK type;
		 * else the parser will fail.
		 * ----------
		 */
		initStringInfo(&querybuf);
		initStringInfo(&qualbuf);
		fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(fkrelname, fk_rel);
		appendStringInfo(&querybuf, "UPDATE %s%s SET",
						 fk_only, fkrelname);
		querysep = "";
		qualsep = "WHERE";
		for (int i = 0, j = riinfo->nkeys; i < riinfo->nkeys; i++, j++)
		{
			Oid			pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
			Oid			fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);

			quoteOneName(attname,
						 RIAttName(fk_rel, riinfo->fk_attnums[i]));
			appendStringInfo(&querybuf,
							 "%s %s = $%d",
							 querysep, attname, i + 1);
			sprintf(paramname, "$%d", j + 1);
			ri_GenerateQual(&qualbuf, qualsep,
							paramname, pk_type,
							riinfo->pf_eq_oprs[i],
							attname, fk_type);
			querysep = ",";
			qualsep = "AND";
			queryoids[i] = pk_type;
			queryoids[j] = pk_type;
		}
		appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);

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

	/*
	 * We have a plan now. Run it to update the existing references.
	 */
	ri_PerformCheck(riinfo, &qkey, qplan,
					fk_rel, pk_rel,
					oldslot, newslot,
					false,
					true,		/* must detect new rows */
					SPI_OK_UPDATE);

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

	table_close(fk_rel, RowExclusiveLock);

	return PointerGetDatum(NULL);
}


/*
 * RI_FKey_setnull_del -
 *
 * Set foreign key references to NULL values at delete event on PK table.
 */
Datum
RI_FKey_setnull_del(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_setnull_del", RI_TRIGTYPE_DELETE);

	/* Share code with UPDATE case */
	return ri_set((TriggerData *) fcinfo->context, true, RI_TRIGTYPE_DELETE);
}

/*
 * RI_FKey_setnull_upd -
 *
 * Set foreign key references to NULL at update event on PK table.
 */
Datum
RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_setnull_upd", RI_TRIGTYPE_UPDATE);

	/* Share code with DELETE case */
	return ri_set((TriggerData *) fcinfo->context, true, RI_TRIGTYPE_UPDATE);
}

/*
 * RI_FKey_setdefault_del -
 *
 * Set foreign key references to defaults at delete event on PK table.
 */
Datum
RI_FKey_setdefault_del(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_setdefault_del", RI_TRIGTYPE_DELETE);

	/* Share code with UPDATE case */
	return ri_set((TriggerData *) fcinfo->context, false, RI_TRIGTYPE_DELETE);
}

/*
 * RI_FKey_setdefault_upd -
 *
 * Set foreign key references to defaults at update event on PK table.
 */
Datum
RI_FKey_setdefault_upd(PG_FUNCTION_ARGS)
{
	/* Check that this is a valid trigger call on the right time and event. */
	ri_CheckTrigger(fcinfo, "RI_FKey_setdefault_upd", RI_TRIGTYPE_UPDATE);

	/* Share code with DELETE case */
	return ri_set((TriggerData *) fcinfo->context, false, RI_TRIGTYPE_UPDATE);
}

/*
 * ri_set -
 *
 * Common code for ON DELETE SET NULL, ON DELETE SET DEFAULT, ON UPDATE SET
 * NULL, and ON UPDATE SET DEFAULT.
 */
static Datum
ri_set(TriggerData *trigdata, bool is_set_null, int tgkind)
{
	const RI_ConstraintInfo *riinfo;
	Relation	fk_rel;
	Relation	pk_rel;
	TupleTableSlot *oldslot;
	RI_QueryKey qkey;
	SPIPlanPtr	qplan;
	int32		queryno;

	riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
									trigdata->tg_relation, true);

	/*
	 * Get the relation descriptors of the FK and PK tables and the old tuple.
	 *
	 * fk_rel is opened in RowExclusiveLock mode since that's what our
	 * eventual UPDATE will get on it.
	 */
	fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
	pk_rel = trigdata->tg_relation;
	oldslot = trigdata->tg_trigslot;

	SPI_connect();

	/*
	 * Fetch or prepare a saved plan for the trigger.
	 */
	switch (tgkind)
	{
		case RI_TRIGTYPE_UPDATE:
			queryno = is_set_null
				? RI_PLAN_SETNULL_ONUPDATE
				: RI_PLAN_SETDEFAULT_ONUPDATE;
			break;
		case RI_TRIGTYPE_DELETE:
			queryno = is_set_null
				? RI_PLAN_SETNULL_ONDELETE
				: RI_PLAN_SETDEFAULT_ONDELETE;
			break;
		default:
			elog(ERROR, "invalid tgkind passed to ri_set");
	}

	ri_BuildQueryKey(&qkey, riinfo, queryno);

	if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
	{
		StringInfoData querybuf;
		char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
		char		attname[MAX_QUOTED_NAME_LEN];
		char		paramname[16];
		const char *querysep;
		const char *qualsep;
		Oid			queryoids[RI_MAX_NUMKEYS];
		const char *fk_only;
		int			num_cols_to_set;
		const int16 *set_cols;

		switch (tgkind)
		{
			case RI_TRIGTYPE_UPDATE:
				num_cols_to_set = riinfo->nkeys;
				set_cols = riinfo->fk_attnums;
				break;
			case RI_TRIGTYPE_DELETE:

				/*
				 * If confdelsetcols are present, then we only update the
				 * columns specified in that array, otherwise we update all
				 * the referencing columns.
				 */
				if (riinfo->ndelsetcols != 0)
				{
					num_cols_to_set = riinfo->ndelsetcols;
					set_cols = riinfo->confdelsetcols;
				}
				else
				{
					num_cols_to_set = riinfo->nkeys;
					set_cols = riinfo->fk_attnums;
				}
				break;
			default:
				elog(ERROR, "invalid tgkind passed to ri_set");
		}

		/* ----------
		 * The query string built is
		 *	UPDATE [ONLY] <fktable> SET fkatt1 = {NULL|DEFAULT} [, ...]
		 *			WHERE $1 = fkatt1 [AND ...]
		 * The type id's for the $ parameters are those of the
		 * corresponding PK attributes.
		 * ----------
		 */
		initStringInfo(&querybuf);
		fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
			"" : "ONLY ";
		quoteRelationName(fkrelname, fk_rel);
		appendStringInfo(&querybuf, "UPDATE %s%s SET",
						 fk_only, fkrelname);

		/*
		 * Add assignment clauses
		 */
		querysep = "";
		for (int i = 0; i < num_cols_to_set; i++)
		{
			quoteOneName(attname, RIAttName(fk_rel, set_cols[i]));
			appendStringInfo(&querybuf,
							 "%s %s = %s",
							 querysep, attname,
							 is_set_null ? "NULL" : "DEFAULT");
			querysep = ",";
		}

		/*
		 * Add WHERE clause
		 */
		qualsep = "WHERE";
		for (int 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(fk_rel, riinfo->fk_attnums[i]));

			sprintf(paramname, "$%d", i + 1);
			ri_GenerateQual(&querybuf, qualsep,
							paramname, pk_type,
							riinfo->pf_eq_oprs[i],
							attname, fk_type);
			qualsep = "AND";
			queryoids[i] = pk_type;
		}

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

	/*
	 * We have a plan now. Run it to update the existing references.
	 */
	ri_PerformCheck(riinfo, &qkey, qplan,
					fk_rel, pk_rel,
					oldslot, NULL,
					false,
					true,		/* must detect new rows */
					SPI_OK_UPDATE);

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

	table_close(fk_rel, RowExclusiveLock);

	if (is_set_null)
		return PointerGetDatum(NULL);
	else
	{
		/*
		 * If we just deleted or updated the PK row whose key was equal to the
		 * FK columns' default values, and a referencing row exists in the FK
		 * table, we would have updated that row to the same values it already
		 * had --- and RI_FKey_fk_upd_check_required would hence believe no
		 * check is necessary.  So we need to do another lookup now and in
		 * case a reference still exists, abort the operation.  That is
		 * already implemented in the NO ACTION trigger, so just run it. (This
		 * recheck is only needed in the SET DEFAULT case, since CASCADE would
		 * remove such rows in case of a DELETE operation or would change the
		 * FK key values in case of an UPDATE, while SET NULL is certain to
		 * result in rows that satisfy the FK constraint.)
		 */
		return ri_restrict(trigdata, true);
	}
}


/*
 * RI_FKey_pk_upd_check_required -
 *
 * Check if we really need to fire the RI trigger for an update or delete to a PK
 * relation.  This is called by the AFTER trigger queue manager to see if
 * it can skip queuing an instance of an RI trigger.  Returns true if the
 * trigger must be fired, false if we can prove the constraint will still
 * be satisfied.
 *
 * newslot will be NULL if this is called for a delete.
 */
bool
RI_FKey_pk_upd_check_required(Trigger *trigger, Relation pk_rel,
							  TupleTableSlot *oldslot, TupleTableSlot *newslot)
{
	const RI_ConstraintInfo *riinfo;

	riinfo = ri_FetchConstraintInfo(trigger, pk_rel, true);

	/*
	 * If any old key value is NULL, the row could not have been referenced by
	 * an FK row, so no check is needed.
	 */
	if (ri_NullCheck(RelationGetDescr(pk_rel), oldslot, riinfo, true) != RI_KEYS_NONE_NULL)
		return false;

	/* If all old and new key values are equal, no check is needed */
	if (newslot && ri_KeysEqual(pk_rel, oldslot, newslot, riinfo, true))
		return false;

	/* Else we need to fire the trigger. */
	return true;
}

/*
 * RI_FKey_fk_upd_check_required -
 *
 * Check if we really need to fire the RI trigger for an update to an FK
 * relation.  This is called by the AFTER trigger queue manager to see if
 * it can skip queuing an instance of an RI trigger.  Returns true if the
 * trigger must be fired, false if we can prove the constraint will still
 * be satisfied.
 */
bool
RI_FKey_fk_upd_check_required(Trigger *trigger, Relation fk_rel,
							  TupleTableSlot *oldslot, TupleTableSlot *newslot)
{
	const RI_ConstraintInfo *riinfo;
	int			ri_nullcheck;

	/*
	 * AfterTriggerSaveEvent() handles things such that this function is never
	 * called for partitioned tables.
	 */
	Assert(fk_rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE);

	riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);

	ri_nullcheck = ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false);

	/*
	 * If all new key values are NULL, the row satisfies the constraint, so no
	 * check is needed.
	 */
	if (ri_nullcheck == RI_KEYS_ALL_NULL)
		return false;

	/*
	 * If some new key values are NULL, the behavior depends on the match
	 * type.
	 */
	else if (ri_nullcheck == RI_KEYS_SOME_NULL)
	{
		switch (riinfo->confmatchtype)
		{
			case FKCONSTR_MATCH_SIMPLE:

				/*
				 * If any new key value is NULL, the row must satisfy the
				 * constraint, so no check is needed.
				 */
				return false;

			case FKCONSTR_MATCH_PARTIAL:

				/*
				 * Don't know, must run full check.
				 */
				break;

			case FKCONSTR_MATCH_FULL:

				/*
				 * If some new key values are NULL, the row fails the
				 * constraint.  We must not throw error here, because the row
				 * might get invalidated before the constraint is to be
				 * checked, but we should queue the event to apply the check
				 * later.
				 */
				return true;
		}
	}

	/*
	 * Continues here for no new key values are NULL, or we couldn't decide
	 * yet.
	 */

	/*
	 * If the original row was inserted by our own transaction, we must fire
	 * the trigger whether or not the keys are equal.  This is because our
	 * UPDATE will invalidate the INSERT so that the INSERT RI trigger will
	 * not do anything; so we had better do the UPDATE check.  (We could skip
	 * this if we knew the INSERT trigger already fired, but there is no easy
	 * way to know that.)
	 */
	if (slot_is_current_xact_tuple(oldslot))
		return true;

	/* If all old and new key values are equal, no check is needed */
	if (ri_KeysEqual(fk_rel, oldslot, newslot, riinfo, false))
		return false;

	/* Else we need to fire the trigger. */
	return true;
}

/*
 * RI_Initial_Check -
 *
 * Check an entire table for non-matching values using a single query.
 * This is not a trigger procedure, but is called during ALTER TABLE
 * ADD FOREIGN KEY to validate the initial table contents.
 *
 * We expect that the caller has made provision to prevent any problems
 * caused by concurrent actions. This could be either by locking rel and
 * pkrel at ShareRowExclusiveLock or higher, or by otherwise ensuring
 * that triggers implementing the checks are already active.
 * Hence, we do not need to lock individual rows for the check.
 *
 * If the check fails because the current user doesn't have permissions
 * to read both tables, return false to let our caller know that they will
 * need to do something else to check the constraint.
 */
bool
RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
{
	const RI_ConstraintInfo *riinfo;
	StringInfoData querybuf;
	char		pkrelname[MAX_QUOTED_REL_NAME_LEN];
	char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
	char		pkattname[MAX_QUOTED_NAME_LEN + 3];
	char		fkattname[MAX_QUOTED_NAME_LEN + 3];
	RangeTblEntry *rte;
	RTEPermissionInfo *pk_perminfo;
	RTEPermissionInfo *fk_perminfo;
	List	   *rtes = NIL;
	List	   *perminfos = NIL;
	const char *sep;
	const char *fk_only;
	const char *pk_only;
	int			save_nestlevel;
	char		workmembuf[32];
	int			spi_result;
	SPIPlanPtr	qplan;

	riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);

	/*
	 * Check to make sure current user has enough permissions to do the test
	 * query.  (If not, caller can fall back to the trigger method, which
	 * works because it changes user IDs on the fly.)
	 *
	 * XXX are there any other show-stopper conditions to check?
	 */
	pk_perminfo = makeNode(RTEPermissionInfo);
	pk_perminfo->relid = RelationGetRelid(pk_rel);
	pk_perminfo->requiredPerms = ACL_SELECT;
	perminfos = lappend(perminfos, pk_perminfo);
	rte = makeNode(RangeTblEntry);
	rte->rtekind = RTE_RELATION;
	rte->relid = RelationGetRelid(pk_rel);
	rte->relkind = pk_rel->rd_rel->relkind;
	rte->rellockmode = AccessShareLock;
	rte->perminfoindex = list_length(perminfos);
	rtes = lappend(rtes, rte);

	fk_perminfo = makeNode(RTEPermissionInfo);
	fk_perminfo->relid = RelationGetRelid(fk_rel);
	fk_perminfo->requiredPerms = ACL_SELECT;
	perminfos = lappend(perminfos, fk_perminfo);
	rte = makeNode(RangeTblEntry);
	rte->rtekind = RTE_RELATION;
	rte->relid = RelationGetRelid(fk_rel);
	rte->relkind = fk_rel->rd_rel->relkind;
	rte->rellockmode = AccessShareLock;
	rte->perminfoindex = list_length(perminfos);
	rtes = lappend(rtes, rte);

	for (int i = 0; i < riinfo->nkeys; i++)
	{
		int			attno;

		attno = riinfo->pk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
		pk_perminfo->selectedCols = bms_add_member(pk_perminfo->selectedCols, attno);

		attno = riinfo->fk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
		fk_perminfo->selectedCols = bms_add_member(fk_perminfo->selectedCols, attno);
	}

	if (!ExecCheckPermissions(rtes, perminfos, false))
		return false;

	/*
	 * Also punt if RLS is enabled on either table unless this role has the
	 * bypassrls right or is the table owner of the table(s) involved which
	 * have RLS enabled.
	 */
	if (!has_bypassrls_privilege(GetUserId()) &&
		((pk_rel->rd_rel->relrowsecurity &&
		  !object_ownercheck(RelationRelationId, RelationGetRelid(pk_rel),
							 GetUserId())) ||
		 (fk_rel->rd_rel->relrowsecurity &&
		  !object_ownercheck(RelationRelationId, RelationGetRelid(fk_rel),
							 GetUserId()))))
		return false;

	/*----------
	 * The query string built is:
	 *	SELECT fk.keycols FROM [ONLY] relname fk
	 *	 LEFT OUTER JOIN [ONLY] pkrelname pk
	 *	 ON (pk.pkkeycol1=fk.keycol1 [AND ...])
	 *	 WHERE pk.pkkeycol1 IS NULL AND
	 * For MATCH SIMPLE:
	 *	 (fk.keycol1 IS NOT NULL [AND ...])
	 * For MATCH FULL:
	 *	 (fk.keycol1 IS NOT NULL [OR ...])
	 *
	 * We attach COLLATE clauses to the operators when comparing columns
	 * that have different collations.
	 *----------
	 */
	initStringInfo(&querybuf);
	appendStringInfoString(&querybuf, "SELECT ");
	sep = "";
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		quoteOneName(fkattname,
					 RIAttName(fk_rel, riinfo->fk_attnums[i]));
		appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
		sep = ", ";
	}

	quoteRelationName(pkrelname, pk_rel);
	quoteRelationName(fkrelname, fk_rel);
	fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
		"" : "ONLY ";
	pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
		"" : "ONLY ";
	appendStringInfo(&querybuf,
					 " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON",
					 fk_only, fkrelname, pk_only, pkrelname);

	strcpy(pkattname, "pk.");
	strcpy(fkattname, "fk.");
	sep = "(";
	for (int 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]);
		Oid			pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
		Oid			fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);

		quoteOneName(pkattname + 3,
					 RIAttName(pk_rel, riinfo->pk_attnums[i]));
		quoteOneName(fkattname + 3,
					 RIAttName(fk_rel, riinfo->fk_attnums[i]));
		ri_GenerateQual(&querybuf, sep,
						pkattname, pk_type,
						riinfo->pf_eq_oprs[i],
						fkattname, fk_type);
		if (pk_coll != fk_coll)
			ri_GenerateQualCollation(&querybuf, pk_coll);
		sep = "AND";
	}

	/*
	 * It's sufficient to test any one pk attribute for null to detect a join
	 * failure.
	 */
	quoteOneName(pkattname, RIAttName(pk_rel, riinfo->pk_attnums[0]));
	appendStringInfo(&querybuf, ") WHERE pk.%s IS NULL AND (", pkattname);

	sep = "";
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
		appendStringInfo(&querybuf,
						 "%sfk.%s IS NOT NULL",
						 sep, fkattname);
		switch (riinfo->confmatchtype)
		{
			case FKCONSTR_MATCH_SIMPLE:
				sep = " AND ";
				break;
			case FKCONSTR_MATCH_FULL:
				sep = " OR ";
				break;
		}
	}
	appendStringInfoChar(&querybuf, ')');

	/*
	 * Temporarily increase work_mem so that the check query can be executed
	 * more efficiently.  It seems okay to do this because the query is simple
	 * enough to not use a multiple of work_mem, and one typically would not
	 * have many large foreign-key validations happening concurrently.  So
	 * this seems to meet the criteria for being considered a "maintenance"
	 * operation, and accordingly we use maintenance_work_mem.  However, we
	 * must also set hash_mem_multiplier to 1, since it is surely not okay to
	 * let that get applied to the maintenance_work_mem value.
	 *
	 * We use the equivalent of a function SET option to allow the setting to
	 * persist for exactly the duration of the check query.  guc.c also takes
	 * care of undoing the setting on error.
	 */
	save_nestlevel = NewGUCNestLevel();

	snprintf(workmembuf, sizeof(workmembuf), "%d", maintenance_work_mem);
	(void) set_config_option("work_mem", workmembuf,
							 PGC_USERSET, PGC_S_SESSION,
							 GUC_ACTION_SAVE, true, 0, false);
	(void) set_config_option("hash_mem_multiplier", "1",
							 PGC_USERSET, PGC_S_SESSION,
							 GUC_ACTION_SAVE, true, 0, false);

	SPI_connect();

	/*
	 * Generate the plan.  We don't need to cache it, and there are no
	 * arguments to the plan.
	 */
	qplan = SPI_prepare(querybuf.data, 0, NULL);

	if (qplan == NULL)
		elog(ERROR, "SPI_prepare returned %s for %s",
			 SPI_result_code_string(SPI_result), querybuf.data);

	/*
	 * Run the plan.  For safety we force a current snapshot to be used. (In
	 * transaction-snapshot mode, this arguably violates transaction isolation
	 * rules, but we really haven't got much choice.) We don't need to
	 * register the snapshot, because SPI_execute_snapshot will see to it. We
	 * need at most one tuple returned, so pass limit = 1.
	 */
	spi_result = SPI_execute_snapshot(qplan,
									  NULL, NULL,
									  GetLatestSnapshot(),
									  InvalidSnapshot,
									  true, false, 1);

	/* Check result */
	if (spi_result != SPI_OK_SELECT)
		elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));

	/* Did we find a tuple violating the constraint? */
	if (SPI_processed > 0)
	{
		TupleTableSlot *slot;
		HeapTuple	tuple = SPI_tuptable->vals[0];
		TupleDesc	tupdesc = SPI_tuptable->tupdesc;
		RI_ConstraintInfo fake_riinfo;

		slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);

		heap_deform_tuple(tuple, tupdesc,
						  slot->tts_values, slot->tts_isnull);
		ExecStoreVirtualTuple(slot);

		/*
		 * The columns to look at in the result tuple are 1..N, not whatever
		 * they are in the fk_rel.  Hack up riinfo so that the subroutines
		 * called here will behave properly.
		 *
		 * In addition to this, we have to pass the correct tupdesc to
		 * ri_ReportViolation, overriding its normal habit of using the pk_rel
		 * or fk_rel's tupdesc.
		 */
		memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
		for (int i = 0; i < fake_riinfo.nkeys; i++)
			fake_riinfo.fk_attnums[i] = i + 1;

		/*
		 * If it's MATCH FULL, and there are any nulls in the FK keys,
		 * complain about that rather than the lack of a match.  MATCH FULL
		 * disallows partially-null FK rows.
		 */
		if (fake_riinfo.confmatchtype == FKCONSTR_MATCH_FULL &&
			ri_NullCheck(tupdesc, slot, &fake_riinfo, false) != RI_KEYS_NONE_NULL)
			ereport(ERROR,
					(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
					 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
							RelationGetRelationName(fk_rel),
							NameStr(fake_riinfo.conname)),
					 errdetail("MATCH FULL does not allow mixing of null and nonnull key values."),
					 errtableconstraint(fk_rel,
										NameStr(fake_riinfo.conname))));

		/*
		 * We tell ri_ReportViolation we were doing the RI_PLAN_CHECK_LOOKUPPK
		 * query, which isn't true, but will cause it to use
		 * fake_riinfo.fk_attnums as we need.
		 */
		ri_ReportViolation(&fake_riinfo,
						   pk_rel, fk_rel,
						   slot, tupdesc,
						   RI_PLAN_CHECK_LOOKUPPK, false, false);

		ExecDropSingleTupleTableSlot(slot);
	}

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

	/*
	 * Restore work_mem and hash_mem_multiplier.
	 */
	AtEOXact_GUC(true, save_nestlevel);

	return true;
}

/*
 * RI_PartitionRemove_Check -
 *
 * Verify no referencing values exist, when a partition is detached on
 * the referenced side of a foreign key constraint.
 */
void
RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
{
	const RI_ConstraintInfo *riinfo;
	StringInfoData querybuf;
	char	   *constraintDef;
	char		pkrelname[MAX_QUOTED_REL_NAME_LEN];
	char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
	char		pkattname[MAX_QUOTED_NAME_LEN + 3];
	char		fkattname[MAX_QUOTED_NAME_LEN + 3];
	const char *sep;
	const char *fk_only;
	int			save_nestlevel;
	char		workmembuf[32];
	int			spi_result;
	SPIPlanPtr	qplan;
	int			i;

	riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);

	/*
	 * We don't check permissions before displaying the error message, on the
	 * assumption that the user detaching the partition must have enough
	 * privileges to examine the table contents anyhow.
	 */

	/*----------
	 * The query string built is:
	 *  SELECT fk.keycols FROM [ONLY] relname fk
	 *    JOIN pkrelname pk
	 *    ON (pk.pkkeycol1=fk.keycol1 [AND ...])
	 *    WHERE (<partition constraint>) AND
	 * For MATCH SIMPLE:
	 *   (fk.keycol1 IS NOT NULL [AND ...])
	 * For MATCH FULL:
	 *   (fk.keycol1 IS NOT NULL [OR ...])
	 *
	 * We attach COLLATE clauses to the operators when comparing columns
	 * that have different collations.
	 *----------
	 */
	initStringInfo(&querybuf);
	appendStringInfoString(&querybuf, "SELECT ");
	sep = "";
	for (i = 0; i < riinfo->nkeys; i++)
	{
		quoteOneName(fkattname,
					 RIAttName(fk_rel, riinfo->fk_attnums[i]));
		appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
		sep = ", ";
	}

	quoteRelationName(pkrelname, pk_rel);
	quoteRelationName(fkrelname, fk_rel);
	fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
		"" : "ONLY ";
	appendStringInfo(&querybuf,
					 " FROM %s%s fk JOIN %s pk ON",
					 fk_only, fkrelname, pkrelname);
	strcpy(pkattname, "pk.");
	strcpy(fkattname, "fk.");
	sep = "(";
	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]);
		Oid			pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
		Oid			fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);

		quoteOneName(pkattname + 3,
					 RIAttName(pk_rel, riinfo->pk_attnums[i]));
		quoteOneName(fkattname + 3,
					 RIAttName(fk_rel, riinfo->fk_attnums[i]));
		ri_GenerateQual(&querybuf, sep,
						pkattname, pk_type,
						riinfo->pf_eq_oprs[i],
						fkattname, fk_type);
		if (pk_coll != fk_coll)
			ri_GenerateQualCollation(&querybuf, pk_coll);
		sep = "AND";
	}

	/*
	 * Start the WHERE clause with the partition constraint (except if this is
	 * the default partition and there's no other partition, because the
	 * partition constraint is the empty string in that case.)
	 */
	constraintDef = pg_get_partconstrdef_string(RelationGetRelid(pk_rel), "pk");
	if (constraintDef && constraintDef[0] != '\0')
		appendStringInfo(&querybuf, ") WHERE %s AND (",
						 constraintDef);
	else
		appendStringInfoString(&querybuf, ") WHERE (");

	sep = "";
	for (i = 0; i < riinfo->nkeys; i++)
	{
		quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
		appendStringInfo(&querybuf,
						 "%sfk.%s IS NOT NULL",
						 sep, fkattname);
		switch (riinfo->confmatchtype)
		{
			case FKCONSTR_MATCH_SIMPLE:
				sep = " AND ";
				break;
			case FKCONSTR_MATCH_FULL:
				sep = " OR ";
				break;
		}
	}
	appendStringInfoChar(&querybuf, ')');

	/*
	 * Temporarily increase work_mem so that the check query can be executed
	 * more efficiently.  It seems okay to do this because the query is simple
	 * enough to not use a multiple of work_mem, and one typically would not
	 * have many large foreign-key validations happening concurrently.  So
	 * this seems to meet the criteria for being considered a "maintenance"
	 * operation, and accordingly we use maintenance_work_mem.  However, we
	 * must also set hash_mem_multiplier to 1, since it is surely not okay to
	 * let that get applied to the maintenance_work_mem value.
	 *
	 * We use the equivalent of a function SET option to allow the setting to
	 * persist for exactly the duration of the check query.  guc.c also takes
	 * care of undoing the setting on error.
	 */
	save_nestlevel = NewGUCNestLevel();

	snprintf(workmembuf, sizeof(workmembuf), "%d", maintenance_work_mem);
	(void) set_config_option("work_mem", workmembuf,
							 PGC_USERSET, PGC_S_SESSION,
							 GUC_ACTION_SAVE, true, 0, false);
	(void) set_config_option("hash_mem_multiplier", "1",
							 PGC_USERSET, PGC_S_SESSION,
							 GUC_ACTION_SAVE, true, 0, false);

	SPI_connect();

	/*
	 * Generate the plan.  We don't need to cache it, and there are no
	 * arguments to the plan.
	 */
	qplan = SPI_prepare(querybuf.data, 0, NULL);

	if (qplan == NULL)
		elog(ERROR, "SPI_prepare returned %s for %s",
			 SPI_result_code_string(SPI_result), querybuf.data);

	/*
	 * Run the plan.  For safety we force a current snapshot to be used. (In
	 * transaction-snapshot mode, this arguably violates transaction isolation
	 * rules, but we really haven't got much choice.) We don't need to
	 * register the snapshot, because SPI_execute_snapshot will see to it. We
	 * need at most one tuple returned, so pass limit = 1.
	 */
	spi_result = SPI_execute_snapshot(qplan,
									  NULL, NULL,
									  GetLatestSnapshot(),
									  InvalidSnapshot,
									  true, false, 1);

	/* Check result */
	if (spi_result != SPI_OK_SELECT)
		elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));

	/* Did we find a tuple that would violate the constraint? */
	if (SPI_processed > 0)
	{
		TupleTableSlot *slot;
		HeapTuple	tuple = SPI_tuptable->vals[0];
		TupleDesc	tupdesc = SPI_tuptable->tupdesc;
		RI_ConstraintInfo fake_riinfo;

		slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);

		heap_deform_tuple(tuple, tupdesc,
						  slot->tts_values, slot->tts_isnull);
		ExecStoreVirtualTuple(slot);

		/*
		 * The columns to look at in the result tuple are 1..N, not whatever
		 * they are in the fk_rel.  Hack up riinfo so that ri_ReportViolation
		 * will behave properly.
		 *
		 * In addition to this, we have to pass the correct tupdesc to
		 * ri_ReportViolation, overriding its normal habit of using the pk_rel
		 * or fk_rel's tupdesc.
		 */
		memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
		for (i = 0; i < fake_riinfo.nkeys; i++)
			fake_riinfo.pk_attnums[i] = i + 1;

		ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel,
						   slot, tupdesc, 0, false, true);
	}

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

	/*
	 * Restore work_mem and hash_mem_multiplier.
	 */
	AtEOXact_GUC(true, save_nestlevel);
}


/* ----------
 * Local functions below
 * ----------
 */


/*
 * quoteOneName --- safely quote a single SQL name
 *
 * buffer must be MAX_QUOTED_NAME_LEN long (includes room for \0)
 */
static void
quoteOneName(char *buffer, const char *name)
{
	/* Rather than trying to be smart, just always quote it. */
	*buffer++ = '"';
	while (*name)
	{
		if (*name == '"')
			*buffer++ = '"';
		*buffer++ = *name++;
	}
	*buffer++ = '"';
	*buffer = '\0';
}

/*
 * quoteRelationName --- safely quote a fully qualified relation name
 *
 * buffer must be MAX_QUOTED_REL_NAME_LEN long (includes room for \0)
 */
static void
quoteRelationName(char *buffer, Relation rel)
{
	quoteOneName(buffer, get_namespace_name(RelationGetNamespace(rel)));
	buffer += strlen(buffer);
	*buffer++ = '.';
	quoteOneName(buffer, RelationGetRelationName(rel));
}

/*
 * ri_GenerateQual --- generate a WHERE clause equating two variables
 *
 * This basically appends " sep leftop op rightop" to buf, adding casts
 * and schema qualification as needed to ensure that the parser will select
 * the operator we specify.  leftop and rightop should be parenthesized
 * if they aren't variables or parameters.
 */
static void
ri_GenerateQual(StringInfo buf,
				const char *sep,
				const char *leftop, Oid leftoptype,
				Oid opoid,
				const char *rightop, Oid rightoptype)
{
	appendStringInfo(buf, " %s ", sep);
	generate_operator_clause(buf, leftop, leftoptype, opoid,
							 rightop, rightoptype);
}

/*
 * ri_GenerateQualCollation --- add a COLLATE spec to a WHERE clause
 *
 * We only have to use this function when directly comparing the referencing
 * and referenced columns, if they are of different collations; else the
 * parser will fail to resolve the collation to use.  We don't need to use
 * this function for RI queries that compare a variable to a $n parameter.
 * Since parameter symbols always have default collation, the effect will be
 * to use the variable's collation.
 *
 * Note that we require that the collations of the referencing and the
 * referenced column have the same notion of equality: Either they have to
 * both be deterministic or else they both have to be the same.  (See also
 * ATAddForeignKeyConstraint().)
 */
static void
ri_GenerateQualCollation(StringInfo buf, Oid collation)
{
	HeapTuple	tp;
	Form_pg_collation colltup;
	char	   *collname;
	char		onename[MAX_QUOTED_NAME_LEN];

	/* Nothing to do if it's a noncollatable data type */
	if (!OidIsValid(collation))
		return;

	tp = SearchSysCache1(COLLOID, ObjectIdGetDatum(collation));
	if (!HeapTupleIsValid(tp))
		elog(ERROR, "cache lookup failed for collation %u", collation);
	colltup = (Form_pg_collation) GETSTRUCT(tp);
	collname = NameStr(colltup->collname);

	/*
	 * We qualify the name always, for simplicity and to ensure the query is
	 * not search-path-dependent.
	 */
	quoteOneName(onename, get_namespace_name(colltup->collnamespace));
	appendStringInfo(buf, " COLLATE %s", onename);
	quoteOneName(onename, collname);
	appendStringInfo(buf, ".%s", onename);

	ReleaseSysCache(tp);
}

/* ----------
 * ri_BuildQueryKey -
 *
 *	Construct a hashtable key for a prepared SPI plan of an FK constraint.
 *
 *		key: output argument, *key is filled in based on the other arguments
 *		riinfo: info derived from pg_constraint entry
 *		constr_queryno: an internal number identifying the query type
 *			(see RI_PLAN_XXX constants at head of file)
 * ----------
 */
static void
ri_BuildQueryKey(RI_QueryKey *key, const RI_ConstraintInfo *riinfo,
				 int32 constr_queryno)
{
	/*
	 * Inherited constraints with a common ancestor can share ri_query_cache
	 * entries for all query types except RI_PLAN_CHECK_LOOKUPPK_FROM_PK.
	 * Except in that case, the query processes the other table involved in
	 * the FK constraint (i.e., not the table on which the trigger has been
	 * fired), and so it will be the same for all members of the inheritance
	 * tree.  So we may use the root constraint's OID in the hash key, rather
	 * than the constraint's own OID.  This avoids creating duplicate SPI
	 * plans, saving lots of work and memory when there are many partitions
	 * with similar FK constraints.
	 *
	 * (Note that we must still have a separate RI_ConstraintInfo for each
	 * constraint, because partitions can have different column orders,
	 * resulting in different pk_attnums[] or fk_attnums[] array contents.)
	 *
	 * We assume struct RI_QueryKey contains no padding bytes, else we'd need
	 * to use memset to clear them.
	 */
	if (constr_queryno != RI_PLAN_CHECK_LOOKUPPK_FROM_PK)
		key->constr_id = riinfo->constraint_root_id;
	else
		key->constr_id = riinfo->constraint_id;
	key->constr_queryno = constr_queryno;
}

/*
 * Check that RI trigger function was called in expected context
 */
static void
ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname, int tgkind)
{
	TriggerData *trigdata = (TriggerData *) fcinfo->context;

	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)));

	/*
	 * Check proper event
	 */
	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)));

	switch (tgkind)
	{
		case RI_TRIGTYPE_INSERT:
			if (!TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
				ereport(ERROR,
						(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
						 errmsg("function \"%s\" must be fired for INSERT", funcname)));
			break;
		case RI_TRIGTYPE_UPDATE:
			if (!TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
				ereport(ERROR,
						(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
						 errmsg("function \"%s\" must be fired for UPDATE", funcname)));
			break;
		case RI_TRIGTYPE_DELETE:
			if (!TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
				ereport(ERROR,
						(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
						 errmsg("function \"%s\" must be fired for DELETE", funcname)));
			break;
	}
}


/*
 * Fetch the RI_ConstraintInfo struct for the trigger's FK constraint.
 */
static RI_ConstraintInfo *
ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
{
	Oid			constraintOid = trigger->tgconstraint;
	RI_ConstraintInfo *riinfo;

	/*
	 * Check that the FK constraint's OID is available; it might not be if
	 * we've been invoked via an ordinary trigger or an old-style "constraint
	 * trigger".
	 */
	if (!OidIsValid(constraintOid))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("no pg_constraint entry for trigger \"%s\" on table \"%s\"",
						trigger->tgname, RelationGetRelationName(trig_rel)),
				 errhint("Remove this referential integrity trigger and its mates, then do ALTER TABLE ADD CONSTRAINT.")));

	/* Find or create a hashtable entry for the constraint */
	riinfo = ri_LoadConstraintInfo(constraintOid);

	/* Do some easy cross-checks against the trigger call data */
	if (rel_is_pk)
	{
		if (riinfo->fk_relid != trigger->tgconstrrelid ||
			riinfo->pk_relid != RelationGetRelid(trig_rel))
			elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
				 trigger->tgname, RelationGetRelationName(trig_rel));
	}
	else
	{
		if (riinfo->fk_relid != RelationGetRelid(trig_rel) ||
			riinfo->pk_relid != trigger->tgconstrrelid)
			elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
				 trigger->tgname, RelationGetRelationName(trig_rel));
	}

	if (riinfo->confmatchtype != FKCONSTR_MATCH_FULL &&
		riinfo->confmatchtype != FKCONSTR_MATCH_PARTIAL &&
		riinfo->confmatchtype != FKCONSTR_MATCH_SIMPLE)
		elog(ERROR, "unrecognized confmatchtype: %d",
			 riinfo->confmatchtype);

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

	return riinfo;
}

/*
 * Fetch or create the RI_ConstraintInfo struct for an FK constraint.
 */
static RI_ConstraintInfo *
ri_LoadConstraintInfo(Oid constraintOid)
{
	RI_ConstraintInfo *riinfo;
	bool		found;
	HeapTuple	tup;
	Form_pg_constraint conForm;

	/*
	 * On the first call initialize the hashtable
	 */
	if (!ri_constraint_cache)
		ri_InitHashTables();

	/*
	 * Find or create a hash entry.  If we find a valid one, just return it.
	 */
	riinfo = (RI_ConstraintInfo *) hash_search(ri_constraint_cache,
											   &constraintOid,
											   HASH_ENTER, &found);
	if (!found)
		riinfo->valid = false;
	else if (riinfo->valid)
		return riinfo;

	/*
	 * Fetch the pg_constraint row so we can fill in the entry.
	 */
	tup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constraintOid));
	if (!HeapTupleIsValid(tup)) /* should not happen */
		elog(ERROR, "cache lookup failed for constraint %u", constraintOid);
	conForm = (Form_pg_constraint) GETSTRUCT(tup);

	if (conForm->contype != CONSTRAINT_FOREIGN) /* should not happen */
		elog(ERROR, "constraint %u is not a foreign key constraint",
			 constraintOid);

	/* And extract data */
	Assert(riinfo->constraint_id == constraintOid);
	if (OidIsValid(conForm->conparentid))
		riinfo->constraint_root_id =
			get_ri_constraint_root(conForm->conparentid);
	else
		riinfo->constraint_root_id = constraintOid;
	riinfo->oidHashValue = GetSysCacheHashValue1(CONSTROID,
												 ObjectIdGetDatum(constraintOid));
	riinfo->rootHashValue = GetSysCacheHashValue1(CONSTROID,
												  ObjectIdGetDatum(riinfo->constraint_root_id));
	memcpy(&riinfo->conname, &conForm->conname, sizeof(NameData));
	riinfo->pk_relid = conForm->confrelid;
	riinfo->fk_relid = conForm->conrelid;
	riinfo->confupdtype = conForm->confupdtype;
	riinfo->confdeltype = conForm->confdeltype;
	riinfo->confmatchtype = conForm->confmatchtype;
	riinfo->hasperiod = conForm->conperiod;

	DeconstructFkConstraintRow(tup,
							   &riinfo->nkeys,
							   riinfo->fk_attnums,
							   riinfo->pk_attnums,
							   riinfo->pf_eq_oprs,
							   riinfo->pp_eq_oprs,
							   riinfo->ff_eq_oprs,
							   &riinfo->ndelsetcols,
							   riinfo->confdelsetcols);

	/*
	 * For temporal FKs, get the operators and functions we need. We ask the
	 * opclass of the PK element for these. This all gets cached (as does the
	 * generated plan), so there's no performance issue.
	 */
	if (riinfo->hasperiod)
	{
		Oid			opclass = get_index_column_opclass(conForm->conindid, riinfo->nkeys);

		FindFKPeriodOpers(opclass,
						  &riinfo->period_contained_by_oper,
						  &riinfo->agged_period_contained_by_oper,
						  &riinfo->period_intersect_oper);
	}

	/* Metadata used by fast path. */
	riinfo->conindid = conForm->conindid;
	riinfo->pk_is_partitioned =
		(get_rel_relkind(riinfo->pk_relid) == RELKIND_PARTITIONED_TABLE);
	riinfo->pk_index_is_btree =
		(get_rel_relam(riinfo->conindid) == BTREE_AM_OID);

	ReleaseSysCache(tup);

	/*
	 * For efficient processing of invalidation messages below, we keep a
	 * doubly-linked count list of all currently valid entries.
	 */
	dclist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);

	riinfo->valid = true;

	riinfo->fpmeta = NULL;

	return riinfo;
}

/*
 * get_ri_constraint_root
 *		Returns the OID of the constraint's root parent
 */
static Oid
get_ri_constraint_root(Oid constrOid)
{
	for (;;)
	{
		HeapTuple	tuple;
		Oid			constrParentOid;

		tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constrOid));
		if (!HeapTupleIsValid(tuple))
			elog(ERROR, "cache lookup failed for constraint %u", constrOid);
		constrParentOid = ((Form_pg_constraint) GETSTRUCT(tuple))->conparentid;
		ReleaseSysCache(tuple);
		if (!OidIsValid(constrParentOid))
			break;				/* we reached the root constraint */
		constrOid = constrParentOid;
	}
	return constrOid;
}

/*
 * Callback for pg_constraint inval events
 *
 * While most syscache callbacks just flush all their entries, pg_constraint
 * gets enough update traffic that it's probably worth being smarter.
 * Invalidate any ri_constraint_cache entry associated with the syscache
 * entry with the specified hash value, or all entries if hashvalue == 0.
 *
 * Note: at the time a cache invalidation message is processed there may be
 * active references to the cache.  Because of this we never remove entries
 * from the cache, but only mark them invalid, which is harmless to active
 * uses.  (Any query using an entry should hold a lock sufficient to keep that
 * data from changing under it --- but we may get cache flushes anyway.)
 */
static void
InvalidateConstraintCacheCallBack(Datum arg, SysCacheIdentifier cacheid,
								  uint32 hashvalue)
{
	dlist_mutable_iter iter;

	Assert(ri_constraint_cache != NULL);

	/*
	 * If the list of currently valid entries gets excessively large, we mark
	 * them all invalid so we can empty the list.  This arrangement avoids
	 * O(N^2) behavior in situations where a session touches many foreign keys
	 * and also does many ALTER TABLEs, such as a restore from pg_dump.
	 */
	if (dclist_count(&ri_constraint_cache_valid_list) > 1000)
		hashvalue = 0;			/* pretend it's a cache reset */

	dclist_foreach_modify(iter, &ri_constraint_cache_valid_list)
	{
		RI_ConstraintInfo *riinfo = dclist_container(RI_ConstraintInfo,
													 valid_link, iter.cur);

		/*
		 * We must invalidate not only entries directly matching the given
		 * hash value, but also child entries, in case the invalidation
		 * affects a root constraint.
		 */
		if (hashvalue == 0 ||
			riinfo->oidHashValue == hashvalue ||
			riinfo->rootHashValue == hashvalue)
		{
			riinfo->valid = false;
			if (riinfo->fpmeta)
			{
				pfree(riinfo->fpmeta);
				riinfo->fpmeta = NULL;
			}
			/* Remove invalidated entries from the list, too */
			dclist_delete_from(&ri_constraint_cache_valid_list, iter.cur);
		}
	}
}


/*
 * Prepare execution plan for a query to enforce an RI restriction
 */
static SPIPlanPtr
ri_PlanCheck(const char *querystr, int nargs, const Oid *argtypes,
			 RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel)
{
	SPIPlanPtr	qplan;
	Relation	query_rel;
	Oid			save_userid;
	int			save_sec_context;

	/*
	 * Use the query type code to determine whether the query is run against
	 * the PK or FK table; we'll do the check as that table's owner
	 */
	if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK)
		query_rel = pk_rel;
	else
		query_rel = fk_rel;

	/* Switch to proper UID to perform check as */
	GetUserIdAndSecContext(&save_userid, &save_sec_context);
	SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
						   SECURITY_NOFORCE_RLS);

	/* Create the plan */
	qplan = SPI_prepare(querystr, nargs, argtypes);

	if (qplan == NULL)
		elog(ERROR, "SPI_prepare returned %s for %s", SPI_result_code_string(SPI_result), querystr);

	/* Restore UID and security context */
	SetUserIdAndSecContext(save_userid, save_sec_context);

	/* Save the plan */
	SPI_keepplan(qplan);
	ri_HashPreparedPlan(qkey, qplan);

	return qplan;
}

/*
 * Perform a query to enforce an RI restriction
 */
static bool
ri_PerformCheck(const RI_ConstraintInfo *riinfo,
				RI_QueryKey *qkey, SPIPlanPtr qplan,
				Relation fk_rel, Relation pk_rel,
				TupleTableSlot *oldslot, TupleTableSlot *newslot,
				bool is_restrict,
				bool detectNewRows, int expect_OK)
{
	Relation	query_rel,
				source_rel;
	bool		source_is_pk;
	Snapshot	test_snapshot;
	Snapshot	crosscheck_snapshot;
	int			limit;
	int			spi_result;
	Oid			save_userid;
	int			save_sec_context;
	Datum		vals[RI_MAX_NUMKEYS * 2];
	char		nulls[RI_MAX_NUMKEYS * 2];

	/*
	 * Use the query type code to determine whether the query is run against
	 * the PK or FK table; we'll do the check as that table's owner
	 */
	if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK)
		query_rel = pk_rel;
	else
		query_rel = fk_rel;

	/*
	 * The values for the query are taken from the table on which the trigger
	 * is called - it is normally the other one with respect to query_rel. An
	 * exception is ri_Check_Pk_Match(), which uses the PK table for both (and
	 * sets queryno to RI_PLAN_CHECK_LOOKUPPK_FROM_PK).  We might eventually
	 * need some less klugy way to determine this.
	 */
	if (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK)
	{
		source_rel = fk_rel;
		source_is_pk = false;
	}
	else
	{
		source_rel = pk_rel;
		source_is_pk = true;
	}

	/* Extract the parameters to be passed into the query */
	if (newslot)
	{
		ri_ExtractValues(source_rel, newslot, riinfo, source_is_pk,
						 vals, nulls);
		if (oldslot)
			ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
							 vals + riinfo->nkeys, nulls + riinfo->nkeys);
	}
	else
	{
		ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
						 vals, nulls);
	}

	/*
	 * In READ COMMITTED mode, we just need to use an up-to-date regular
	 * snapshot, and we will see all rows that could be interesting. But in
	 * transaction-snapshot mode, we can't change the transaction snapshot. If
	 * the caller passes detectNewRows == false then it's okay to do the query
	 * with the transaction snapshot; otherwise we use a current snapshot, and
	 * tell the executor to error out if it finds any rows under the current
	 * snapshot that wouldn't be visible per the transaction snapshot.  Note
	 * that SPI_execute_snapshot will register the snapshots, so we don't need
	 * to bother here.
	 */
	if (IsolationUsesXactSnapshot() && detectNewRows)
	{
		CommandCounterIncrement();	/* be sure all my own work is visible */
		test_snapshot = GetLatestSnapshot();
		crosscheck_snapshot = GetTransactionSnapshot();
	}
	else
	{
		/* the default SPI behavior is okay */
		test_snapshot = InvalidSnapshot;
		crosscheck_snapshot = InvalidSnapshot;
	}

	/*
	 * If this is a select query (e.g., for a 'no action' or 'restrict'
	 * trigger), we only need to see if there is a single row in the table,
	 * matching the key.  Otherwise, limit = 0 - because we want the query to
	 * affect ALL the matching rows.
	 */
	limit = (expect_OK == SPI_OK_SELECT) ? 1 : 0;

	/* Switch to proper UID to perform check as */
	GetUserIdAndSecContext(&save_userid, &save_sec_context);
	SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
						   SECURITY_NOFORCE_RLS);

	/*
	 * Finally we can run the query.
	 *
	 * Set fire_triggers to false to ensure that AFTER triggers are queued in
	 * the outer query's after-trigger context and fire after all RI updates
	 * on the same row are complete, rather than immediately.
	 */
	spi_result = SPI_execute_snapshot(qplan,
									  vals, nulls,
									  test_snapshot, crosscheck_snapshot,
									  false, false, limit);

	/* Restore UID and security context */
	SetUserIdAndSecContext(save_userid, save_sec_context);

	/* Check result */
	if (spi_result < 0)
		elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));

	if (expect_OK >= 0 && spi_result != expect_OK)
		ereport(ERROR,
				(errcode(ERRCODE_INTERNAL_ERROR),
				 errmsg("referential integrity query on \"%s\" from constraint \"%s\" on \"%s\" gave unexpected result",
						RelationGetRelationName(pk_rel),
						NameStr(riinfo->conname),
						RelationGetRelationName(fk_rel)),
				 errhint("This is most likely due to a rule having rewritten the query.")));

	/* XXX wouldn't it be clearer to do this part at the caller? */
	if (qkey->constr_queryno != RI_PLAN_CHECK_LOOKUPPK_FROM_PK &&
		expect_OK == SPI_OK_SELECT &&
		(SPI_processed == 0) == (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK))
		ri_ReportViolation(riinfo,
						   pk_rel, fk_rel,
						   newslot ? newslot : oldslot,
						   NULL,
						   qkey->constr_queryno, is_restrict, false);

	return SPI_processed != 0;
}

/*
 * ri_FastPathCheck
 *		Perform per row FK existence check via direct index probe,
 *		bypassing SPI.
 *
 * If no matching PK row exists, report the violation via ri_ReportViolation(),
 * otherwise, the function returns normally.
 */
static void
ri_FastPathCheck(RI_ConstraintInfo *riinfo,
				 Relation fk_rel, TupleTableSlot *newslot)
{
	Relation	pk_rel;
	Relation	idx_rel;
	IndexScanDesc scandesc;
	TupleTableSlot *slot;
	Datum		pk_vals[INDEX_MAX_KEYS];
	char		pk_nulls[INDEX_MAX_KEYS];
	ScanKeyData skey[INDEX_MAX_KEYS];
	bool		found = false;
	Oid			saved_userid;
	int			saved_sec_context;
	Snapshot	snapshot;

	/*
	 * Advance the command counter so the snapshot sees the effects of prior
	 * triggers in this statement.  Mirrors what the SPI path does in
	 * ri_PerformCheck().
	 */
	CommandCounterIncrement();
	snapshot = RegisterSnapshot(GetTransactionSnapshot());

	pk_rel = table_open(riinfo->pk_relid, RowShareLock);
	idx_rel = index_open(riinfo->conindid, AccessShareLock);

	slot = table_slot_create(pk_rel, NULL);

	GetUserIdAndSecContext(&saved_userid, &saved_sec_context);
	SetUserIdAndSecContext(RelationGetForm(pk_rel)->relowner,
						   saved_sec_context |
						   SECURITY_LOCAL_USERID_CHANGE |
						   SECURITY_NOFORCE_RLS);
	ri_CheckPermissions(pk_rel);

	/*
	 * Begin the scan under the switched user id, so that any access method
	 * code invoked by index_beginscan() runs as the PK relation's owner.  For
	 * btree this has no functional consequence, but it keeps the ordering
	 * correct for out-of-tree access methods.
	 */
	scandesc = index_beginscan(pk_rel, idx_rel,
							   snapshot, NULL,
							   riinfo->nkeys, 0,
							   SO_NONE);

	if (riinfo->fpmeta == NULL)
	{
		/* Reload to ensure it's valid. */
		riinfo = ri_LoadConstraintInfo(riinfo->constraint_id);
		ri_populate_fastpath_metadata(riinfo, fk_rel, idx_rel);
	}
	Assert(riinfo->fpmeta);
	ri_ExtractValues(fk_rel, newslot, riinfo, false, pk_vals, pk_nulls);
	build_index_scankeys(riinfo, idx_rel, pk_vals, pk_nulls, skey);
	found = ri_FastPathProbeOne(pk_rel, idx_rel, scandesc, slot,
								snapshot, riinfo, skey, riinfo->nkeys);
	SetUserIdAndSecContext(saved_userid, saved_sec_context);
	index_endscan(scandesc);
	ExecDropSingleTupleTableSlot(slot);
	UnregisterSnapshot(snapshot);

	if (!found)
		ri_ReportViolation(riinfo, pk_rel, fk_rel,
						   newslot, NULL,
						   RI_PLAN_CHECK_LOOKUPPK, false, false);

	index_close(idx_rel, NoLock);
	table_close(pk_rel, NoLock);
}

/*
 * ri_FastPathBatchAdd
 *		Buffer a FK row for batched probing.
 *
 * Adds the row to the batch buffer.  When the buffer is full, flushes all
 * buffered rows by probing the PK index.  Any violation is reported
 * immediately during the flush via ri_ReportViolation (which does not return).
 *
 * Uses the per-batch cache (RI_FastPathEntry) to avoid per-row relation
 * open/close, slot creation, etc.
 *
 * The batch is also flushed at end of trigger-firing cycle via
 * ri_FastPathEndBatch().
 */
static void
ri_FastPathBatchAdd(RI_ConstraintInfo *riinfo,
					Relation fk_rel, TupleTableSlot *newslot)
{
	RI_FastPathEntry *fpentry = ri_FastPathGetEntry(riinfo, fk_rel);

	/*
	 * If this entry is already being flushed, a cast function or an operator
	 * invoked during the flush has re-entered with DML on the same FK.  Fall
	 * back to the per-row path rather than touching the batch array, which is
	 * mid-flush.
	 */
	if (unlikely(fpentry->flushing))
	{
		ri_FastPathCheck(riinfo, fk_rel, newslot);
		return;
	}

	/*
	 * Buffer the row.  A full batch is flushed below and re-entry is handled
	 * above, so there is always room here; the bounds check just guards the
	 * array write.
	 */
	if (fpentry->batch_count < RI_FASTPATH_BATCH_SIZE)
	{
		MemoryContext oldcxt = MemoryContextSwitchTo(fpentry->flush_cxt);

		fpentry->batch[fpentry->batch_count] =
			ExecCopySlotHeapTuple(newslot);
		fpentry->batch_count++;
		MemoryContextSwitchTo(oldcxt);
	}
	else
		elog(ERROR, "RI fast-path batch unexpectedly full");

	/* Flush as soon as the batch is full. */
	if (fpentry->batch_count == RI_FASTPATH_BATCH_SIZE)
		ri_FastPathBatchFlush(fpentry, fk_rel, riinfo);
}

/*
 * ri_FastPathBatchFlush
 *		Flush all buffered FK rows by probing the PK index.
 *
 * Dispatches to ri_FastPathFlushArray() for single-column FKs
 * (using SK_SEARCHARRAY) or ri_FastPathFlushLoop() for multi-column
 * FKs (per-row probing).  Violations are reported immediately via
 * ri_ReportViolation(), which does not return.
 */
static void
ri_FastPathBatchFlush(RI_FastPathEntry *fpentry, Relation fk_rel,
					  RI_ConstraintInfo *riinfo)
{
	Relation	pk_rel = fpentry->pk_rel;
	Relation	idx_rel = fpentry->idx_rel;
	TupleTableSlot *fk_slot = fpentry->fk_slot;
	Snapshot	snapshot;
	IndexScanDesc scandesc;
	Oid			saved_userid;
	int			saved_sec_context;
	MemoryContext oldcxt;
	int			violation_index;

	if (fpentry->batch_count == 0)
		return;

	/*
	 * CCI and security context switch are done once for the entire batch.
	 * Per-row CCI is unnecessary because by the time a flush runs, all AFTER
	 * triggers for the buffered rows have already fired (trigger invocations
	 * strictly alternate per row), so a single CCI advances past all their
	 * effects.  Per-row security context switch is unnecessary because each
	 * row's probe runs entirely as the PK table owner, same as the SPI path
	 * -- the only difference is that the SPI path sets and restores the
	 * context per row whereas we do it once around the whole batch.
	 */
	CommandCounterIncrement();
	snapshot = RegisterSnapshot(GetTransactionSnapshot());

	/*
	 * build_index_scankeys() may palloc cast results for cross-type FKs. Use
	 * the entry's short-lived flush context so these don't accumulate across
	 * batches.
	 */
	oldcxt = MemoryContextSwitchTo(fpentry->flush_cxt);

	GetUserIdAndSecContext(&saved_userid, &saved_sec_context);
	SetUserIdAndSecContext(RelationGetForm(pk_rel)->relowner,
						   saved_sec_context |
						   SECURITY_LOCAL_USERID_CHANGE |
						   SECURITY_NOFORCE_RLS);

	/*
	 * Check that the current user has permission to access pk_rel. Done here
	 * rather than at entry creation so that permission changes between
	 * flushes are respected, matching the per-row behavior of the SPI path,
	 * albeit checked once per flush rather than once per row, like in
	 * ri_FastPathCheck().
	 */
	ri_CheckPermissions(pk_rel);

	/*
	 * Begin the scan under the switched user id, so that any access method
	 * code invoked by index_beginscan() runs as the PK relation's owner.  For
	 * btree this has no functional consequence, but it keeps the ordering
	 * correct for out-of-tree access methods.
	 */
	scandesc = index_beginscan(pk_rel, idx_rel, snapshot, NULL,
							   riinfo->nkeys, 0, SO_NONE);

	if (riinfo->fpmeta == NULL)
	{
		/* Reload to ensure it's valid. */
		riinfo = ri_LoadConstraintInfo(riinfo->constraint_id);
		ri_populate_fastpath_metadata(riinfo, fk_rel, idx_rel);
	}
	Assert(riinfo->fpmeta);

	/*
	 * The probe runs user-defined cast and equality functions.  Set the
	 * flushing flag around it so a re-entrant ri_FastPathBatchAdd on this
	 * entry takes the per-row path, and clear it even on error so the entry
	 * is reusable if the error is caught by a savepoint.
	 */
	Assert(!fpentry->flushing);
	fpentry->flushing = true;
	PG_TRY();
	{
		/* Skip array overhead for single-row batches. */
		if (riinfo->nkeys == 1 && fpentry->batch_count > 1)
			violation_index = ri_FastPathFlushArray(fpentry, fk_slot, riinfo,
													fk_rel, snapshot, scandesc);
		else
			violation_index = ri_FastPathFlushLoop(fpentry, fk_slot, riinfo,
												   fk_rel, snapshot, scandesc);
	}
	PG_FINALLY();
	{
		fpentry->flushing = false;
		fpentry->batch_count = 0;
	}
	PG_END_TRY();

	SetUserIdAndSecContext(saved_userid, saved_sec_context);
	UnregisterSnapshot(snapshot);
	index_endscan(scandesc);

	if (violation_index >= 0)
	{
		ExecStoreHeapTuple(fpentry->batch[violation_index], fk_slot, false);
		ri_ReportViolation(riinfo, pk_rel, fk_rel,
						   fk_slot, NULL,
						   RI_PLAN_CHECK_LOOKUPPK, false, false);
	}

	MemoryContextReset(fpentry->flush_cxt);
	MemoryContextSwitchTo(oldcxt);
}

/*
 * ri_FastPathFlushLoop
 *		Multi-column fallback: probe the index once per buffered row.
 *
 * Used for composite foreign keys where SK_SEARCHARRAY does not
 * apply, and also for single-row batches of single-column FKs where
 * the array overhead is not worth it.
 *
 * Returns the index of the first violating row in the batch array, or -1 if
 * all rows are valid.
 */
static int
ri_FastPathFlushLoop(RI_FastPathEntry *fpentry, TupleTableSlot *fk_slot,
					 const RI_ConstraintInfo *riinfo, Relation fk_rel,
					 Snapshot snapshot, IndexScanDesc scandesc)
{
	Relation	pk_rel = fpentry->pk_rel;
	Relation	idx_rel = fpentry->idx_rel;
	TupleTableSlot *pk_slot = fpentry->pk_slot;
	Datum		pk_vals[INDEX_MAX_KEYS];
	char		pk_nulls[INDEX_MAX_KEYS];
	ScanKeyData skey[INDEX_MAX_KEYS];
	bool		found = true;

	for (int i = 0; i < fpentry->batch_count; i++)
	{
		ExecStoreHeapTuple(fpentry->batch[i], fk_slot, false);
		ri_ExtractValues(fk_rel, fk_slot, riinfo, false, pk_vals, pk_nulls);
		build_index_scankeys(riinfo, idx_rel, pk_vals, pk_nulls, skey);

		found = ri_FastPathProbeOne(pk_rel, idx_rel, scandesc, pk_slot,
									snapshot, riinfo, skey, riinfo->nkeys);

		/* Report first unmatched row */
		if (!found)
			return i;
	}

	/* All pass. */
	return -1;
}

/*
 * ri_FastPathFlushArray
 *		Single-column fast path using SK_SEARCHARRAY.
 *
 * Builds an array of FK values and does one index scan with
 * SK_SEARCHARRAY.  The index AM sorts and deduplicates the array
 * internally, then walks matching leaf pages in order.  Each
 * matched PK tuple is locked and rechecked as before; a matched[]
 * bitmap tracks which batch items were satisfied.
 *
 * Returns the index of the first violating row in the batch array, or -1 if
 * all rows are valid.
 */
static int
ri_FastPathFlushArray(RI_FastPathEntry *fpentry, TupleTableSlot *fk_slot,
					  const RI_ConstraintInfo *riinfo, Relation fk_rel,
					  Snapshot snapshot, IndexScanDesc scandesc)
{
	FastPathMeta *fpmeta = riinfo->fpmeta;
	Relation	pk_rel = fpentry->pk_rel;
	Relation	idx_rel = fpentry->idx_rel;
	TupleTableSlot *pk_slot = fpentry->pk_slot;
	Datum		search_vals[RI_FASTPATH_BATCH_SIZE];
	bool		matched[RI_FASTPATH_BATCH_SIZE];
	int			nvals = fpentry->batch_count;
	Datum		pk_vals[INDEX_MAX_KEYS];
	char		pk_nulls[INDEX_MAX_KEYS];
	ScanKeyData skey[1];
	FmgrInfo   *cast_func_finfo;
	FmgrInfo   *eq_opr_finfo;
	Oid			elem_type;
	int16		elem_len;
	bool		elem_byval;
	char		elem_align;
	ArrayType  *arr;

	Assert(fpmeta);

	memset(matched, 0, nvals * sizeof(bool));

	/*
	 * Extract FK values, casting to the operator's expected input type if
	 * needed (e.g. int8 FK -> int4 for int48eq).
	 */
	cast_func_finfo = &fpmeta->cast_func_finfo[0];
	eq_opr_finfo = &fpmeta->eq_opr_finfo[0];
	for (int i = 0; i < nvals; i++)
	{
		ExecStoreHeapTuple(fpentry->batch[i], fk_slot, false);
		ri_ExtractValues(fk_rel, fk_slot, riinfo, false, pk_vals, pk_nulls);

		/* Cast if needed (e.g. int8 FK -> numeric PK) */
		if (OidIsValid(cast_func_finfo->fn_oid))
			search_vals[i] = FunctionCall3(cast_func_finfo,
										   pk_vals[0],
										   Int32GetDatum(-1),
										   BoolGetDatum(false));
		else
			search_vals[i] = pk_vals[0];
	}

	/*
	 * Array element type must match the operator's right-hand input type,
	 * which is what the index comparison expects on the search side.
	 * ri_populate_fastpath_metadata() stores exactly this via
	 * get_op_opfamily_properties(), which returns the operator's right-hand
	 * type as the subtype for cross-type operators (e.g. int8 for int48eq)
	 * and the common type for same-type operators.
	 */
	elem_type = fpmeta->subtypes[0];
	Assert(OidIsValid(elem_type));
	get_typlenbyvalalign(elem_type, &elem_len, &elem_byval, &elem_align);

	arr = construct_array(search_vals, nvals,
						  elem_type, elem_len, elem_byval, elem_align);

	/*
	 * Build scan key with SK_SEARCHARRAY.  The index AM code will internally
	 * sort and deduplicate, then walk leaf pages in order.
	 *
	 * ri_fastpath_is_applicable() restricts the fast path to btree indexes,
	 * which support SK_SEARCHARRAY.
	 *
	 * This path handles single-column FKs only, so index_attnos[0] == 1.
	 */
	Assert(idx_rel->rd_indam->amsearcharray);
	Assert(fpmeta->index_attnos[0] == 1);
	ScanKeyEntryInitialize(&skey[0],
						   SK_SEARCHARRAY,
						   fpmeta->index_attnos[0],
						   fpmeta->strats[0],
						   fpmeta->subtypes[0],
						   idx_rel->rd_indcollation[fpmeta->index_attnos[0] - 1],
						   fpmeta->regops[0],
						   PointerGetDatum(arr));

	index_rescan(scandesc, skey, 1, NULL, 0);

	/*
	 * Walk all matches.  The index AM returns them in index order.  For each
	 * match, find which batch item(s) it satisfies.
	 */
	while (index_getnext_slot(scandesc, ForwardScanDirection, pk_slot))
	{
		Datum		found_val;
		bool		found_null;
		bool		concurrently_updated;
		ScanKeyData recheck_skey[1];

		if (!ri_LockPKTuple(pk_rel, pk_slot, snapshot, &concurrently_updated))
			continue;

		/*
		 * Extract the PK value from the matched and locked tuple.
		 *
		 * A foreign key may reference a nullable unique column, not just a
		 * NOT NULL primary key.  If ri_LockPKTuple() chased an update chain
		 * to a version whose referenced key is now NULL, that version cannot
		 * equal any buffered (non-null) FK value, so skip it.  This mirrors
		 * the SPI path, where the requalifying "pkatt = $n" yields NULL and
		 * the row is not returned.
		 */
		found_val = slot_getattr(pk_slot, riinfo->pk_attnums[0], &found_null);
		if (found_null)
			continue;

		if (concurrently_updated)
		{
			/*
			 * Build a single-key scankey for recheck.  We need the actual PK
			 * value that was found, not the FK search value.
			 */
			ScanKeyEntryInitialize(&recheck_skey[0], 0, 1,
								   fpmeta->strats[0],
								   fpmeta->subtypes[0],
								   idx_rel->rd_indcollation[0],
								   fpmeta->regops[0],
								   found_val);
			if (!recheck_matched_pk_tuple(idx_rel, recheck_skey, 1, pk_slot))
				continue;
		}

		/*
		 * Linear scan to mark all batch items matching this PK value.
		 * O(batch_size) per match, O(batch_size^2) worst case -- fine for the
		 * current batch size of 64.
		 */
		for (int i = 0; i < nvals; i++)
		{
			if (!matched[i] &&
				DatumGetBool(FunctionCall2Coll(eq_opr_finfo,
											   idx_rel->rd_indcollation[0],
											   found_val,
											   search_vals[i])))
				matched[i] = true;
		}
	}

	/* Report first unmatched row */
	for (int i = 0; i < nvals; i++)
		if (!matched[i])
			return i;

	/* All pass. */
	return -1;
}

/*
 * ri_FastPathProbeOne
 *		Probe the PK index for one set of scan keys, lock the matching
 *		tuple
 *
 * Returns true if a matching PK row was found, locked, and (if
 * applicable) visible to the transaction snapshot.
 */
static bool
ri_FastPathProbeOne(Relation pk_rel, Relation idx_rel,
					IndexScanDesc scandesc, TupleTableSlot *slot,
					Snapshot snapshot, const RI_ConstraintInfo *riinfo,
					ScanKeyData *skey, int nkeys)
{
	bool		found = false;

	index_rescan(scandesc, skey, nkeys, NULL, 0);

	if (index_getnext_slot(scandesc, ForwardScanDirection, slot))
	{
		bool		concurrently_updated;

		if (ri_LockPKTuple(pk_rel, slot, snapshot,
						   &concurrently_updated))
		{
			if (concurrently_updated)
				found = recheck_matched_pk_tuple(idx_rel, skey, nkeys, slot);
			else
				found = true;
		}
	}

	return found;
}

/*
 * ri_LockPKTuple
 *		Lock a PK tuple found by the fast-path index scan.
 *
 * Calls table_tuple_lock() directly with handling specific to RI checks.
 * Returns true if the tuple was successfully locked.
 *
 * Sets *concurrently_updated to true if the locked tuple was reached
 * by following an update chain (tmfd.traversed), indicating the caller
 * should recheck the key.
 */
static bool
ri_LockPKTuple(Relation pk_rel, TupleTableSlot *slot, Snapshot snap,
			   bool *concurrently_updated)
{
	TM_FailureData tmfd;
	TM_Result	result;
	int			lockflags = TUPLE_LOCK_FLAG_LOCK_UPDATE_IN_PROGRESS;

	*concurrently_updated = false;

	if (!IsolationUsesXactSnapshot())
		lockflags |= TUPLE_LOCK_FLAG_FIND_LAST_VERSION;

	result = table_tuple_lock(pk_rel, &slot->tts_tid, snap,
							  slot, GetCurrentCommandId(false),
							  LockTupleKeyShare, LockWaitBlock,
							  lockflags, &tmfd);

	switch (result)
	{
		case TM_Ok:
			if (tmfd.traversed)
				*concurrently_updated = true;
			return true;

		case TM_Deleted:
			if (IsolationUsesXactSnapshot())
				ereport(ERROR,
						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
						 errmsg("could not serialize access due to concurrent delete")));
			return false;

		case TM_Updated:
			if (IsolationUsesXactSnapshot())
				ereport(ERROR,
						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
						 errmsg("could not serialize access due to concurrent update")));

			/*
			 * In READ COMMITTED, FIND_LAST_VERSION should have chased the
			 * chain and returned TM_Ok.  Getting here means something
			 * unexpected -- fall through to error.
			 */
			elog(ERROR, "unexpected table_tuple_lock status: %u", result);
			break;

		case TM_SelfModified:

			/*
			 * The current command or a later command in this transaction
			 * modified the PK row.  This shouldn't normally happen during an
			 * FK check (we're not modifying pk_rel), but handle it safely by
			 * treating the tuple as not found.
			 */
			return false;

		case TM_Invisible:
			elog(ERROR, "attempted to lock invisible tuple");
			break;

		default:
			elog(ERROR, "unrecognized table_tuple_lock status: %u", result);
			break;
	}

	return false;				/* keep compiler quiet */
}

static bool
ri_fastpath_is_applicable(const RI_ConstraintInfo *riinfo)
{
	/*
	 * Partitioned referenced tables are skipped for simplicity, since they
	 * require routing the probe through the correct partition using
	 * PartitionDirectory.
	 */
	if (riinfo->pk_is_partitioned)
		return false;

	/*
	 * Temporal foreign keys use range overlap and containment semantics (&&,
	 * <@, range_agg()) that inherently involve aggregation and multiple-row
	 * reasoning, so they stay on the SPI path.
	 */
	if (riinfo->hasperiod)
		return false;

	/*
	 * The fast path probes the referenced index directly and, for
	 * single-column keys, uses SK_SEARCHARRAY.  A foreign key's referenced
	 * index need not be a primary key; transformFkeyCheckAttrs() accepts any
	 * unique index, so an out-of-tree amcanunique access method could reach
	 * here.  Restrict the fast path to btree, which is what the direct probe
	 * and SK_SEARCHARRAY assume; other access methods fall back to SPI.
	 */
	if (!riinfo->pk_index_is_btree)
		return false;

	return true;
}

/*
 * ri_CheckPermissions
 *   Check that the current user has permissions to look into the schema of
 *   and SELECT from 'query_rel'
 */
static void
ri_CheckPermissions(Relation query_rel)
{
	AclResult	aclresult;

	/* USAGE on schema. */
	aclresult = object_aclcheck(NamespaceRelationId,
								RelationGetNamespace(query_rel),
								GetUserId(), ACL_USAGE);
	if (aclresult != ACLCHECK_OK)
		aclcheck_error(aclresult, OBJECT_SCHEMA,
					   get_namespace_name(RelationGetNamespace(query_rel)));

	/* SELECT on relation. */
	aclresult = pg_class_aclcheck(RelationGetRelid(query_rel), GetUserId(),
								  ACL_SELECT);
	if (aclresult != ACLCHECK_OK)
		aclcheck_error(aclresult, OBJECT_TABLE,
					   RelationGetRelationName(query_rel));
}

/*
 * recheck_matched_pk_tuple
 *		After following an update chain (tmfd.traversed), verify that
 *		the locked PK tuple still matches the original search keys.
 *
 * A non-key update (e.g. changing a non-PK column) creates a new tuple version
 * that we've now locked, but the key is unchanged -- that's fine.  A key
 * update means the value we were looking for is gone, so we should treat it as
 * not found.
 */
static bool
recheck_matched_pk_tuple(Relation idxrel, ScanKeyData *skeys, int nkeys,
						 TupleTableSlot *new_slot)
{
	/*
	 * TODO: BuildIndexInfo does a syscache lookup + palloc on every call.
	 * This only fires on the concurrent-update path (tmfd.traversed), which
	 * should be rare, so the cost is acceptable for now.  If profiling shows
	 * otherwise, cache the IndexInfo in FastPathMeta.
	 */
	IndexInfo  *indexInfo = BuildIndexInfo(idxrel);
	Datum		values[INDEX_MAX_KEYS];
	bool		isnull[INDEX_MAX_KEYS];
	bool		matched = true;

	/* PK indexes never have these. */
	Assert(indexInfo->ii_Expressions == NIL &&
		   indexInfo->ii_ExclusionOps == NULL);

	/* Form the index values and isnull flags given the table tuple. */
	Assert(nkeys == indexInfo->ii_NumIndexKeyAttrs);
	FormIndexDatum(indexInfo, new_slot, NULL, values, isnull);
	for (int i = 0; i < nkeys; i++)
	{
		ScanKeyData *skey = &skeys[i];

		/*
		 * A foreign key may reference a nullable unique column, so the
		 * version we chased the update chain to may have a NULL in a key
		 * column.  A NULL never equals the value we searched for, so treat it
		 * as no match, as the SPI path's requalification would.
		 */
		if (isnull[i] ||
			!DatumGetBool(FunctionCall2Coll(&skey->sk_func,
											skey->sk_collation,
											values[i],
											skey->sk_argument)))
		{
			matched = false;
			break;
		}
	}

	return matched;
}

/*
 * build_index_scankeys
 *		Build ScanKeys for a direct index probe of the PK's unique index.
 *
 * Uses cached compare entries, operator procedures, and strategy numbers
 * from ri_populate_fastpath_metadata() rather than looking them up on
 * each invocation.  Casts FK values to the operator's expected input
 * type if needed.
 */
static void
build_index_scankeys(const RI_ConstraintInfo *riinfo,
					 Relation idx_rel, Datum *pk_vals,
					 char *pk_nulls, ScanKey skeys)
{
	FastPathMeta *fpmeta = riinfo->fpmeta;

	Assert(fpmeta);

	/*
	 * May need to cast each of the individual values of the foreign key to
	 * the corresponding PK column's type if the equality operator demands it.
	 */
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		if (pk_nulls[i] != 'n' &&
			OidIsValid(fpmeta->cast_func_finfo[i].fn_oid))
			pk_vals[i] = FunctionCall3(&fpmeta->cast_func_finfo[i],
									   pk_vals[i],
									   Int32GetDatum(-1),	/* typmod */
									   BoolGetDatum(false));	/* implicit coercion */
	}

	/*
	 * Set up ScanKeys for the index scan. This is essentially how
	 * ExecIndexBuildScanKeys() sets them up.  Use the cached index_attnos and
	 * the corresponding collation since FK columns may be in a different
	 * order than PK index columns.  Place each scan key at the array position
	 * corresponding to its index column, since btree requires keys to be
	 * ordered by attribute number.
	 */
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		AttrNumber	pkattrno = fpmeta->index_attnos[i];
		int			skey_pos = pkattrno - 1;	/* 0-based array position */

		ScanKeyEntryInitialize(&skeys[skey_pos], 0, pkattrno,
							   fpmeta->strats[i], fpmeta->subtypes[i],
							   idx_rel->rd_indcollation[skey_pos], fpmeta->regops[i],
							   pk_vals[i]);
	}
}

/*
 * ri_populate_fastpath_metadata
 *		Cache per-key metadata needed by build_index_scankeys().
 *
 * Looks up the compare hash entry, operator procedure OID, and index
 * strategy/subtype for each key column.  Called lazily on first use
 * and persists for the lifetime of the RI_ConstraintInfo entry.
 */
static void
ri_populate_fastpath_metadata(RI_ConstraintInfo *riinfo,
							  Relation fk_rel, Relation idx_rel)
{
	FastPathMeta *fpmeta;
	MemoryContext oldcxt = MemoryContextSwitchTo(TopMemoryContext);

	Assert(riinfo != NULL && riinfo->valid);

	fpmeta = palloc_object(FastPathMeta);
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		Oid			eq_opr = riinfo->pf_eq_oprs[i];
		Oid			typeid = RIAttType(fk_rel, riinfo->fk_attnums[i]);
		Oid			lefttype;
		RI_CompareHashEntry *entry = ri_HashCompareOp(eq_opr, typeid);
		int			idx_col;

		/*
		 * Find the index column position for this constraint key.  The FK
		 * constraint may reference columns in a different order than they
		 * appear in the PK index, so we must map pk_attnums[i] to the
		 * corresponding index column position.
		 */
		for (idx_col = 0; idx_col < riinfo->nkeys; idx_col++)
		{
			if (idx_rel->rd_index->indkey.values[idx_col] == riinfo->pk_attnums[i])
				break;
		}
		Assert(idx_col < riinfo->nkeys);

		/* 1-based attribute number */
		fpmeta->index_attnos[i] = idx_col + 1;

		fmgr_info_copy(&fpmeta->cast_func_finfo[i], &entry->cast_func_finfo,
					   CurrentMemoryContext);
		fmgr_info_copy(&fpmeta->eq_opr_finfo[i], &entry->eq_opr_finfo,
					   CurrentMemoryContext);
		fpmeta->regops[i] = get_opcode(eq_opr);

		get_op_opfamily_properties(eq_opr,
								   idx_rel->rd_opfamily[idx_col],
								   false,
								   &fpmeta->strats[i],
								   &lefttype,
								   &fpmeta->subtypes[i]);
	}

	riinfo->fpmeta = fpmeta;
	MemoryContextSwitchTo(oldcxt);
}

/*
 * Extract fields from a tuple into Datum/nulls arrays
 */
static void
ri_ExtractValues(Relation rel, TupleTableSlot *slot,
				 const RI_ConstraintInfo *riinfo, bool rel_is_pk,
				 Datum *vals, char *nulls)
{
	const int16 *attnums;
	bool		isnull;

	if (rel_is_pk)
		attnums = riinfo->pk_attnums;
	else
		attnums = riinfo->fk_attnums;

	for (int i = 0; i < riinfo->nkeys; i++)
	{
		vals[i] = slot_getattr(slot, attnums[i], &isnull);
		nulls[i] = isnull ? 'n' : ' ';
	}
}

/*
 * Produce an error report
 *
 * If the failed constraint was on insert/update to the FK table,
 * we want the key names and values extracted from there, and the error
 * message to look like 'key blah is not present in PK'.
 * Otherwise, the attr names and values come from the PK table and the
 * message looks like 'key blah is still referenced from FK'.
 */
static void
ri_ReportViolation(const RI_ConstraintInfo *riinfo,
				   Relation pk_rel, Relation fk_rel,
				   TupleTableSlot *violatorslot, TupleDesc tupdesc,
				   int queryno, bool is_restrict, bool partgone)
{
	StringInfoData key_names;
	StringInfoData key_values;
	bool		onfk;
	const int16 *attnums;
	Oid			rel_oid;
	AclResult	aclresult;
	bool		has_perm = true;

	/*
	 * Determine which relation to complain about.  If tupdesc wasn't passed
	 * by caller, assume the violator tuple came from there.
	 */
	onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK);
	if (onfk)
	{
		attnums = riinfo->fk_attnums;
		rel_oid = fk_rel->rd_id;
		if (tupdesc == NULL)
			tupdesc = fk_rel->rd_att;
	}
	else
	{
		attnums = riinfo->pk_attnums;
		rel_oid = pk_rel->rd_id;
		if (tupdesc == NULL)
			tupdesc = pk_rel->rd_att;
	}

	/*
	 * Check permissions- if the user does not have access to view the data in
	 * any of the key columns then we don't include the errdetail() below.
	 *
	 * Check if RLS is enabled on the relation first.  If so, we don't return
	 * any specifics to avoid leaking data.
	 *
	 * Check table-level permissions next and, failing that, column-level
	 * privileges.
	 *
	 * When a partition at the referenced side is being detached/dropped, we
	 * needn't check, since the user must be the table owner anyway.
	 */
	if (partgone)
		has_perm = true;
	else if (check_enable_rls(rel_oid, InvalidOid, true) != RLS_ENABLED)
	{
		aclresult = pg_class_aclcheck(rel_oid, GetUserId(), ACL_SELECT);
		if (aclresult != ACLCHECK_OK)
		{
			/* Try for column-level permissions */
			for (int idx = 0; idx < riinfo->nkeys; idx++)
			{
				aclresult = pg_attribute_aclcheck(rel_oid, attnums[idx],
												  GetUserId(),
												  ACL_SELECT);

				/* No access to the key */
				if (aclresult != ACLCHECK_OK)
				{
					has_perm = false;
					break;
				}
			}
		}
	}
	else
		has_perm = false;

	if (has_perm)
	{
		/* Get printable versions of the keys involved */
		initStringInfo(&key_names);
		initStringInfo(&key_values);
		for (int idx = 0; idx < riinfo->nkeys; idx++)
		{
			int			fnum = attnums[idx];
			Form_pg_attribute att = TupleDescAttr(tupdesc, fnum - 1);
			char	   *name,
					   *val;
			Datum		datum;
			bool		isnull;

			name = NameStr(att->attname);

			datum = slot_getattr(violatorslot, fnum, &isnull);
			if (!isnull)
			{
				Oid			foutoid;
				bool		typisvarlena;

				getTypeOutputInfo(att->atttypid, &foutoid, &typisvarlena);
				val = OidOutputFunctionCall(foutoid, datum);
			}
			else
				val = "null";

			if (idx > 0)
			{
				appendStringInfoString(&key_names, ", ");
				appendStringInfoString(&key_values, ", ");
			}
			appendStringInfoString(&key_names, name);
			appendStringInfoString(&key_values, val);
		}
	}

	if (partgone)
		ereport(ERROR,
				(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
				 errmsg("removing partition \"%s\" violates foreign key constraint \"%s\"",
						RelationGetRelationName(pk_rel),
						NameStr(riinfo->conname)),
				 errdetail("Key (%s)=(%s) is still referenced from table \"%s\".",
						   key_names.data, key_values.data,
						   RelationGetRelationName(fk_rel)),
				 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
	else if (onfk)
		ereport(ERROR,
				(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
				 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
						RelationGetRelationName(fk_rel),
						NameStr(riinfo->conname)),
				 has_perm ?
				 errdetail("Key (%s)=(%s) is not present in table \"%s\".",
						   key_names.data, key_values.data,
						   RelationGetRelationName(pk_rel)) :
				 errdetail("Key is not present in table \"%s\".",
						   RelationGetRelationName(pk_rel)),
				 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
	else if (is_restrict)
		ereport(ERROR,
				(errcode(ERRCODE_RESTRICT_VIOLATION),
				 errmsg("update or delete on table \"%s\" violates RESTRICT setting of foreign key constraint \"%s\" on table \"%s\"",
						RelationGetRelationName(pk_rel),
						NameStr(riinfo->conname),
						RelationGetRelationName(fk_rel)),
				 has_perm ?
				 errdetail("Key (%s)=(%s) is referenced from table \"%s\".",
						   key_names.data, key_values.data,
						   RelationGetRelationName(fk_rel)) :
				 errdetail("Key is referenced from table \"%s\".",
						   RelationGetRelationName(fk_rel)),
				 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
	else
		ereport(ERROR,
				(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
				 errmsg("update or delete on table \"%s\" violates foreign key constraint \"%s\" on table \"%s\"",
						RelationGetRelationName(pk_rel),
						NameStr(riinfo->conname),
						RelationGetRelationName(fk_rel)),
				 has_perm ?
				 errdetail("Key (%s)=(%s) is still referenced from table \"%s\".",
						   key_names.data, key_values.data,
						   RelationGetRelationName(fk_rel)) :
				 errdetail("Key is still referenced from table \"%s\".",
						   RelationGetRelationName(fk_rel)),
				 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
}


/*
 * ri_NullCheck -
 *
 * Determine the NULL state of all key values in a tuple
 *
 * Returns one of RI_KEYS_ALL_NULL, RI_KEYS_NONE_NULL or RI_KEYS_SOME_NULL.
 */
static int
ri_NullCheck(TupleDesc tupDesc,
			 TupleTableSlot *slot,
			 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
{
	const int16 *attnums;
	bool		allnull = true;
	bool		nonenull = true;

	if (rel_is_pk)
		attnums = riinfo->pk_attnums;
	else
		attnums = riinfo->fk_attnums;

	for (int i = 0; i < riinfo->nkeys; i++)
	{
		if (slot_attisnull(slot, attnums[i]))
			nonenull = false;
		else
			allnull = false;
	}

	if (allnull)
		return RI_KEYS_ALL_NULL;

	if (nonenull)
		return RI_KEYS_NONE_NULL;

	return RI_KEYS_SOME_NULL;
}


/*
 * ri_InitHashTables -
 *
 * Initialize our internal hash tables.
 */
static void
ri_InitHashTables(void)
{
	HASHCTL		ctl;

	ctl.keysize = sizeof(Oid);
	ctl.entrysize = sizeof(RI_ConstraintInfo);
	ri_constraint_cache = hash_create("RI constraint cache",
									  RI_INIT_CONSTRAINTHASHSIZE,
									  &ctl, HASH_ELEM | HASH_BLOBS);

	/* Arrange to flush cache on pg_constraint changes */
	CacheRegisterSyscacheCallback(CONSTROID,
								  InvalidateConstraintCacheCallBack,
								  (Datum) 0);

	ctl.keysize = sizeof(RI_QueryKey);
	ctl.entrysize = sizeof(RI_QueryHashEntry);
	ri_query_cache = hash_create("RI query cache",
								 RI_INIT_QUERYHASHSIZE,
								 &ctl, HASH_ELEM | HASH_BLOBS);

	ctl.keysize = sizeof(RI_CompareKey);
	ctl.entrysize = sizeof(RI_CompareHashEntry);
	ri_compare_cache = hash_create("RI compare cache",
								   RI_INIT_QUERYHASHSIZE,
								   &ctl, HASH_ELEM | HASH_BLOBS);
}


/*
 * ri_FetchPreparedPlan -
 *
 * Lookup for a query key in our private hash table of prepared
 * and saved SPI execution plans. Return the plan if found or NULL.
 */
static SPIPlanPtr
ri_FetchPreparedPlan(RI_QueryKey *key)
{
	RI_QueryHashEntry *entry;
	SPIPlanPtr	plan;

	/*
	 * On the first call initialize the hashtable
	 */
	if (!ri_query_cache)
		ri_InitHashTables();

	/*
	 * Lookup for the key
	 */
	entry = (RI_QueryHashEntry *) hash_search(ri_query_cache,
											  key,
											  HASH_FIND, NULL);
	if (entry == NULL)
		return NULL;

	/*
	 * Check whether the plan is still valid.  If it isn't, we don't want to
	 * simply rely on plancache.c to regenerate it; rather we should start
	 * from scratch and rebuild the query text too.  This is to cover cases
	 * such as table/column renames.  We depend on the plancache machinery to
	 * detect possible invalidations, though.
	 *
	 * CAUTION: this check is only trustworthy if the caller has already
	 * locked both FK and PK rels.
	 */
	plan = entry->plan;
	if (plan && SPI_plan_is_valid(plan))
		return plan;

	/*
	 * Otherwise we might as well flush the cached plan now, to free a little
	 * memory space before we make a new one.
	 */
	entry->plan = NULL;
	if (plan)
		SPI_freeplan(plan);

	return NULL;
}


/*
 * ri_HashPreparedPlan -
 *
 * Add another plan to our private SPI query plan hashtable.
 */
static void
ri_HashPreparedPlan(RI_QueryKey *key, SPIPlanPtr plan)
{
	RI_QueryHashEntry *entry;
	bool		found;

	/*
	 * On the first call initialize the hashtable
	 */
	if (!ri_query_cache)
		ri_InitHashTables();

	/*
	 * Add the new plan.  We might be overwriting an entry previously found
	 * invalid by ri_FetchPreparedPlan.
	 */
	entry = (RI_QueryHashEntry *) hash_search(ri_query_cache,
											  key,
											  HASH_ENTER, &found);
	Assert(!found || entry->plan == NULL);
	entry->plan = plan;
}


/*
 * ri_KeysEqual -
 *
 * Check if all key values in OLD and NEW are "equivalent":
 * For normal FKs we check for equality.
 * For temporal FKs we check that the PK side is a superset of its old value,
 * or the FK side is a subset of its old value.
 *
 * Note: at some point we might wish to redefine this as checking for
 * "IS NOT DISTINCT" rather than "=", that is, allow two nulls to be
 * considered equal.  Currently there is no need since all callers have
 * previously found at least one of the rows to contain no nulls.
 */
static bool
ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
			 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
{
	const int16 *attnums;

	if (rel_is_pk)
		attnums = riinfo->pk_attnums;
	else
		attnums = riinfo->fk_attnums;

	/* XXX: could be worthwhile to fetch all necessary attrs at once */
	for (int i = 0; i < riinfo->nkeys; i++)
	{
		Datum		oldvalue;
		Datum		newvalue;
		bool		isnull;

		/*
		 * Get one attribute's oldvalue. If it is NULL - they're not equal.
		 */
		oldvalue = slot_getattr(oldslot, attnums[i], &isnull);
		if (isnull)
			return false;

		/*
		 * Get one attribute's newvalue. If it is NULL - they're not equal.
		 */
		newvalue = slot_getattr(newslot, attnums[i], &isnull);
		if (isnull)
			return false;

		if (rel_is_pk)
		{
			/*
			 * If we are looking at the PK table, then do a bytewise
			 * comparison.  We must propagate PK changes if the value is
			 * changed to one that "looks" different but would compare as
			 * equal using the equality operator.  This only makes a
			 * difference for ON UPDATE CASCADE, but for consistency we treat
			 * all changes to the PK the same.
			 */
			CompactAttribute *att = TupleDescCompactAttr(oldslot->tts_tupleDescriptor, attnums[i] - 1);

			if (!datum_image_eq(oldvalue, newvalue, att->attbyval, att->attlen))
				return false;
		}
		else
		{
			Oid			eq_opr;

			/*
			 * When comparing the PERIOD columns we can skip the check
			 * whenever the referencing column stayed equal or shrank, so test
			 * with the contained-by operator instead.
			 */
			if (riinfo->hasperiod && i == riinfo->nkeys - 1)
				eq_opr = riinfo->period_contained_by_oper;
			else
				eq_opr = riinfo->ff_eq_oprs[i];

			/*
			 * For the FK table, compare with the appropriate equality
			 * operator.  Changes that compare equal will still satisfy the
			 * constraint after the update.
			 */
			if (!ri_CompareWithCast(eq_opr, RIAttType(rel, attnums[i]), RIAttCollation(rel, attnums[i]),
									newvalue, oldvalue))
				return false;
		}
	}

	return true;
}


/*
 * ri_CompareWithCast -
 *
 * Call the appropriate comparison operator for two values.
 * Normally this is equality, but for the PERIOD part of foreign keys
 * it is ContainedBy, so the order of lhs vs rhs is significant.
 * See below for how the collation is applied.
 *
 * NB: we have already checked that neither value is null.
 */
static bool
ri_CompareWithCast(Oid eq_opr, Oid typeid, Oid collid,
				   Datum lhs, Datum rhs)
{
	RI_CompareHashEntry *entry = ri_HashCompareOp(eq_opr, typeid);

	/* Do we need to cast the values? */
	if (OidIsValid(entry->cast_func_finfo.fn_oid))
	{
		lhs = FunctionCall3(&entry->cast_func_finfo,
							lhs,
							Int32GetDatum(-1),	/* typmod */
							BoolGetDatum(false));	/* implicit coercion */
		rhs = FunctionCall3(&entry->cast_func_finfo,
							rhs,
							Int32GetDatum(-1),	/* typmod */
							BoolGetDatum(false));	/* implicit coercion */
	}

	/*
	 * Apply the comparison operator.
	 *
	 * Note: This function is part of a call stack that determines whether an
	 * update to a row is significant enough that it needs checking or action
	 * on the other side of a foreign-key constraint.  Therefore, the
	 * comparison here would need to be done with the collation of the *other*
	 * table.  For simplicity (e.g., we might not even have the other table
	 * open), we'll use our own collation.  This is fine because we require
	 * that both collations have the same notion of equality (either they are
	 * both deterministic or else they are both the same).
	 *
	 * With range/multirangetypes, the collation of the base type is stored as
	 * part of the rangetype (pg_range.rngcollation), and always used, so
	 * there is no danger of inconsistency even using a non-equals operator.
	 * But if we support arbitrary types with PERIOD, we should perhaps just
	 * always force a re-check.
	 */
	return DatumGetBool(FunctionCall2Coll(&entry->eq_opr_finfo, collid, lhs, rhs));
}

/*
 * ri_HashCompareOp -
 *
 * Look up or create a cache entry for the given equality operator and
 * the caller's value type (typeid).  The entry holds the operator's
 * FmgrInfo and, if typeid doesn't match what the operator expects as
 * its right-hand input, a cast function to coerce the value before
 * comparison.
 */
static RI_CompareHashEntry *
ri_HashCompareOp(Oid eq_opr, Oid typeid)
{
	RI_CompareKey key;
	RI_CompareHashEntry *entry;
	bool		found;

	/*
	 * On the first call initialize the hashtable
	 */
	if (!ri_compare_cache)
		ri_InitHashTables();

	/*
	 * Find or create a hash entry.  Note we're assuming RI_CompareKey
	 * contains no struct padding.
	 */
	key.eq_opr = eq_opr;
	key.typeid = typeid;
	entry = (RI_CompareHashEntry *) hash_search(ri_compare_cache,
												&key,
												HASH_ENTER, &found);
	if (!found)
		entry->valid = false;

	/*
	 * If not already initialized, do so.  Since we'll keep this hash entry
	 * for the life of the backend, put any subsidiary info for the function
	 * cache structs into TopMemoryContext.
	 */
	if (!entry->valid)
	{
		Oid			lefttype,
					righttype,
					castfunc;
		CoercionPathType pathtype;

		/* We always need to know how to call the equality operator */
		fmgr_info_cxt(get_opcode(eq_opr), &entry->eq_opr_finfo,
					  TopMemoryContext);

		/*
		 * If we chose to use a cast from FK to PK type, we may have to apply
		 * the cast function to get to the operator's input type.
		 *
		 * XXX eventually it would be good to support array-coercion cases
		 * here and in ri_CompareWithCast().  At the moment there is no point
		 * because cases involving nonidentical array types will be rejected
		 * at constraint creation time.
		 *
		 * XXX perhaps also consider supporting CoerceViaIO?  No need at the
		 * moment since that will never be generated for implicit coercions.
		 */
		op_input_types(eq_opr, &lefttype, &righttype);

		/*
		 * pf_eq_oprs (used by the fast path) can be cross-type when the FK
		 * and PK columns differ in type, e.g. int48eq for int4 PK / int8 FK.
		 * If the FK column's type, or the base type of a domain over it,
		 * already matches what the operator expects as its right-hand input,
		 * no cast is needed.
		 */
		if (getBaseType(typeid) == righttype)
			castfunc = InvalidOid;	/* simplest case */
		else
		{
			pathtype = find_coercion_pathway(lefttype, typeid,
											 COERCION_IMPLICIT,
											 &castfunc);
			if (pathtype != COERCION_PATH_FUNC &&
				pathtype != COERCION_PATH_RELABELTYPE)
			{
				/*
				 * The declared input type of the eq_opr might be a
				 * polymorphic type such as ANYARRAY or ANYENUM, or other
				 * special cases such as RECORD; find_coercion_pathway
				 * currently doesn't subsume these special cases.
				 */
				if (!IsBinaryCoercible(typeid, lefttype))
					elog(ERROR, "no conversion function from %s to %s",
						 format_type_be(typeid),
						 format_type_be(lefttype));
			}
		}
		if (OidIsValid(castfunc))
			fmgr_info_cxt(castfunc, &entry->cast_func_finfo,
						  TopMemoryContext);
		else
			entry->cast_func_finfo.fn_oid = InvalidOid;
		entry->valid = true;
	}

	return entry;
}


/*
 * Given a trigger function OID, determine whether it is an RI trigger,
 * and if so whether it is attached to PK or FK relation.
 */
int
RI_FKey_trigger_type(Oid tgfoid)
{
	switch (tgfoid)
	{
		case F_RI_FKEY_CASCADE_DEL:
		case F_RI_FKEY_CASCADE_UPD:
		case F_RI_FKEY_RESTRICT_DEL:
		case F_RI_FKEY_RESTRICT_UPD:
		case F_RI_FKEY_SETNULL_DEL:
		case F_RI_FKEY_SETNULL_UPD:
		case F_RI_FKEY_SETDEFAULT_DEL:
		case F_RI_FKEY_SETDEFAULT_UPD:
		case F_RI_FKEY_NOACTION_DEL:
		case F_RI_FKEY_NOACTION_UPD:
			return RI_TRIGGER_PK;

		case F_RI_FKEY_CHECK_INS:
		case F_RI_FKEY_CHECK_UPD:
			return RI_TRIGGER_FK;
	}

	return RI_TRIGGER_NONE;
}

/*
 * ri_FastPathEndBatch
 *		Flush remaining rows and tear down cached state.
 *
 * Registered as an AfterTriggerBatchCallback.  Note: the flush can
 * do real work (CCI, security context switch, index probes) and can
 * throw ERROR on a constraint violation.  If that happens,
 * ri_FastPathTeardown never runs; ResourceOwner releases the cached
 * relations and AtEOXact_RI() resets the static state on the abort path.
 */
static void
ri_FastPathEndBatch(void *arg)
{
	HASH_SEQ_STATUS status;
	RI_FastPathEntry *entry;

	if (ri_fastpath_cache == NULL)
		return;

	/*
	 * Set a flag for the duration of the scan so that any FK check triggered
	 * by user cast or operator code during a flush takes the per-row path
	 * instead of adding a new entry to the cache we are iterating.  A new
	 * entry could land in an already-scanned bucket and then be torn down
	 * unflushed below.
	 *
	 * The flush can throw ERROR (a reported constraint violation, or an error
	 * from the user code it runs).  In that case ri_FastPathTeardown below is
	 * skipped; the ResourceOwner and the transaction-end callback handle
	 * resource cleanup on the abort path.  The PG_FINALLY only resets the
	 * flag and deliberately does not attempt teardown.
	 */
	Assert(!ri_fastpath_flushing);
	ri_fastpath_flushing = true;
	PG_TRY();
	{
		hash_seq_init(&status, ri_fastpath_cache);
		while ((entry = hash_seq_search(&status)) != NULL)
		{
			if (entry->batch_count > 0)
			{
				Relation	fk_rel = table_open(entry->fk_relid, AccessShareLock);
				RI_ConstraintInfo *riinfo = ri_LoadConstraintInfo(entry->conoid);

				ri_FastPathBatchFlush(entry, fk_rel, riinfo);
				table_close(fk_rel, NoLock);
			}
		}
	}
	PG_FINALLY();
	{
		ri_fastpath_flushing = false;
	}
	PG_END_TRY();

	ri_FastPathTeardown();
}

/*
 * ri_FastPathTeardown
 *		Tear down all cached fast-path state.
 *
 * Called from ri_FastPathEndBatch() after flushing any remaining rows.
 */
static void
ri_FastPathTeardown(void)
{
	HASH_SEQ_STATUS status;
	RI_FastPathEntry *entry;

	if (ri_fastpath_cache == NULL)
		return;

	hash_seq_init(&status, ri_fastpath_cache);
	while ((entry = hash_seq_search(&status)) != NULL)
	{
		if (entry->idx_rel)
			index_close(entry->idx_rel, NoLock);
		if (entry->pk_rel)
			table_close(entry->pk_rel, NoLock);
		if (entry->pk_slot)
			ExecDropSingleTupleTableSlot(entry->pk_slot);
		if (entry->fk_slot)
			ExecDropSingleTupleTableSlot(entry->fk_slot);
		if (entry->flush_cxt)
			MemoryContextDelete(entry->flush_cxt);
	}

	hash_destroy(ri_fastpath_cache);
	ri_fastpath_cache = NULL;
	ri_fastpath_callback_registered = false;
}

/*
 * AtEOXact_RI
 *		Reset fast-path batching state at end of transaction.
 *
 * Called from CommitTransaction() and PrepareTransaction() with isCommit
 * true, and from AbortTransaction() with isCommit false.
 *
 * By the time we get here on a clean commit or prepare, the fast-path cache
 * has already been flushed and torn down by ri_FastPathEndBatch() (an
 * AfterTriggerBatchCallback fired from AfterTriggerFireDeferred(), well before
 * this point), so the static pointers are already clear and the reset below is
 * a no-op.  A surviving cache at commit means a trigger batch was never
 * flushed, which would have silently skipped FK checks, so we complain.
 *
 * On abort, ri_FastPathEndBatch()/ri_FastPathTeardown() may not have run (a
 * flush can error out partway): the ResourceOwner releases the cached
 * relations and the TopTransactionContext reset frees the cache memory, but
 * the process-local static pointers below would dangle into the next
 * transaction.  This resets them so they don't.
 *
 * The reset touches only backend-local static state (no relations, locks,
 * buffers or catalog access), so it has no ordering dependency on the
 * surrounding ResourceOwnerRelease() / AtEOXact_* steps.
 */
void
AtEOXact_RI(bool isCommit)
{
	/*
	 * The cache must be empty on a clean commit or prepare; a survivor means
	 * a trigger batch went unflushed.  Assert for assert-enabled builds and,
	 * since the transaction is already committed by now and FK checks may
	 * have been skipped, also warn in production builds.
	 */
	Assert(ri_fastpath_cache == NULL || !isCommit);
	if (isCommit && ri_fastpath_cache != NULL)
		elog(WARNING, "RI fast-path cache not flushed at end of transaction");

	/*
	 * Clear the static pointers/flags.  The cache memory lives in
	 * TopTransactionContext and is freed by the end-of-transaction
	 * memory-context reset; here we only drop the references to it.
	 */
	ri_fastpath_cache = NULL;
	ri_fastpath_callback_registered = false;

	/*
	 * Also clear the in-flush flag.  ri_FastPathEndBatch() already clears it
	 * via PG_FINALLY, so this is just defensive: it keeps a stale flag from
	 * surviving into the next transaction should any future path leave it
	 * set.
	 */
	ri_fastpath_flushing = false;
}

/*
 * ri_FastPathGetEntry
 *		Look up or create a per-batch cache entry for the given constraint.
 *
 * On first call for a constraint within a batch: opens pk_rel and the index,
 * allocates slots for both FK row and the looked up PK row, and registers the
 * cleanup callback.
 *
 * On subsequent calls: returns the existing entry.
 */
static RI_FastPathEntry *
ri_FastPathGetEntry(const RI_ConstraintInfo *riinfo, Relation fk_rel)
{
	RI_FastPathEntry *entry;
	bool		found;

	/* Create hash table on first use in this batch */
	if (ri_fastpath_cache == NULL)
	{
		HASHCTL		ctl;

		ctl.keysize = sizeof(Oid);
		ctl.entrysize = sizeof(RI_FastPathEntry);
		ctl.hcxt = TopTransactionContext;
		ri_fastpath_cache = hash_create("RI fast-path cache",
										16,
										&ctl,
										HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
	}

	entry = hash_search(ri_fastpath_cache, &riinfo->constraint_id,
						HASH_ENTER, &found);

	if (!found)
	{
		MemoryContext oldcxt;

		/*
		 * Zero out non-key fields so ri_FastPathTeardown is safe if we error
		 * out during partial initialization below.
		 */
		memset(((char *) entry) + offsetof(RI_FastPathEntry, pk_rel), 0,
			   sizeof(RI_FastPathEntry) - offsetof(RI_FastPathEntry, pk_rel));

		oldcxt = MemoryContextSwitchTo(TopTransactionContext);

		entry->fk_relid = RelationGetRelid(fk_rel);

		/*
		 * Open PK table and its unique index.
		 *
		 * RowShareLock on pk_rel matches what the SPI path's SELECT ... FOR
		 * KEY SHARE would acquire as a relation-level lock. AccessShareLock
		 * on the index is standard for index scans.
		 *
		 * We don't release these locks until end of transaction, matching SPI
		 * behavior.
		 */
		entry->pk_rel = table_open(riinfo->pk_relid, RowShareLock);
		entry->idx_rel = index_open(riinfo->conindid, AccessShareLock);
		entry->pk_slot = table_slot_create(entry->pk_rel, NULL);

		/*
		 * Must be TTSOpsHeapTuple because ExecStoreHeapTuple() is used to
		 * load entries from batch[] into this slot for value extraction.
		 */
		entry->fk_slot = MakeSingleTupleTableSlot(RelationGetDescr(fk_rel),
												  &TTSOpsHeapTuple);

		entry->flush_cxt = AllocSetContextCreate(TopTransactionContext,
												 "RI fast path flush temporary context",
												 ALLOCSET_SMALL_SIZES);
		MemoryContextSwitchTo(oldcxt);

		/* Ensure cleanup at end of this trigger-firing batch */
		if (!ri_fastpath_callback_registered)
		{
			RegisterAfterTriggerBatchCallback(ri_FastPathEndBatch, NULL);
			ri_fastpath_callback_registered = true;
		}

		entry->flushing = false;
		entry->batch_count = 0;
	}

	return entry;
}