?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Cache NO ACTION foreign keys separately from RESTRICT foreign keys
[pgsql.git] / src / backend / utils / adt / ri_triggers.c
blob8473448849cfce6d54117ce6cbf1f19537fe0040
1 /*-------------------------------------------------------------------------
2 *
3 * ri_triggers.c
4 *
5 * Generic trigger procedures for referential integrity constraint
6 * checks.
7 *
8 * Note about memory management: the private hashtables kept here live
9 * across query and transaction boundaries, in fact they live as long as
10 * the backend does. This works because the hashtable structures
11 * themselves are allocated by dynahash.c in its permanent DynaHashCxt,
12 * and the SPI plans they point to are saved using SPI_keepplan().
13 * There is not currently any provision for throwing away a no-longer-needed
14 * plan --- consider improving this someday.
15 *
16 *
17 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
18 *
19 * src/backend/utils/adt/ri_triggers.c
20 *
21 *-------------------------------------------------------------------------
22 */
23
24 #include "postgres.h"
25
26 #include "access/htup_details.h"
27 #include "access/sysattr.h"
28 #include "access/table.h"
29 #include "access/tableam.h"
30 #include "access/xact.h"
31 #include "catalog/pg_collation.h"
32 #include "catalog/pg_constraint.h"
33 #include "catalog/pg_proc.h"
34 #include "commands/trigger.h"
35 #include "executor/executor.h"
36 #include "executor/spi.h"
37 #include "lib/ilist.h"
38 #include "miscadmin.h"
39 #include "parser/parse_coerce.h"
40 #include "parser/parse_relation.h"
41 #include "utils/acl.h"
42 #include "utils/builtins.h"
43 #include "utils/datum.h"
44 #include "utils/fmgroids.h"
45 #include "utils/guc.h"
46 #include "utils/inval.h"
47 #include "utils/lsyscache.h"
48 #include "utils/memutils.h"
49 #include "utils/rangetypes.h"
50 #include "utils/rel.h"
51 #include "utils/rls.h"
52 #include "utils/ruleutils.h"
53 #include "utils/snapmgr.h"
54 #include "utils/syscache.h"
55
56 /*
57 * Local definitions
58 */
59
60 #define RI_MAX_NUMKEYS INDEX_MAX_KEYS
61
62 #define RI_INIT_CONSTRAINTHASHSIZE 64
63 #define RI_INIT_QUERYHASHSIZE (RI_INIT_CONSTRAINTHASHSIZE * 4)
64
65 #define RI_KEYS_ALL_NULL 0
66 #define RI_KEYS_SOME_NULL 1
67 #define RI_KEYS_NONE_NULL 2
68
69 /* RI query type codes */
70 /* these queries are executed against the PK (referenced) table: */
71 #define RI_PLAN_CHECK_LOOKUPPK 1
72 #define RI_PLAN_CHECK_LOOKUPPK_FROM_PK 2
73 #define RI_PLAN_LAST_ON_PK RI_PLAN_CHECK_LOOKUPPK_FROM_PK
74 /* these queries are executed against the FK (referencing) table: */
75 #define RI_PLAN_CASCADE_ONDELETE 3
76 #define RI_PLAN_CASCADE_ONUPDATE 4
77 #define RI_PLAN_NO_ACTION 5
78 /* For RESTRICT, the same plan can be used for both ON DELETE and ON UPDATE triggers. */
79 #define RI_PLAN_RESTRICT 6
80 #define RI_PLAN_SETNULL_ONDELETE 7
81 #define RI_PLAN_SETNULL_ONUPDATE 8
82 #define RI_PLAN_SETDEFAULT_ONDELETE 9
83 #define RI_PLAN_SETDEFAULT_ONUPDATE 10
84
85 #define MAX_QUOTED_NAME_LEN (NAMEDATALEN*2+3)
86 #define MAX_QUOTED_REL_NAME_LEN (MAX_QUOTED_NAME_LEN*2)
87
88 #define RIAttName(rel, attnum) NameStr(*attnumAttName(rel, attnum))
89 #define RIAttType(rel, attnum) attnumTypeId(rel, attnum)
90 #define RIAttCollation(rel, attnum) attnumCollationId(rel, attnum)
91
92 #define RI_TRIGTYPE_INSERT 1
93 #define RI_TRIGTYPE_UPDATE 2
94 #define RI_TRIGTYPE_DELETE 3
95
96
97 /*
98 * RI_ConstraintInfo
99 *
100 * Information extracted from an FK pg_constraint entry. This is cached in
101 * ri_constraint_cache.
102 *
103 * Note that pf/pp/ff_eq_oprs may hold the overlaps operator instead of equals
104 * for the PERIOD part of a temporal foreign key.
105 */
106 typedef struct RI_ConstraintInfo
107 {
108 Oid constraint_id; /* OID of pg_constraint entry (hash key) */
109 bool valid; /* successfully initialized? */
110 Oid constraint_root_id; /* OID of topmost ancestor constraint;
111 * same as constraint_id if not inherited */
112 uint32 oidHashValue; /* hash value of constraint_id */
113 uint32 rootHashValue; /* hash value of constraint_root_id */
114 NameData conname; /* name of the FK constraint */
115 Oid pk_relid; /* referenced relation */
116 Oid fk_relid; /* referencing relation */
117 char confupdtype; /* foreign key's ON UPDATE action */
118 char confdeltype; /* foreign key's ON DELETE action */
119 int ndelsetcols; /* number of columns referenced in ON DELETE
120 * SET clause */
121 int16 confdelsetcols[RI_MAX_NUMKEYS]; /* attnums of cols to set on
122 * delete */
123 char confmatchtype; /* foreign key's match type */
124 bool hasperiod; /* if the foreign key uses PERIOD */
125 int nkeys; /* number of key columns */
126 int16 pk_attnums[RI_MAX_NUMKEYS]; /* attnums of referenced cols */
127 int16 fk_attnums[RI_MAX_NUMKEYS]; /* attnums of referencing cols */
128 Oid pf_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = FK) */
129 Oid pp_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (PK = PK) */
130 Oid ff_eq_oprs[RI_MAX_NUMKEYS]; /* equality operators (FK = FK) */
131 Oid period_contained_by_oper; /* anyrange <@ anyrange */
132 Oid agged_period_contained_by_oper; /* fkattr <@ range_agg(pkattr) */
133 Oid period_intersect_oper; /* anyrange * anyrange */
134 dlist_node valid_link; /* Link in list of valid entries */
135 } RI_ConstraintInfo;
136
137 /*
138 * RI_QueryKey
139 *
140 * The key identifying a prepared SPI plan in our query hashtable
141 */
142 typedef struct RI_QueryKey
143 {
144 Oid constr_id; /* OID of pg_constraint entry */
145 int32 constr_queryno; /* query type ID, see RI_PLAN_XXX above */
146 } RI_QueryKey;
147
148 /*
149 * RI_QueryHashEntry
150 */
151 typedef struct RI_QueryHashEntry
152 {
153 RI_QueryKey key;
154 SPIPlanPtr plan;
155 } RI_QueryHashEntry;
156
157 /*
158 * RI_CompareKey
159 *
160 * The key identifying an entry showing how to compare two values
161 */
162 typedef struct RI_CompareKey
163 {
164 Oid eq_opr; /* the equality operator to apply */
165 Oid typeid; /* the data type to apply it to */
166 } RI_CompareKey;
167
168 /*
169 * RI_CompareHashEntry
170 */
171 typedef struct RI_CompareHashEntry
172 {
173 RI_CompareKey key;
174 bool valid; /* successfully initialized? */
175 FmgrInfo eq_opr_finfo; /* call info for equality fn */
176 FmgrInfo cast_func_finfo; /* in case we must coerce input */
177 } RI_CompareHashEntry;
178
179
180 /*
181 * Local data
182 */
183 static HTAB *ri_constraint_cache = NULL;
184 static HTAB *ri_query_cache = NULL;
185 static HTAB *ri_compare_cache = NULL;
186 static dclist_head ri_constraint_cache_valid_list;
187
188
189 /*
190 * Local function prototypes
191 */
192 static bool ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
193 TupleTableSlot *oldslot,
194 const RI_ConstraintInfo *riinfo);
195 static Datum ri_restrict(TriggerData *trigdata, bool is_no_action);
196 static Datum ri_set(TriggerData *trigdata, bool is_set_null, int tgkind);
197 static void quoteOneName(char *buffer, const char *name);
198 static void quoteRelationName(char *buffer, Relation rel);
199 static void ri_GenerateQual(StringInfo buf,
200 const char *sep,
201 const char *leftop, Oid leftoptype,
202 Oid opoid,
203 const char *rightop, Oid rightoptype);
204 static void ri_GenerateQualCollation(StringInfo buf, Oid collation);
205 static int ri_NullCheck(TupleDesc tupDesc, TupleTableSlot *slot,
206 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
207 static void ri_BuildQueryKey(RI_QueryKey *key,
208 const RI_ConstraintInfo *riinfo,
209 int32 constr_queryno);
210 static bool ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
211 const RI_ConstraintInfo *riinfo, bool rel_is_pk);
212 static bool ri_CompareWithCast(Oid eq_opr, Oid typeid, Oid collid,
213 Datum lhs, Datum rhs);
214
215 static void ri_InitHashTables(void);
216 static void InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue);
217 static SPIPlanPtr ri_FetchPreparedPlan(RI_QueryKey *key);
218 static void ri_HashPreparedPlan(RI_QueryKey *key, SPIPlanPtr plan);
219 static RI_CompareHashEntry *ri_HashCompareOp(Oid eq_opr, Oid typeid);
220
221 static void ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname,
222 int tgkind);
223 static const RI_ConstraintInfo *ri_FetchConstraintInfo(Trigger *trigger,
224 Relation trig_rel, bool rel_is_pk);
225 static const RI_ConstraintInfo *ri_LoadConstraintInfo(Oid constraintOid);
226 static Oid get_ri_constraint_root(Oid constrOid);
227 static SPIPlanPtr ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
228 RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel);
229 static bool ri_PerformCheck(const RI_ConstraintInfo *riinfo,
230 RI_QueryKey *qkey, SPIPlanPtr qplan,
231 Relation fk_rel, Relation pk_rel,
232 TupleTableSlot *oldslot, TupleTableSlot *newslot,
233 bool is_restrict,
234 bool detectNewRows, int expect_OK);
235 static void ri_ExtractValues(Relation rel, TupleTableSlot *slot,
236 const RI_ConstraintInfo *riinfo, bool rel_is_pk,
237 Datum *vals, char *nulls);
238 static void ri_ReportViolation(const RI_ConstraintInfo *riinfo,
239 Relation pk_rel, Relation fk_rel,
240 TupleTableSlot *violatorslot, TupleDesc tupdesc,
241 int queryno, bool is_restrict, bool partgone) pg_attribute_noreturn();
242
243
244 /*
245 * RI_FKey_check -
246 *
247 * Check foreign key existence (combined for INSERT and UPDATE).
248 */
249 static Datum
250 RI_FKey_check(TriggerData *trigdata)
251 {
252 const RI_ConstraintInfo *riinfo;
253 Relation fk_rel;
254 Relation pk_rel;
255 TupleTableSlot *newslot;
256 RI_QueryKey qkey;
257 SPIPlanPtr qplan;
258
259 riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
260 trigdata->tg_relation, false);
261
262 if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
263 newslot = trigdata->tg_newslot;
264 else
265 newslot = trigdata->tg_trigslot;
266
267 /*
268 * We should not even consider checking the row if it is no longer valid,
269 * since it was either deleted (so the deferred check should be skipped)
270 * or updated (in which case only the latest version of the row should be
271 * checked). Test its liveness according to SnapshotSelf. We need pin
272 * and lock on the buffer to call HeapTupleSatisfiesVisibility. Caller
273 * should be holding pin, but not lock.
274 */
275 if (!table_tuple_satisfies_snapshot(trigdata->tg_relation, newslot, SnapshotSelf))
276 return PointerGetDatum(NULL);
277
278 /*
279 * Get the relation descriptors of the FK and PK tables.
280 *
281 * pk_rel is opened in RowShareLock mode since that's what our eventual
282 * SELECT FOR KEY SHARE will get on it.
283 */
284 fk_rel = trigdata->tg_relation;
285 pk_rel = table_open(riinfo->pk_relid, RowShareLock);
286
287 switch (ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false))
288 {
289 case RI_KEYS_ALL_NULL:
290
291 /*
292 * No further check needed - an all-NULL key passes every type of
293 * foreign key constraint.
294 */
295 table_close(pk_rel, RowShareLock);
296 return PointerGetDatum(NULL);
297
298 case RI_KEYS_SOME_NULL:
299
300 /*
301 * This is the only case that differs between the three kinds of
302 * MATCH.
303 */
304 switch (riinfo->confmatchtype)
305 {
306 case FKCONSTR_MATCH_FULL:
307
308 /*
309 * Not allowed - MATCH FULL says either all or none of the
310 * attributes can be NULLs
311 */
312 ereport(ERROR,
313 (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
314 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
315 RelationGetRelationName(fk_rel),
316 NameStr(riinfo->conname)),
317 errdetail("MATCH FULL does not allow mixing of null and nonnull key values."),
318 errtableconstraint(fk_rel,
319 NameStr(riinfo->conname))));
320 table_close(pk_rel, RowShareLock);
321 return PointerGetDatum(NULL);
322
323 case FKCONSTR_MATCH_SIMPLE:
324
325 /*
326 * MATCH SIMPLE - if ANY column is null, the key passes
327 * the constraint.
328 */
329 table_close(pk_rel, RowShareLock);
330 return PointerGetDatum(NULL);
331
332 #ifdef NOT_USED
333 case FKCONSTR_MATCH_PARTIAL:
334
335 /*
336 * MATCH PARTIAL - all non-null columns must match. (not
337 * implemented, can be done by modifying the query below
338 * to only include non-null columns, or by writing a
339 * special version here)
340 */
341 break;
342 #endif
343 }
344
345 case RI_KEYS_NONE_NULL:
346
347 /*
348 * Have a full qualified key - continue below for all three kinds
349 * of MATCH.
350 */
351 break;
352 }
353
354 SPI_connect();
355
356 /* Fetch or prepare a saved plan for the real check */
357 ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK);
358
359 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
360 {
361 StringInfoData querybuf;
362 char pkrelname[MAX_QUOTED_REL_NAME_LEN];
363 char attname[MAX_QUOTED_NAME_LEN];
364 char paramname[16];
365 const char *querysep;
366 Oid queryoids[RI_MAX_NUMKEYS];
367 const char *pk_only;
368
369 /* ----------
370 * The query string built is
371 * SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
372 * FOR KEY SHARE OF x
373 * The type id's for the $ parameters are those of the
374 * corresponding FK attributes.
375 *
376 * But for temporal FKs we need to make sure
377 * the FK's range is completely covered.
378 * So we use this query instead:
379 * SELECT 1
380 * FROM (
381 * SELECT pkperiodatt AS r
382 * FROM [ONLY] pktable x
383 * WHERE pkatt1 = $1 [AND ...]
384 * AND pkperiodatt && $n
385 * FOR KEY SHARE OF x
386 * ) x1
387 * HAVING $n <@ range_agg(x1.r)
388 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
389 * we can make this a bit simpler.
390 * ----------
391 */
392 initStringInfo(&querybuf);
393 pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
394 "" : "ONLY ";
395 quoteRelationName(pkrelname, pk_rel);
396 if (riinfo->hasperiod)
397 {
398 quoteOneName(attname,
399 RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));
400
401 appendStringInfo(&querybuf,
402 "SELECT 1 FROM (SELECT %s AS r FROM %s%s x",
403 attname, pk_only, pkrelname);
404 }
405 else
406 {
407 appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
408 pk_only, pkrelname);
409 }
410 querysep = "WHERE";
411 for (int i = 0; i < riinfo->nkeys; i++)
412 {
413 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
414 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
415
416 quoteOneName(attname,
417 RIAttName(pk_rel, riinfo->pk_attnums[i]));
418 sprintf(paramname, "$%d", i + 1);
419 ri_GenerateQual(&querybuf, querysep,
420 attname, pk_type,
421 riinfo->pf_eq_oprs[i],
422 paramname, fk_type);
423 querysep = "AND";
424 queryoids[i] = fk_type;
425 }
426 appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
427 if (riinfo->hasperiod)
428 {
429 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
430
431 appendStringInfo(&querybuf, ") x1 HAVING ");
432 sprintf(paramname, "$%d", riinfo->nkeys);
433 ri_GenerateQual(&querybuf, "",
434 paramname, fk_type,
435 riinfo->agged_period_contained_by_oper,
436 "pg_catalog.range_agg", ANYMULTIRANGEOID);
437 appendStringInfo(&querybuf, "(x1.r)");
438 }
439
440 /* Prepare and save the plan */
441 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
442 &qkey, fk_rel, pk_rel);
443 }
444
445 /*
446 * Now check that foreign key exists in PK table
447 *
448 * XXX detectNewRows must be true when a partitioned table is on the
449 * referenced side. The reason is that our snapshot must be fresh in
450 * order for the hack in find_inheritance_children() to work.
451 */
452 ri_PerformCheck(riinfo, &qkey, qplan,
453 fk_rel, pk_rel,
454 NULL, newslot,
455 false,
456 pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE,
457 SPI_OK_SELECT);
458
459 if (SPI_finish() != SPI_OK_FINISH)
460 elog(ERROR, "SPI_finish failed");
461
462 table_close(pk_rel, RowShareLock);
463
464 return PointerGetDatum(NULL);
465 }
466
467
468 /*
469 * RI_FKey_check_ins -
470 *
471 * Check foreign key existence at insert event on FK table.
472 */
473 Datum
474 RI_FKey_check_ins(PG_FUNCTION_ARGS)
475 {
476 /* Check that this is a valid trigger call on the right time and event. */
477 ri_CheckTrigger(fcinfo, "RI_FKey_check_ins", RI_TRIGTYPE_INSERT);
478
479 /* Share code with UPDATE case. */
480 return RI_FKey_check((TriggerData *) fcinfo->context);
481 }
482
483
484 /*
485 * RI_FKey_check_upd -
486 *
487 * Check foreign key existence at update event on FK table.
488 */
489 Datum
490 RI_FKey_check_upd(PG_FUNCTION_ARGS)
491 {
492 /* Check that this is a valid trigger call on the right time and event. */
493 ri_CheckTrigger(fcinfo, "RI_FKey_check_upd", RI_TRIGTYPE_UPDATE);
494
495 /* Share code with INSERT case. */
496 return RI_FKey_check((TriggerData *) fcinfo->context);
497 }
498
499
500 /*
501 * ri_Check_Pk_Match
502 *
503 * Check to see if another PK row has been created that provides the same
504 * key values as the "oldslot" that's been modified or deleted in our trigger
505 * event. Returns true if a match is found in the PK table.
506 *
507 * We assume the caller checked that the oldslot contains no NULL key values,
508 * since otherwise a match is impossible.
509 */
510 static bool
511 ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
512 TupleTableSlot *oldslot,
513 const RI_ConstraintInfo *riinfo)
514 {
515 SPIPlanPtr qplan;
516 RI_QueryKey qkey;
517 bool result;
518
519 /* Only called for non-null rows */
520 Assert(ri_NullCheck(RelationGetDescr(pk_rel), oldslot, riinfo, true) == RI_KEYS_NONE_NULL);
521
522 SPI_connect();
523
524 /*
525 * Fetch or prepare a saved plan for checking PK table with values coming
526 * from a PK row
527 */
528 ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CHECK_LOOKUPPK_FROM_PK);
529
530 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
531 {
532 StringInfoData querybuf;
533 char pkrelname[MAX_QUOTED_REL_NAME_LEN];
534 char attname[MAX_QUOTED_NAME_LEN];
535 char paramname[16];
536 const char *querysep;
537 const char *pk_only;
538 Oid queryoids[RI_MAX_NUMKEYS];
539
540 /* ----------
541 * The query string built is
542 * SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
543 * FOR KEY SHARE OF x
544 * The type id's for the $ parameters are those of the
545 * PK attributes themselves.
546 *
547 * But for temporal FKs we need to make sure
548 * the old PK's range is completely covered.
549 * So we use this query instead:
550 * SELECT 1
551 * FROM (
552 * SELECT pkperiodatt AS r
553 * FROM [ONLY] pktable x
554 * WHERE pkatt1 = $1 [AND ...]
555 * AND pkperiodatt && $n
556 * FOR KEY SHARE OF x
557 * ) x1
558 * HAVING $n <@ range_agg(x1.r)
559 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
560 * we can make this a bit simpler.
561 * ----------
562 */
563 initStringInfo(&querybuf);
564 pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
565 "" : "ONLY ";
566 quoteRelationName(pkrelname, pk_rel);
567 if (riinfo->hasperiod)
568 {
569 quoteOneName(attname, RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));
570
571 appendStringInfo(&querybuf,
572 "SELECT 1 FROM (SELECT %s AS r FROM %s%s x",
573 attname, pk_only, pkrelname);
574 }
575 else
576 {
577 appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
578 pk_only, pkrelname);
579 }
580 querysep = "WHERE";
581 for (int i = 0; i < riinfo->nkeys; i++)
582 {
583 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
584
585 quoteOneName(attname,
586 RIAttName(pk_rel, riinfo->pk_attnums[i]));
587 sprintf(paramname, "$%d", i + 1);
588 ri_GenerateQual(&querybuf, querysep,
589 attname, pk_type,
590 riinfo->pp_eq_oprs[i],
591 paramname, pk_type);
592 querysep = "AND";
593 queryoids[i] = pk_type;
594 }
595 appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
596 if (riinfo->hasperiod)
597 {
598 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
599
600 appendStringInfo(&querybuf, ") x1 HAVING ");
601 sprintf(paramname, "$%d", riinfo->nkeys);
602 ri_GenerateQual(&querybuf, "",
603 paramname, fk_type,
604 riinfo->agged_period_contained_by_oper,
605 "pg_catalog.range_agg", ANYMULTIRANGEOID);
606 appendStringInfo(&querybuf, "(x1.r)");
607 }
608
609 /* Prepare and save the plan */
610 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
611 &qkey, fk_rel, pk_rel);
612 }
613
614 /*
615 * We have a plan now. Run it.
616 */
617 result = ri_PerformCheck(riinfo, &qkey, qplan,
618 fk_rel, pk_rel,
619 oldslot, NULL,
620 false,
621 true, /* treat like update */
622 SPI_OK_SELECT);
623
624 if (SPI_finish() != SPI_OK_FINISH)
625 elog(ERROR, "SPI_finish failed");
626
627 return result;
628 }
629
630
631 /*
632 * RI_FKey_noaction_del -
633 *
634 * Give an error and roll back the current transaction if the
635 * delete has resulted in a violation of the given referential
636 * integrity constraint.
637 */
638 Datum
639 RI_FKey_noaction_del(PG_FUNCTION_ARGS)
640 {
641 /* Check that this is a valid trigger call on the right time and event. */
642 ri_CheckTrigger(fcinfo, "RI_FKey_noaction_del", RI_TRIGTYPE_DELETE);
643
644 /* Share code with RESTRICT/UPDATE cases. */
645 return ri_restrict((TriggerData *) fcinfo->context, true);
646 }
647
648 /*
649 * RI_FKey_restrict_del -
650 *
651 * Restrict delete from PK table to rows unreferenced by foreign key.
652 *
653 * The SQL standard intends that this referential action occur exactly when
654 * the delete is performed, rather than after. This appears to be
655 * the only difference between "NO ACTION" and "RESTRICT". In Postgres
656 * we still implement this as an AFTER trigger, but it's non-deferrable.
657 */
658 Datum
659 RI_FKey_restrict_del(PG_FUNCTION_ARGS)
660 {
661 /* Check that this is a valid trigger call on the right time and event. */
662 ri_CheckTrigger(fcinfo, "RI_FKey_restrict_del", RI_TRIGTYPE_DELETE);
663
664 /* Share code with NO ACTION/UPDATE cases. */
665 return ri_restrict((TriggerData *) fcinfo->context, false);
666 }
667
668 /*
669 * RI_FKey_noaction_upd -
670 *
671 * Give an error and roll back the current transaction if the
672 * update has resulted in a violation of the given referential
673 * integrity constraint.
674 */
675 Datum
676 RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
677 {
678 /* Check that this is a valid trigger call on the right time and event. */
679 ri_CheckTrigger(fcinfo, "RI_FKey_noaction_upd", RI_TRIGTYPE_UPDATE);
680
681 /* Share code with RESTRICT/DELETE cases. */
682 return ri_restrict((TriggerData *) fcinfo->context, true);
683 }
684
685 /*
686 * RI_FKey_restrict_upd -
687 *
688 * Restrict update of PK to rows unreferenced by foreign key.
689 *
690 * The SQL standard intends that this referential action occur exactly when
691 * the update is performed, rather than after. This appears to be
692 * the only difference between "NO ACTION" and "RESTRICT". In Postgres
693 * we still implement this as an AFTER trigger, but it's non-deferrable.
694 */
695 Datum
696 RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
697 {
698 /* Check that this is a valid trigger call on the right time and event. */
699 ri_CheckTrigger(fcinfo, "RI_FKey_restrict_upd", RI_TRIGTYPE_UPDATE);
700
701 /* Share code with NO ACTION/DELETE cases. */
702 return ri_restrict((TriggerData *) fcinfo->context, false);
703 }
704
705 /*
706 * ri_restrict -
707 *
708 * Common code for ON DELETE RESTRICT, ON DELETE NO ACTION,
709 * ON UPDATE RESTRICT, and ON UPDATE NO ACTION.
710 */
711 static Datum
712 ri_restrict(TriggerData *trigdata, bool is_no_action)
713 {
714 const RI_ConstraintInfo *riinfo;
715 Relation fk_rel;
716 Relation pk_rel;
717 TupleTableSlot *oldslot;
718 RI_QueryKey qkey;
719 SPIPlanPtr qplan;
720
721 riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
722 trigdata->tg_relation, true);
723
724 /*
725 * Get the relation descriptors of the FK and PK tables and the old tuple.
726 *
727 * fk_rel is opened in RowShareLock mode since that's what our eventual
728 * SELECT FOR KEY SHARE will get on it.
729 */
730 fk_rel = table_open(riinfo->fk_relid, RowShareLock);
731 pk_rel = trigdata->tg_relation;
732 oldslot = trigdata->tg_trigslot;
733
734 /*
735 * If another PK row now exists providing the old key values, we should
736 * not do anything. However, this check should only be made in the NO
737 * ACTION case; in RESTRICT cases we don't wish to allow another row to be
738 * substituted.
739 *
740 * If the foreign key has PERIOD, we incorporate looking for replacement
741 * rows in the main SQL query below, so we needn't do it here.
742 */
743 if (is_no_action && !riinfo->hasperiod &&
744 ri_Check_Pk_Match(pk_rel, fk_rel, oldslot, riinfo))
745 {
746 table_close(fk_rel, RowShareLock);
747 return PointerGetDatum(NULL);
748 }
749
750 SPI_connect();
751
752 /*
753 * Fetch or prepare a saved plan for the restrict lookup (it's the same
754 * query for delete and update cases)
755 */
756 ri_BuildQueryKey(&qkey, riinfo, is_no_action ? RI_PLAN_NO_ACTION : RI_PLAN_RESTRICT);
757
758 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
759 {
760 StringInfoData querybuf;
761 char pkrelname[MAX_QUOTED_REL_NAME_LEN];
762 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
763 char attname[MAX_QUOTED_NAME_LEN];
764 char periodattname[MAX_QUOTED_NAME_LEN];
765 char paramname[16];
766 const char *querysep;
767 Oid queryoids[RI_MAX_NUMKEYS];
768 const char *fk_only;
769
770 /* ----------
771 * The query string built is
772 * SELECT 1 FROM [ONLY] <fktable> x WHERE $1 = fkatt1 [AND ...]
773 * FOR KEY SHARE OF x
774 * The type id's for the $ parameters are those of the
775 * corresponding PK attributes.
776 * ----------
777 */
778 initStringInfo(&querybuf);
779 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
780 "" : "ONLY ";
781 quoteRelationName(fkrelname, fk_rel);
782 appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",
783 fk_only, fkrelname);
784 querysep = "WHERE";
785 for (int i = 0; i < riinfo->nkeys; i++)
786 {
787 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
788 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
789
790 quoteOneName(attname,
791 RIAttName(fk_rel, riinfo->fk_attnums[i]));
792 sprintf(paramname, "$%d", i + 1);
793 ri_GenerateQual(&querybuf, querysep,
794 paramname, pk_type,
795 riinfo->pf_eq_oprs[i],
796 attname, fk_type);
797 querysep = "AND";
798 queryoids[i] = pk_type;
799 }
800
801 /*----------
802 * For temporal foreign keys, a reference could still be valid if the
803 * referenced range didn't change too much. Also if a referencing
804 * range extends past the current PK row, we don't want to check that
805 * part: some other PK row should fulfill it. We only want to check
806 * the part matching the PK record we've changed. Therefore to find
807 * invalid records we do this:
808 *
809 * SELECT 1 FROM [ONLY] <fktable> x WHERE $1 = x.fkatt1 [AND ...]
810 * -- begin temporal
811 * AND $n && x.fkperiod
812 * AND NOT coalesce((x.fkperiod * $n) <@
813 * (SELECT range_agg(r)
814 * FROM (SELECT y.pkperiod r
815 * FROM [ONLY] <pktable> y
816 * WHERE $1 = y.pkatt1 [AND ...] AND $n && y.pkperiod
817 * FOR KEY SHARE OF y) y2), false)
818 * -- end temporal
819 * FOR KEY SHARE OF x
820 *
821 * We need the coalesce in case the first subquery returns no rows.
822 * We need the second subquery because FOR KEY SHARE doesn't support
823 * aggregate queries.
824 */
825 if (riinfo->hasperiod && is_no_action)
826 {
827 Oid pk_period_type = RIAttType(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]);
828 Oid fk_period_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
829 StringInfoData intersectbuf;
830 StringInfoData replacementsbuf;
831 char *pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
832 "" : "ONLY ";
833
834 quoteOneName(attname, RIAttName(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]));
835 sprintf(paramname, "$%d", riinfo->nkeys);
836
837 appendStringInfoString(&querybuf, " AND NOT coalesce(");
838
839 /* Intersect the fk with the old pk range */
840 initStringInfo(&intersectbuf);
841 appendStringInfoString(&intersectbuf, "(");
842 ri_GenerateQual(&intersectbuf, "",
843 attname, fk_period_type,
844 riinfo->period_intersect_oper,
845 paramname, pk_period_type);
846 appendStringInfoString(&intersectbuf, ")");
847
848 /* Find the remaining history */
849 initStringInfo(&replacementsbuf);
850 appendStringInfoString(&replacementsbuf, "(SELECT pg_catalog.range_agg(r) FROM ");
851
852 quoteOneName(periodattname, RIAttName(pk_rel, riinfo->pk_attnums[riinfo->nkeys - 1]));
853 quoteRelationName(pkrelname, pk_rel);
854 appendStringInfo(&replacementsbuf, "(SELECT y.%s r FROM %s%s y",
855 periodattname, pk_only, pkrelname);
856
857 /* Restrict pk rows to what matches */
858 querysep = "WHERE";
859 for (int i = 0; i < riinfo->nkeys; i++)
860 {
861 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
862
863 quoteOneName(attname,
864 RIAttName(pk_rel, riinfo->pk_attnums[i]));
865 sprintf(paramname, "$%d", i + 1);
866 ri_GenerateQual(&replacementsbuf, querysep,
867 paramname, pk_type,
868 riinfo->pp_eq_oprs[i],
869 attname, pk_type);
870 querysep = "AND";
871 queryoids[i] = pk_type;
872 }
873 appendStringInfoString(&replacementsbuf, " FOR KEY SHARE OF y) y2)");
874
875 ri_GenerateQual(&querybuf, "",
876 intersectbuf.data, fk_period_type,
877 riinfo->agged_period_contained_by_oper,
878 replacementsbuf.data, ANYMULTIRANGEOID);
879 /* end of coalesce: */
880 appendStringInfoString(&querybuf, ", false)");
881 }
882
883 appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
884
885 /* Prepare and save the plan */
886 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
887 &qkey, fk_rel, pk_rel);
888 }
889
890 /*
891 * We have a plan now. Run it to check for existing references.
892 */
893 ri_PerformCheck(riinfo, &qkey, qplan,
894 fk_rel, pk_rel,
895 oldslot, NULL,
896 !is_no_action,
897 true, /* must detect new rows */
898 SPI_OK_SELECT);
899
900 if (SPI_finish() != SPI_OK_FINISH)
901 elog(ERROR, "SPI_finish failed");
902
903 table_close(fk_rel, RowShareLock);
904
905 return PointerGetDatum(NULL);
906 }
907
908
909 /*
910 * RI_FKey_cascade_del -
911 *
912 * Cascaded delete foreign key references at delete event on PK table.
913 */
914 Datum
915 RI_FKey_cascade_del(PG_FUNCTION_ARGS)
916 {
917 TriggerData *trigdata = (TriggerData *) fcinfo->context;
918 const RI_ConstraintInfo *riinfo;
919 Relation fk_rel;
920 Relation pk_rel;
921 TupleTableSlot *oldslot;
922 RI_QueryKey qkey;
923 SPIPlanPtr qplan;
924
925 /* Check that this is a valid trigger call on the right time and event. */
926 ri_CheckTrigger(fcinfo, "RI_FKey_cascade_del", RI_TRIGTYPE_DELETE);
927
928 riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
929 trigdata->tg_relation, true);
930
931 /*
932 * Get the relation descriptors of the FK and PK tables and the old tuple.
933 *
934 * fk_rel is opened in RowExclusiveLock mode since that's what our
935 * eventual DELETE will get on it.
936 */
937 fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
938 pk_rel = trigdata->tg_relation;
939 oldslot = trigdata->tg_trigslot;
940
941 SPI_connect();
942
943 /* Fetch or prepare a saved plan for the cascaded delete */
944 ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_ONDELETE);
945
946 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
947 {
948 StringInfoData querybuf;
949 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
950 char attname[MAX_QUOTED_NAME_LEN];
951 char paramname[16];
952 const char *querysep;
953 Oid queryoids[RI_MAX_NUMKEYS];
954 const char *fk_only;
955
956 /* ----------
957 * The query string built is
958 * DELETE FROM [ONLY] <fktable> WHERE $1 = fkatt1 [AND ...]
959 * The type id's for the $ parameters are those of the
960 * corresponding PK attributes.
961 * ----------
962 */
963 initStringInfo(&querybuf);
964 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
965 "" : "ONLY ";
966 quoteRelationName(fkrelname, fk_rel);
967 appendStringInfo(&querybuf, "DELETE FROM %s%s",
968 fk_only, fkrelname);
969 querysep = "WHERE";
970 for (int i = 0; i < riinfo->nkeys; i++)
971 {
972 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
973 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
974
975 quoteOneName(attname,
976 RIAttName(fk_rel, riinfo->fk_attnums[i]));
977 sprintf(paramname, "$%d", i + 1);
978 ri_GenerateQual(&querybuf, querysep,
979 paramname, pk_type,
980 riinfo->pf_eq_oprs[i],
981 attname, fk_type);
982 querysep = "AND";
983 queryoids[i] = pk_type;
984 }
985
986 /* Prepare and save the plan */
987 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
988 &qkey, fk_rel, pk_rel);
989 }
990
991 /*
992 * We have a plan now. Build up the arguments from the key values in the
993 * deleted PK tuple and delete the referencing rows
994 */
995 ri_PerformCheck(riinfo, &qkey, qplan,
996 fk_rel, pk_rel,
997 oldslot, NULL,
998 false,
999 true, /* must detect new rows */
1000 SPI_OK_DELETE);
1001
1002 if (SPI_finish() != SPI_OK_FINISH)
1003 elog(ERROR, "SPI_finish failed");
1004
1005 table_close(fk_rel, RowExclusiveLock);
1006
1007 return PointerGetDatum(NULL);
1008 }
1009
1010
1011 /*
1012 * RI_FKey_cascade_upd -
1013 *
1014 * Cascaded update foreign key references at update event on PK table.
1015 */
1016 Datum
1017 RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
1018 {
1019 TriggerData *trigdata = (TriggerData *) fcinfo->context;
1020 const RI_ConstraintInfo *riinfo;
1021 Relation fk_rel;
1022 Relation pk_rel;
1023 TupleTableSlot *newslot;
1024 TupleTableSlot *oldslot;
1025 RI_QueryKey qkey;
1026 SPIPlanPtr qplan;
1027
1028 /* Check that this is a valid trigger call on the right time and event. */
1029 ri_CheckTrigger(fcinfo, "RI_FKey_cascade_upd", RI_TRIGTYPE_UPDATE);
1030
1031 riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
1032 trigdata->tg_relation, true);
1033
1034 /*
1035 * Get the relation descriptors of the FK and PK tables and the new and
1036 * old tuple.
1037 *
1038 * fk_rel is opened in RowExclusiveLock mode since that's what our
1039 * eventual UPDATE will get on it.
1040 */
1041 fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
1042 pk_rel = trigdata->tg_relation;
1043 newslot = trigdata->tg_newslot;
1044 oldslot = trigdata->tg_trigslot;
1045
1046 SPI_connect();
1047
1048 /* Fetch or prepare a saved plan for the cascaded update */
1049 ri_BuildQueryKey(&qkey, riinfo, RI_PLAN_CASCADE_ONUPDATE);
1050
1051 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
1052 {
1053 StringInfoData querybuf;
1054 StringInfoData qualbuf;
1055 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
1056 char attname[MAX_QUOTED_NAME_LEN];
1057 char paramname[16];
1058 const char *querysep;
1059 const char *qualsep;
1060 Oid queryoids[RI_MAX_NUMKEYS * 2];
1061 const char *fk_only;
1062
1063 /* ----------
1064 * The query string built is
1065 * UPDATE [ONLY] <fktable> SET fkatt1 = $1 [, ...]
1066 * WHERE $n = fkatt1 [AND ...]
1067 * The type id's for the $ parameters are those of the
1068 * corresponding PK attributes. Note that we are assuming
1069 * there is an assignment cast from the PK to the FK type;
1070 * else the parser will fail.
1071 * ----------
1072 */
1073 initStringInfo(&querybuf);
1074 initStringInfo(&qualbuf);
1075 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
1076 "" : "ONLY ";
1077 quoteRelationName(fkrelname, fk_rel);
1078 appendStringInfo(&querybuf, "UPDATE %s%s SET",
1079 fk_only, fkrelname);
1080 querysep = "";
1081 qualsep = "WHERE";
1082 for (int i = 0, j = riinfo->nkeys; i < riinfo->nkeys; i++, j++)
1083 {
1084 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
1085 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
1086
1087 quoteOneName(attname,
1088 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1089 appendStringInfo(&querybuf,
1090 "%s %s = $%d",
1091 querysep, attname, i + 1);
1092 sprintf(paramname, "$%d", j + 1);
1093 ri_GenerateQual(&qualbuf, qualsep,
1094 paramname, pk_type,
1095 riinfo->pf_eq_oprs[i],
1096 attname, fk_type);
1097 querysep = ",";
1098 qualsep = "AND";
1099 queryoids[i] = pk_type;
1100 queryoids[j] = pk_type;
1101 }
1102 appendBinaryStringInfo(&querybuf, qualbuf.data, qualbuf.len);
1103
1104 /* Prepare and save the plan */
1105 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys * 2, queryoids,
1106 &qkey, fk_rel, pk_rel);
1107 }
1108
1109 /*
1110 * We have a plan now. Run it to update the existing references.
1111 */
1112 ri_PerformCheck(riinfo, &qkey, qplan,
1113 fk_rel, pk_rel,
1114 oldslot, newslot,
1115 false,
1116 true, /* must detect new rows */
1117 SPI_OK_UPDATE);
1118
1119 if (SPI_finish() != SPI_OK_FINISH)
1120 elog(ERROR, "SPI_finish failed");
1121
1122 table_close(fk_rel, RowExclusiveLock);
1123
1124 return PointerGetDatum(NULL);
1125 }
1126
1127
1128 /*
1129 * RI_FKey_setnull_del -
1130 *
1131 * Set foreign key references to NULL values at delete event on PK table.
1132 */
1133 Datum
1134 RI_FKey_setnull_del(PG_FUNCTION_ARGS)
1135 {
1136 /* Check that this is a valid trigger call on the right time and event. */
1137 ri_CheckTrigger(fcinfo, "RI_FKey_setnull_del", RI_TRIGTYPE_DELETE);
1138
1139 /* Share code with UPDATE case */
1140 return ri_set((TriggerData *) fcinfo->context, true, RI_TRIGTYPE_DELETE);
1141 }
1142
1143 /*
1144 * RI_FKey_setnull_upd -
1145 *
1146 * Set foreign key references to NULL at update event on PK table.
1147 */
1148 Datum
1149 RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
1150 {
1151 /* Check that this is a valid trigger call on the right time and event. */
1152 ri_CheckTrigger(fcinfo, "RI_FKey_setnull_upd", RI_TRIGTYPE_UPDATE);
1153
1154 /* Share code with DELETE case */
1155 return ri_set((TriggerData *) fcinfo->context, true, RI_TRIGTYPE_UPDATE);
1156 }
1157
1158 /*
1159 * RI_FKey_setdefault_del -
1160 *
1161 * Set foreign key references to defaults at delete event on PK table.
1162 */
1163 Datum
1164 RI_FKey_setdefault_del(PG_FUNCTION_ARGS)
1165 {
1166 /* Check that this is a valid trigger call on the right time and event. */
1167 ri_CheckTrigger(fcinfo, "RI_FKey_setdefault_del", RI_TRIGTYPE_DELETE);
1168
1169 /* Share code with UPDATE case */
1170 return ri_set((TriggerData *) fcinfo->context, false, RI_TRIGTYPE_DELETE);
1171 }
1172
1173 /*
1174 * RI_FKey_setdefault_upd -
1175 *
1176 * Set foreign key references to defaults at update event on PK table.
1177 */
1178 Datum
1179 RI_FKey_setdefault_upd(PG_FUNCTION_ARGS)
1180 {
1181 /* Check that this is a valid trigger call on the right time and event. */
1182 ri_CheckTrigger(fcinfo, "RI_FKey_setdefault_upd", RI_TRIGTYPE_UPDATE);
1183
1184 /* Share code with DELETE case */
1185 return ri_set((TriggerData *) fcinfo->context, false, RI_TRIGTYPE_UPDATE);
1186 }
1187
1188 /*
1189 * ri_set -
1190 *
1191 * Common code for ON DELETE SET NULL, ON DELETE SET DEFAULT, ON UPDATE SET
1192 * NULL, and ON UPDATE SET DEFAULT.
1193 */
1194 static Datum
1195 ri_set(TriggerData *trigdata, bool is_set_null, int tgkind)
1196 {
1197 const RI_ConstraintInfo *riinfo;
1198 Relation fk_rel;
1199 Relation pk_rel;
1200 TupleTableSlot *oldslot;
1201 RI_QueryKey qkey;
1202 SPIPlanPtr qplan;
1203 int32 queryno;
1204
1205 riinfo = ri_FetchConstraintInfo(trigdata->tg_trigger,
1206 trigdata->tg_relation, true);
1207
1208 /*
1209 * Get the relation descriptors of the FK and PK tables and the old tuple.
1210 *
1211 * fk_rel is opened in RowExclusiveLock mode since that's what our
1212 * eventual UPDATE will get on it.
1213 */
1214 fk_rel = table_open(riinfo->fk_relid, RowExclusiveLock);
1215 pk_rel = trigdata->tg_relation;
1216 oldslot = trigdata->tg_trigslot;
1217
1218 SPI_connect();
1219
1220 /*
1221 * Fetch or prepare a saved plan for the trigger.
1222 */
1223 switch (tgkind)
1224 {
1225 case RI_TRIGTYPE_UPDATE:
1226 queryno = is_set_null
1227 ? RI_PLAN_SETNULL_ONUPDATE
1228 : RI_PLAN_SETDEFAULT_ONUPDATE;
1229 break;
1230 case RI_TRIGTYPE_DELETE:
1231 queryno = is_set_null
1232 ? RI_PLAN_SETNULL_ONDELETE
1233 : RI_PLAN_SETDEFAULT_ONDELETE;
1234 break;
1235 default:
1236 elog(ERROR, "invalid tgkind passed to ri_set");
1237 }
1238
1239 ri_BuildQueryKey(&qkey, riinfo, queryno);
1240
1241 if ((qplan = ri_FetchPreparedPlan(&qkey)) == NULL)
1242 {
1243 StringInfoData querybuf;
1244 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
1245 char attname[MAX_QUOTED_NAME_LEN];
1246 char paramname[16];
1247 const char *querysep;
1248 const char *qualsep;
1249 Oid queryoids[RI_MAX_NUMKEYS];
1250 const char *fk_only;
1251 int num_cols_to_set;
1252 const int16 *set_cols;
1253
1254 switch (tgkind)
1255 {
1256 case RI_TRIGTYPE_UPDATE:
1257 num_cols_to_set = riinfo->nkeys;
1258 set_cols = riinfo->fk_attnums;
1259 break;
1260 case RI_TRIGTYPE_DELETE:
1261
1262 /*
1263 * If confdelsetcols are present, then we only update the
1264 * columns specified in that array, otherwise we update all
1265 * the referencing columns.
1266 */
1267 if (riinfo->ndelsetcols != 0)
1268 {
1269 num_cols_to_set = riinfo->ndelsetcols;
1270 set_cols = riinfo->confdelsetcols;
1271 }
1272 else
1273 {
1274 num_cols_to_set = riinfo->nkeys;
1275 set_cols = riinfo->fk_attnums;
1276 }
1277 break;
1278 default:
1279 elog(ERROR, "invalid tgkind passed to ri_set");
1280 }
1281
1282 /* ----------
1283 * The query string built is
1284 * UPDATE [ONLY] <fktable> SET fkatt1 = {NULL|DEFAULT} [, ...]
1285 * WHERE $1 = fkatt1 [AND ...]
1286 * The type id's for the $ parameters are those of the
1287 * corresponding PK attributes.
1288 * ----------
1289 */
1290 initStringInfo(&querybuf);
1291 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
1292 "" : "ONLY ";
1293 quoteRelationName(fkrelname, fk_rel);
1294 appendStringInfo(&querybuf, "UPDATE %s%s SET",
1295 fk_only, fkrelname);
1296
1297 /*
1298 * Add assignment clauses
1299 */
1300 querysep = "";
1301 for (int i = 0; i < num_cols_to_set; i++)
1302 {
1303 quoteOneName(attname, RIAttName(fk_rel, set_cols[i]));
1304 appendStringInfo(&querybuf,
1305 "%s %s = %s",
1306 querysep, attname,
1307 is_set_null ? "NULL" : "DEFAULT");
1308 querysep = ",";
1309 }
1310
1311 /*
1312 * Add WHERE clause
1313 */
1314 qualsep = "WHERE";
1315 for (int i = 0; i < riinfo->nkeys; i++)
1316 {
1317 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
1318 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
1319
1320 quoteOneName(attname,
1321 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1322
1323 sprintf(paramname, "$%d", i + 1);
1324 ri_GenerateQual(&querybuf, qualsep,
1325 paramname, pk_type,
1326 riinfo->pf_eq_oprs[i],
1327 attname, fk_type);
1328 qualsep = "AND";
1329 queryoids[i] = pk_type;
1330 }
1331
1332 /* Prepare and save the plan */
1333 qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
1334 &qkey, fk_rel, pk_rel);
1335 }
1336
1337 /*
1338 * We have a plan now. Run it to update the existing references.
1339 */
1340 ri_PerformCheck(riinfo, &qkey, qplan,
1341 fk_rel, pk_rel,
1342 oldslot, NULL,
1343 false,
1344 true, /* must detect new rows */
1345 SPI_OK_UPDATE);
1346
1347 if (SPI_finish() != SPI_OK_FINISH)
1348 elog(ERROR, "SPI_finish failed");
1349
1350 table_close(fk_rel, RowExclusiveLock);
1351
1352 if (is_set_null)
1353 return PointerGetDatum(NULL);
1354 else
1355 {
1356 /*
1357 * If we just deleted or updated the PK row whose key was equal to the
1358 * FK columns' default values, and a referencing row exists in the FK
1359 * table, we would have updated that row to the same values it already
1360 * had --- and RI_FKey_fk_upd_check_required would hence believe no
1361 * check is necessary. So we need to do another lookup now and in
1362 * case a reference still exists, abort the operation. That is
1363 * already implemented in the NO ACTION trigger, so just run it. (This
1364 * recheck is only needed in the SET DEFAULT case, since CASCADE would
1365 * remove such rows in case of a DELETE operation or would change the
1366 * FK key values in case of an UPDATE, while SET NULL is certain to
1367 * result in rows that satisfy the FK constraint.)
1368 */
1369 return ri_restrict(trigdata, true);
1370 }
1371 }
1372
1373
1374 /*
1375 * RI_FKey_pk_upd_check_required -
1376 *
1377 * Check if we really need to fire the RI trigger for an update or delete to a PK
1378 * relation. This is called by the AFTER trigger queue manager to see if
1379 * it can skip queuing an instance of an RI trigger. Returns true if the
1380 * trigger must be fired, false if we can prove the constraint will still
1381 * be satisfied.
1382 *
1383 * newslot will be NULL if this is called for a delete.
1384 */
1385 bool
1386 RI_FKey_pk_upd_check_required(Trigger *trigger, Relation pk_rel,
1387 TupleTableSlot *oldslot, TupleTableSlot *newslot)
1388 {
1389 const RI_ConstraintInfo *riinfo;
1390
1391 riinfo = ri_FetchConstraintInfo(trigger, pk_rel, true);
1392
1393 /*
1394 * If any old key value is NULL, the row could not have been referenced by
1395 * an FK row, so no check is needed.
1396 */
1397 if (ri_NullCheck(RelationGetDescr(pk_rel), oldslot, riinfo, true) != RI_KEYS_NONE_NULL)
1398 return false;
1399
1400 /* If all old and new key values are equal, no check is needed */
1401 if (newslot && ri_KeysEqual(pk_rel, oldslot, newslot, riinfo, true))
1402 return false;
1403
1404 /* Else we need to fire the trigger. */
1405 return true;
1406 }
1407
1408 /*
1409 * RI_FKey_fk_upd_check_required -
1410 *
1411 * Check if we really need to fire the RI trigger for an update to an FK
1412 * relation. This is called by the AFTER trigger queue manager to see if
1413 * it can skip queuing an instance of an RI trigger. Returns true if the
1414 * trigger must be fired, false if we can prove the constraint will still
1415 * be satisfied.
1416 */
1417 bool
1418 RI_FKey_fk_upd_check_required(Trigger *trigger, Relation fk_rel,
1419 TupleTableSlot *oldslot, TupleTableSlot *newslot)
1420 {
1421 const RI_ConstraintInfo *riinfo;
1422 int ri_nullcheck;
1423
1424 /*
1425 * AfterTriggerSaveEvent() handles things such that this function is never
1426 * called for partitioned tables.
1427 */
1428 Assert(fk_rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE);
1429
1430 riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);
1431
1432 ri_nullcheck = ri_NullCheck(RelationGetDescr(fk_rel), newslot, riinfo, false);
1433
1434 /*
1435 * If all new key values are NULL, the row satisfies the constraint, so no
1436 * check is needed.
1437 */
1438 if (ri_nullcheck == RI_KEYS_ALL_NULL)
1439 return false;
1440
1441 /*
1442 * If some new key values are NULL, the behavior depends on the match
1443 * type.
1444 */
1445 else if (ri_nullcheck == RI_KEYS_SOME_NULL)
1446 {
1447 switch (riinfo->confmatchtype)
1448 {
1449 case FKCONSTR_MATCH_SIMPLE:
1450
1451 /*
1452 * If any new key value is NULL, the row must satisfy the
1453 * constraint, so no check is needed.
1454 */
1455 return false;
1456
1457 case FKCONSTR_MATCH_PARTIAL:
1458
1459 /*
1460 * Don't know, must run full check.
1461 */
1462 break;
1463
1464 case FKCONSTR_MATCH_FULL:
1465
1466 /*
1467 * If some new key values are NULL, the row fails the
1468 * constraint. We must not throw error here, because the row
1469 * might get invalidated before the constraint is to be
1470 * checked, but we should queue the event to apply the check
1471 * later.
1472 */
1473 return true;
1474 }
1475 }
1476
1477 /*
1478 * Continues here for no new key values are NULL, or we couldn't decide
1479 * yet.
1480 */
1481
1482 /*
1483 * If the original row was inserted by our own transaction, we must fire
1484 * the trigger whether or not the keys are equal. This is because our
1485 * UPDATE will invalidate the INSERT so that the INSERT RI trigger will
1486 * not do anything; so we had better do the UPDATE check. (We could skip
1487 * this if we knew the INSERT trigger already fired, but there is no easy
1488 * way to know that.)
1489 */
1490 if (slot_is_current_xact_tuple(oldslot))
1491 return true;
1492
1493 /* If all old and new key values are equal, no check is needed */
1494 if (ri_KeysEqual(fk_rel, oldslot, newslot, riinfo, false))
1495 return false;
1496
1497 /* Else we need to fire the trigger. */
1498 return true;
1499 }
1500
1501 /*
1502 * RI_Initial_Check -
1503 *
1504 * Check an entire table for non-matching values using a single query.
1505 * This is not a trigger procedure, but is called during ALTER TABLE
1506 * ADD FOREIGN KEY to validate the initial table contents.
1507 *
1508 * We expect that the caller has made provision to prevent any problems
1509 * caused by concurrent actions. This could be either by locking rel and
1510 * pkrel at ShareRowExclusiveLock or higher, or by otherwise ensuring
1511 * that triggers implementing the checks are already active.
1512 * Hence, we do not need to lock individual rows for the check.
1513 *
1514 * If the check fails because the current user doesn't have permissions
1515 * to read both tables, return false to let our caller know that they will
1516 * need to do something else to check the constraint.
1517 */
1518 bool
1519 RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
1520 {
1521 const RI_ConstraintInfo *riinfo;
1522 StringInfoData querybuf;
1523 char pkrelname[MAX_QUOTED_REL_NAME_LEN];
1524 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
1525 char pkattname[MAX_QUOTED_NAME_LEN + 3];
1526 char fkattname[MAX_QUOTED_NAME_LEN + 3];
1527 RangeTblEntry *rte;
1528 RTEPermissionInfo *pk_perminfo;
1529 RTEPermissionInfo *fk_perminfo;
1530 List *rtes = NIL;
1531 List *perminfos = NIL;
1532 const char *sep;
1533 const char *fk_only;
1534 const char *pk_only;
1535 int save_nestlevel;
1536 char workmembuf[32];
1537 int spi_result;
1538 SPIPlanPtr qplan;
1539
1540 riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);
1541
1542 /*
1543 * Check to make sure current user has enough permissions to do the test
1544 * query. (If not, caller can fall back to the trigger method, which
1545 * works because it changes user IDs on the fly.)
1546 *
1547 * XXX are there any other show-stopper conditions to check?
1548 */
1549 pk_perminfo = makeNode(RTEPermissionInfo);
1550 pk_perminfo->relid = RelationGetRelid(pk_rel);
1551 pk_perminfo->requiredPerms = ACL_SELECT;
1552 perminfos = lappend(perminfos, pk_perminfo);
1553 rte = makeNode(RangeTblEntry);
1554 rte->rtekind = RTE_RELATION;
1555 rte->relid = RelationGetRelid(pk_rel);
1556 rte->relkind = pk_rel->rd_rel->relkind;
1557 rte->rellockmode = AccessShareLock;
1558 rte->perminfoindex = list_length(perminfos);
1559 rtes = lappend(rtes, rte);
1560
1561 fk_perminfo = makeNode(RTEPermissionInfo);
1562 fk_perminfo->relid = RelationGetRelid(fk_rel);
1563 fk_perminfo->requiredPerms = ACL_SELECT;
1564 perminfos = lappend(perminfos, fk_perminfo);
1565 rte = makeNode(RangeTblEntry);
1566 rte->rtekind = RTE_RELATION;
1567 rte->relid = RelationGetRelid(fk_rel);
1568 rte->relkind = fk_rel->rd_rel->relkind;
1569 rte->rellockmode = AccessShareLock;
1570 rte->perminfoindex = list_length(perminfos);
1571 rtes = lappend(rtes, rte);
1572
1573 for (int i = 0; i < riinfo->nkeys; i++)
1574 {
1575 int attno;
1576
1577 attno = riinfo->pk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
1578 pk_perminfo->selectedCols = bms_add_member(pk_perminfo->selectedCols, attno);
1579
1580 attno = riinfo->fk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
1581 fk_perminfo->selectedCols = bms_add_member(fk_perminfo->selectedCols, attno);
1582 }
1583
1584 if (!ExecCheckPermissions(rtes, perminfos, false))
1585 return false;
1586
1587 /*
1588 * Also punt if RLS is enabled on either table unless this role has the
1589 * bypassrls right or is the table owner of the table(s) involved which
1590 * have RLS enabled.
1591 */
1592 if (!has_bypassrls_privilege(GetUserId()) &&
1593 ((pk_rel->rd_rel->relrowsecurity &&
1594 !object_ownercheck(RelationRelationId, RelationGetRelid(pk_rel),
1595 GetUserId())) ||
1596 (fk_rel->rd_rel->relrowsecurity &&
1597 !object_ownercheck(RelationRelationId, RelationGetRelid(fk_rel),
1598 GetUserId()))))
1599 return false;
1600
1601 /*----------
1602 * The query string built is:
1603 * SELECT fk.keycols FROM [ONLY] relname fk
1604 * LEFT OUTER JOIN [ONLY] pkrelname pk
1605 * ON (pk.pkkeycol1=fk.keycol1 [AND ...])
1606 * WHERE pk.pkkeycol1 IS NULL AND
1607 * For MATCH SIMPLE:
1608 * (fk.keycol1 IS NOT NULL [AND ...])
1609 * For MATCH FULL:
1610 * (fk.keycol1 IS NOT NULL [OR ...])
1611 *
1612 * We attach COLLATE clauses to the operators when comparing columns
1613 * that have different collations.
1614 *----------
1615 */
1616 initStringInfo(&querybuf);
1617 appendStringInfoString(&querybuf, "SELECT ");
1618 sep = "";
1619 for (int i = 0; i < riinfo->nkeys; i++)
1620 {
1621 quoteOneName(fkattname,
1622 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1623 appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
1624 sep = ", ";
1625 }
1626
1627 quoteRelationName(pkrelname, pk_rel);
1628 quoteRelationName(fkrelname, fk_rel);
1629 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
1630 "" : "ONLY ";
1631 pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
1632 "" : "ONLY ";
1633 appendStringInfo(&querybuf,
1634 " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON",
1635 fk_only, fkrelname, pk_only, pkrelname);
1636
1637 strcpy(pkattname, "pk.");
1638 strcpy(fkattname, "fk.");
1639 sep = "(";
1640 for (int i = 0; i < riinfo->nkeys; i++)
1641 {
1642 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
1643 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
1644 Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
1645 Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
1646
1647 quoteOneName(pkattname + 3,
1648 RIAttName(pk_rel, riinfo->pk_attnums[i]));
1649 quoteOneName(fkattname + 3,
1650 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1651 ri_GenerateQual(&querybuf, sep,
1652 pkattname, pk_type,
1653 riinfo->pf_eq_oprs[i],
1654 fkattname, fk_type);
1655 if (pk_coll != fk_coll)
1656 ri_GenerateQualCollation(&querybuf, pk_coll);
1657 sep = "AND";
1658 }
1659
1660 /*
1661 * It's sufficient to test any one pk attribute for null to detect a join
1662 * failure.
1663 */
1664 quoteOneName(pkattname, RIAttName(pk_rel, riinfo->pk_attnums[0]));
1665 appendStringInfo(&querybuf, ") WHERE pk.%s IS NULL AND (", pkattname);
1666
1667 sep = "";
1668 for (int i = 0; i < riinfo->nkeys; i++)
1669 {
1670 quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
1671 appendStringInfo(&querybuf,
1672 "%sfk.%s IS NOT NULL",
1673 sep, fkattname);
1674 switch (riinfo->confmatchtype)
1675 {
1676 case FKCONSTR_MATCH_SIMPLE:
1677 sep = " AND ";
1678 break;
1679 case FKCONSTR_MATCH_FULL:
1680 sep = " OR ";
1681 break;
1682 }
1683 }
1684 appendStringInfoChar(&querybuf, ')');
1685
1686 /*
1687 * Temporarily increase work_mem so that the check query can be executed
1688 * more efficiently. It seems okay to do this because the query is simple
1689 * enough to not use a multiple of work_mem, and one typically would not
1690 * have many large foreign-key validations happening concurrently. So
1691 * this seems to meet the criteria for being considered a "maintenance"
1692 * operation, and accordingly we use maintenance_work_mem. However, we
1693 * must also set hash_mem_multiplier to 1, since it is surely not okay to
1694 * let that get applied to the maintenance_work_mem value.
1695 *
1696 * We use the equivalent of a function SET option to allow the setting to
1697 * persist for exactly the duration of the check query. guc.c also takes
1698 * care of undoing the setting on error.
1699 */
1700 save_nestlevel = NewGUCNestLevel();
1701
1702 snprintf(workmembuf, sizeof(workmembuf), "%d", maintenance_work_mem);
1703 (void) set_config_option("work_mem", workmembuf,
1704 PGC_USERSET, PGC_S_SESSION,
1705 GUC_ACTION_SAVE, true, 0, false);
1706 (void) set_config_option("hash_mem_multiplier", "1",
1707 PGC_USERSET, PGC_S_SESSION,
1708 GUC_ACTION_SAVE, true, 0, false);
1709
1710 SPI_connect();
1711
1712 /*
1713 * Generate the plan. We don't need to cache it, and there are no
1714 * arguments to the plan.
1715 */
1716 qplan = SPI_prepare(querybuf.data, 0, NULL);
1717
1718 if (qplan == NULL)
1719 elog(ERROR, "SPI_prepare returned %s for %s",
1720 SPI_result_code_string(SPI_result), querybuf.data);
1721
1722 /*
1723 * Run the plan. For safety we force a current snapshot to be used. (In
1724 * transaction-snapshot mode, this arguably violates transaction isolation
1725 * rules, but we really haven't got much choice.) We don't need to
1726 * register the snapshot, because SPI_execute_snapshot will see to it. We
1727 * need at most one tuple returned, so pass limit = 1.
1728 */
1729 spi_result = SPI_execute_snapshot(qplan,
1730 NULL, NULL,
1731 GetLatestSnapshot(),
1732 InvalidSnapshot,
1733 true, false, 1);
1734
1735 /* Check result */
1736 if (spi_result != SPI_OK_SELECT)
1737 elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));
1738
1739 /* Did we find a tuple violating the constraint? */
1740 if (SPI_processed > 0)
1741 {
1742 TupleTableSlot *slot;
1743 HeapTuple tuple = SPI_tuptable->vals[0];
1744 TupleDesc tupdesc = SPI_tuptable->tupdesc;
1745 RI_ConstraintInfo fake_riinfo;
1746
1747 slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);
1748
1749 heap_deform_tuple(tuple, tupdesc,
1750 slot->tts_values, slot->tts_isnull);
1751 ExecStoreVirtualTuple(slot);
1752
1753 /*
1754 * The columns to look at in the result tuple are 1..N, not whatever
1755 * they are in the fk_rel. Hack up riinfo so that the subroutines
1756 * called here will behave properly.
1757 *
1758 * In addition to this, we have to pass the correct tupdesc to
1759 * ri_ReportViolation, overriding its normal habit of using the pk_rel
1760 * or fk_rel's tupdesc.
1761 */
1762 memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
1763 for (int i = 0; i < fake_riinfo.nkeys; i++)
1764 fake_riinfo.fk_attnums[i] = i + 1;
1765
1766 /*
1767 * If it's MATCH FULL, and there are any nulls in the FK keys,
1768 * complain about that rather than the lack of a match. MATCH FULL
1769 * disallows partially-null FK rows.
1770 */
1771 if (fake_riinfo.confmatchtype == FKCONSTR_MATCH_FULL &&
1772 ri_NullCheck(tupdesc, slot, &fake_riinfo, false) != RI_KEYS_NONE_NULL)
1773 ereport(ERROR,
1774 (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
1775 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
1776 RelationGetRelationName(fk_rel),
1777 NameStr(fake_riinfo.conname)),
1778 errdetail("MATCH FULL does not allow mixing of null and nonnull key values."),
1779 errtableconstraint(fk_rel,
1780 NameStr(fake_riinfo.conname))));
1781
1782 /*
1783 * We tell ri_ReportViolation we were doing the RI_PLAN_CHECK_LOOKUPPK
1784 * query, which isn't true, but will cause it to use
1785 * fake_riinfo.fk_attnums as we need.
1786 */
1787 ri_ReportViolation(&fake_riinfo,
1788 pk_rel, fk_rel,
1789 slot, tupdesc,
1790 RI_PLAN_CHECK_LOOKUPPK, false, false);
1791
1792 ExecDropSingleTupleTableSlot(slot);
1793 }
1794
1795 if (SPI_finish() != SPI_OK_FINISH)
1796 elog(ERROR, "SPI_finish failed");
1797
1798 /*
1799 * Restore work_mem and hash_mem_multiplier.
1800 */
1801 AtEOXact_GUC(true, save_nestlevel);
1802
1803 return true;
1804 }
1805
1806 /*
1807 * RI_PartitionRemove_Check -
1808 *
1809 * Verify no referencing values exist, when a partition is detached on
1810 * the referenced side of a foreign key constraint.
1811 */
1812 void
1813 RI_PartitionRemove_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
1814 {
1815 const RI_ConstraintInfo *riinfo;
1816 StringInfoData querybuf;
1817 char *constraintDef;
1818 char pkrelname[MAX_QUOTED_REL_NAME_LEN];
1819 char fkrelname[MAX_QUOTED_REL_NAME_LEN];
1820 char pkattname[MAX_QUOTED_NAME_LEN + 3];
1821 char fkattname[MAX_QUOTED_NAME_LEN + 3];
1822 const char *sep;
1823 const char *fk_only;
1824 int save_nestlevel;
1825 char workmembuf[32];
1826 int spi_result;
1827 SPIPlanPtr qplan;
1828 int i;
1829
1830 riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);
1831
1832 /*
1833 * We don't check permissions before displaying the error message, on the
1834 * assumption that the user detaching the partition must have enough
1835 * privileges to examine the table contents anyhow.
1836 */
1837
1838 /*----------
1839 * The query string built is:
1840 * SELECT fk.keycols FROM [ONLY] relname fk
1841 * JOIN pkrelname pk
1842 * ON (pk.pkkeycol1=fk.keycol1 [AND ...])
1843 * WHERE (<partition constraint>) AND
1844 * For MATCH SIMPLE:
1845 * (fk.keycol1 IS NOT NULL [AND ...])
1846 * For MATCH FULL:
1847 * (fk.keycol1 IS NOT NULL [OR ...])
1848 *
1849 * We attach COLLATE clauses to the operators when comparing columns
1850 * that have different collations.
1851 *----------
1852 */
1853 initStringInfo(&querybuf);
1854 appendStringInfoString(&querybuf, "SELECT ");
1855 sep = "";
1856 for (i = 0; i < riinfo->nkeys; i++)
1857 {
1858 quoteOneName(fkattname,
1859 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1860 appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);
1861 sep = ", ";
1862 }
1863
1864 quoteRelationName(pkrelname, pk_rel);
1865 quoteRelationName(fkrelname, fk_rel);
1866 fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
1867 "" : "ONLY ";
1868 appendStringInfo(&querybuf,
1869 " FROM %s%s fk JOIN %s pk ON",
1870 fk_only, fkrelname, pkrelname);
1871 strcpy(pkattname, "pk.");
1872 strcpy(fkattname, "fk.");
1873 sep = "(";
1874 for (i = 0; i < riinfo->nkeys; i++)
1875 {
1876 Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);
1877 Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);
1878 Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);
1879 Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);
1880
1881 quoteOneName(pkattname + 3,
1882 RIAttName(pk_rel, riinfo->pk_attnums[i]));
1883 quoteOneName(fkattname + 3,
1884 RIAttName(fk_rel, riinfo->fk_attnums[i]));
1885 ri_GenerateQual(&querybuf, sep,
1886 pkattname, pk_type,
1887 riinfo->pf_eq_oprs[i],
1888 fkattname, fk_type);
1889 if (pk_coll != fk_coll)
1890 ri_GenerateQualCollation(&querybuf, pk_coll);
1891 sep = "AND";
1892 }
1893
1894 /*
1895 * Start the WHERE clause with the partition constraint (except if this is
1896 * the default partition and there's no other partition, because the
1897 * partition constraint is the empty string in that case.)
1898 */
1899 constraintDef = pg_get_partconstrdef_string(RelationGetRelid(pk_rel), "pk");
1900 if (constraintDef && constraintDef[0] != '\0')
1901 appendStringInfo(&querybuf, ") WHERE %s AND (",
1902 constraintDef);
1903 else
1904 appendStringInfoString(&querybuf, ") WHERE (");
1905
1906 sep = "";
1907 for (i = 0; i < riinfo->nkeys; i++)
1908 {
1909 quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));
1910 appendStringInfo(&querybuf,
1911 "%sfk.%s IS NOT NULL",
1912 sep, fkattname);
1913 switch (riinfo->confmatchtype)
1914 {
1915 case FKCONSTR_MATCH_SIMPLE:
1916 sep = " AND ";
1917 break;
1918 case FKCONSTR_MATCH_FULL:
1919 sep = " OR ";
1920 break;
1921 }
1922 }
1923 appendStringInfoChar(&querybuf, ')');
1924
1925 /*
1926 * Temporarily increase work_mem so that the check query can be executed
1927 * more efficiently. It seems okay to do this because the query is simple
1928 * enough to not use a multiple of work_mem, and one typically would not
1929 * have many large foreign-key validations happening concurrently. So
1930 * this seems to meet the criteria for being considered a "maintenance"
1931 * operation, and accordingly we use maintenance_work_mem. However, we
1932 * must also set hash_mem_multiplier to 1, since it is surely not okay to
1933 * let that get applied to the maintenance_work_mem value.
1934 *
1935 * We use the equivalent of a function SET option to allow the setting to
1936 * persist for exactly the duration of the check query. guc.c also takes
1937 * care of undoing the setting on error.
1938 */
1939 save_nestlevel = NewGUCNestLevel();
1940
1941 snprintf(workmembuf, sizeof(workmembuf), "%d", maintenance_work_mem);
1942 (void) set_config_option("work_mem", workmembuf,
1943 PGC_USERSET, PGC_S_SESSION,
1944 GUC_ACTION_SAVE, true, 0, false);
1945 (void) set_config_option("hash_mem_multiplier", "1",
1946 PGC_USERSET, PGC_S_SESSION,
1947 GUC_ACTION_SAVE, true, 0, false);
1948
1949 SPI_connect();
1950
1951 /*
1952 * Generate the plan. We don't need to cache it, and there are no
1953 * arguments to the plan.
1954 */
1955 qplan = SPI_prepare(querybuf.data, 0, NULL);
1956
1957 if (qplan == NULL)
1958 elog(ERROR, "SPI_prepare returned %s for %s",
1959 SPI_result_code_string(SPI_result), querybuf.data);
1960
1961 /*
1962 * Run the plan. For safety we force a current snapshot to be used. (In
1963 * transaction-snapshot mode, this arguably violates transaction isolation
1964 * rules, but we really haven't got much choice.) We don't need to
1965 * register the snapshot, because SPI_execute_snapshot will see to it. We
1966 * need at most one tuple returned, so pass limit = 1.
1967 */
1968 spi_result = SPI_execute_snapshot(qplan,
1969 NULL, NULL,
1970 GetLatestSnapshot(),
1971 InvalidSnapshot,
1972 true, false, 1);
1973
1974 /* Check result */
1975 if (spi_result != SPI_OK_SELECT)
1976 elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));
1977
1978 /* Did we find a tuple that would violate the constraint? */
1979 if (SPI_processed > 0)
1980 {
1981 TupleTableSlot *slot;
1982 HeapTuple tuple = SPI_tuptable->vals[0];
1983 TupleDesc tupdesc = SPI_tuptable->tupdesc;
1984 RI_ConstraintInfo fake_riinfo;
1985
1986 slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual);
1987
1988 heap_deform_tuple(tuple, tupdesc,
1989 slot->tts_values, slot->tts_isnull);
1990 ExecStoreVirtualTuple(slot);
1991
1992 /*
1993 * The columns to look at in the result tuple are 1..N, not whatever
1994 * they are in the fk_rel. Hack up riinfo so that ri_ReportViolation
1995 * will behave properly.
1996 *
1997 * In addition to this, we have to pass the correct tupdesc to
1998 * ri_ReportViolation, overriding its normal habit of using the pk_rel
1999 * or fk_rel's tupdesc.
2000 */
2001 memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));
2002 for (i = 0; i < fake_riinfo.nkeys; i++)
2003 fake_riinfo.pk_attnums[i] = i + 1;
2004
2005 ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel,
2006 slot, tupdesc, 0, false, true);
2007 }
2008
2009 if (SPI_finish() != SPI_OK_FINISH)
2010 elog(ERROR, "SPI_finish failed");
2011
2012 /*
2013 * Restore work_mem and hash_mem_multiplier.
2014 */
2015 AtEOXact_GUC(true, save_nestlevel);
2016 }
2017
2018
2019 /* ----------
2020 * Local functions below
2021 * ----------
2022 */
2023
2024
2025 /*
2026 * quoteOneName --- safely quote a single SQL name
2027 *
2028 * buffer must be MAX_QUOTED_NAME_LEN long (includes room for \0)
2029 */
2030 static void
2031 quoteOneName(char *buffer, const char *name)
2032 {
2033 /* Rather than trying to be smart, just always quote it. */
2034 *buffer++ = '"';
2035 while (*name)
2036 {
2037 if (*name == '"')
2038 *buffer++ = '"';
2039 *buffer++ = *name++;
2040 }
2041 *buffer++ = '"';
2042 *buffer = '\0';
2043 }
2044
2045 /*
2046 * quoteRelationName --- safely quote a fully qualified relation name
2047 *
2048 * buffer must be MAX_QUOTED_REL_NAME_LEN long (includes room for \0)
2049 */
2050 static void
2051 quoteRelationName(char *buffer, Relation rel)
2052 {
2053 quoteOneName(buffer, get_namespace_name(RelationGetNamespace(rel)));
2054 buffer += strlen(buffer);
2055 *buffer++ = '.';
2056 quoteOneName(buffer, RelationGetRelationName(rel));
2057 }
2058
2059 /*
2060 * ri_GenerateQual --- generate a WHERE clause equating two variables
2061 *
2062 * This basically appends " sep leftop op rightop" to buf, adding casts
2063 * and schema qualification as needed to ensure that the parser will select
2064 * the operator we specify. leftop and rightop should be parenthesized
2065 * if they aren't variables or parameters.
2066 */
2067 static void
2068 ri_GenerateQual(StringInfo buf,
2069 const char *sep,
2070 const char *leftop, Oid leftoptype,
2071 Oid opoid,
2072 const char *rightop, Oid rightoptype)
2073 {
2074 appendStringInfo(buf, " %s ", sep);
2075 generate_operator_clause(buf, leftop, leftoptype, opoid,
2076 rightop, rightoptype);
2077 }
2078
2079 /*
2080 * ri_GenerateQualCollation --- add a COLLATE spec to a WHERE clause
2081 *
2082 * We only have to use this function when directly comparing the referencing
2083 * and referenced columns, if they are of different collations; else the
2084 * parser will fail to resolve the collation to use. We don't need to use
2085 * this function for RI queries that compare a variable to a $n parameter.
2086 * Since parameter symbols always have default collation, the effect will be
2087 * to use the variable's collation.
2088 *
2089 * Note that we require that the collations of the referencing and the
2090 * referenced column have the same notion of equality: Either they have to
2091 * both be deterministic or else they both have to be the same. (See also
2092 * ATAddForeignKeyConstraint().)
2093 */
2094 static void
2095 ri_GenerateQualCollation(StringInfo buf, Oid collation)
2096 {
2097 HeapTuple tp;
2098 Form_pg_collation colltup;
2099 char *collname;
2100 char onename[MAX_QUOTED_NAME_LEN];
2101
2102 /* Nothing to do if it's a noncollatable data type */
2103 if (!OidIsValid(collation))
2104 return;
2105
2106 tp = SearchSysCache1(COLLOID, ObjectIdGetDatum(collation));
2107 if (!HeapTupleIsValid(tp))
2108 elog(ERROR, "cache lookup failed for collation %u", collation);
2109 colltup = (Form_pg_collation) GETSTRUCT(tp);
2110 collname = NameStr(colltup->collname);
2111
2112 /*
2113 * We qualify the name always, for simplicity and to ensure the query is
2114 * not search-path-dependent.
2115 */
2116 quoteOneName(onename, get_namespace_name(colltup->collnamespace));
2117 appendStringInfo(buf, " COLLATE %s", onename);
2118 quoteOneName(onename, collname);
2119 appendStringInfo(buf, ".%s", onename);
2120
2121 ReleaseSysCache(tp);
2122 }
2123
2124 /* ----------
2125 * ri_BuildQueryKey -
2126 *
2127 * Construct a hashtable key for a prepared SPI plan of an FK constraint.
2128 *
2129 * key: output argument, *key is filled in based on the other arguments
2130 * riinfo: info derived from pg_constraint entry
2131 * constr_queryno: an internal number identifying the query type
2132 * (see RI_PLAN_XXX constants at head of file)
2133 * ----------
2134 */
2135 static void
2136 ri_BuildQueryKey(RI_QueryKey *key, const RI_ConstraintInfo *riinfo,
2137 int32 constr_queryno)
2138 {
2139 /*
2140 * Inherited constraints with a common ancestor can share ri_query_cache
2141 * entries for all query types except RI_PLAN_CHECK_LOOKUPPK_FROM_PK.
2142 * Except in that case, the query processes the other table involved in
2143 * the FK constraint (i.e., not the table on which the trigger has been
2144 * fired), and so it will be the same for all members of the inheritance
2145 * tree. So we may use the root constraint's OID in the hash key, rather
2146 * than the constraint's own OID. This avoids creating duplicate SPI
2147 * plans, saving lots of work and memory when there are many partitions
2148 * with similar FK constraints.
2149 *
2150 * (Note that we must still have a separate RI_ConstraintInfo for each
2151 * constraint, because partitions can have different column orders,
2152 * resulting in different pk_attnums[] or fk_attnums[] array contents.)
2153 *
2154 * We assume struct RI_QueryKey contains no padding bytes, else we'd need
2155 * to use memset to clear them.
2156 */
2157 if (constr_queryno != RI_PLAN_CHECK_LOOKUPPK_FROM_PK)
2158 key->constr_id = riinfo->constraint_root_id;
2159 else
2160 key->constr_id = riinfo->constraint_id;
2161 key->constr_queryno = constr_queryno;
2162 }
2163
2164 /*
2165 * Check that RI trigger function was called in expected context
2166 */
2167 static void
2168 ri_CheckTrigger(FunctionCallInfo fcinfo, const char *funcname, int tgkind)
2169 {
2170 TriggerData *trigdata = (TriggerData *) fcinfo->context;
2171
2172 if (!CALLED_AS_TRIGGER(fcinfo))
2173 ereport(ERROR,
2174 (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
2175 errmsg("function \"%s\" was not called by trigger manager", funcname)));
2176
2177 /*
2178 * Check proper event
2179 */
2180 if (!TRIGGER_FIRED_AFTER(trigdata->tg_event) ||
2181 !TRIGGER_FIRED_FOR_ROW(trigdata->tg_event))
2182 ereport(ERROR,
2183 (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
2184 errmsg("function \"%s\" must be fired AFTER ROW", funcname)));
2185
2186 switch (tgkind)
2187 {
2188 case RI_TRIGTYPE_INSERT:
2189 if (!TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
2190 ereport(ERROR,
2191 (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
2192 errmsg("function \"%s\" must be fired for INSERT", funcname)));
2193 break;
2194 case RI_TRIGTYPE_UPDATE:
2195 if (!TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
2196 ereport(ERROR,
2197 (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
2198 errmsg("function \"%s\" must be fired for UPDATE", funcname)));
2199 break;
2200 case RI_TRIGTYPE_DELETE:
2201 if (!TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
2202 ereport(ERROR,
2203 (errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
2204 errmsg("function \"%s\" must be fired for DELETE", funcname)));
2205 break;
2206 }
2207 }
2208
2209
2210 /*
2211 * Fetch the RI_ConstraintInfo struct for the trigger's FK constraint.
2212 */
2213 static const RI_ConstraintInfo *
2214 ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
2215 {
2216 Oid constraintOid = trigger->tgconstraint;
2217 const RI_ConstraintInfo *riinfo;
2218
2219 /*
2220 * Check that the FK constraint's OID is available; it might not be if
2221 * we've been invoked via an ordinary trigger or an old-style "constraint
2222 * trigger".
2223 */
2224 if (!OidIsValid(constraintOid))
2225 ereport(ERROR,
2226 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
2227 errmsg("no pg_constraint entry for trigger \"%s\" on table \"%s\"",
2228 trigger->tgname, RelationGetRelationName(trig_rel)),
2229 errhint("Remove this referential integrity trigger and its mates, then do ALTER TABLE ADD CONSTRAINT.")));
2230
2231 /* Find or create a hashtable entry for the constraint */
2232 riinfo = ri_LoadConstraintInfo(constraintOid);
2233
2234 /* Do some easy cross-checks against the trigger call data */
2235 if (rel_is_pk)
2236 {
2237 if (riinfo->fk_relid != trigger->tgconstrrelid ||
2238 riinfo->pk_relid != RelationGetRelid(trig_rel))
2239 elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
2240 trigger->tgname, RelationGetRelationName(trig_rel));
2241 }
2242 else
2243 {
2244 if (riinfo->fk_relid != RelationGetRelid(trig_rel) ||
2245 riinfo->pk_relid != trigger->tgconstrrelid)
2246 elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
2247 trigger->tgname, RelationGetRelationName(trig_rel));
2248 }
2249
2250 if (riinfo->confmatchtype != FKCONSTR_MATCH_FULL &&
2251 riinfo->confmatchtype != FKCONSTR_MATCH_PARTIAL &&
2252 riinfo->confmatchtype != FKCONSTR_MATCH_SIMPLE)
2253 elog(ERROR, "unrecognized confmatchtype: %d",
2254 riinfo->confmatchtype);
2255
2256 if (riinfo->confmatchtype == FKCONSTR_MATCH_PARTIAL)
2257 ereport(ERROR,
2258 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2259 errmsg("MATCH PARTIAL not yet implemented")));
2260
2261 return riinfo;
2262 }
2263
2264 /*
2265 * Fetch or create the RI_ConstraintInfo struct for an FK constraint.
2266 */
2267 static const RI_ConstraintInfo *
2268 ri_LoadConstraintInfo(Oid constraintOid)
2269 {
2270 RI_ConstraintInfo *riinfo;
2271 bool found;
2272 HeapTuple tup;
2273 Form_pg_constraint conForm;
2274
2275 /*
2276 * On the first call initialize the hashtable
2277 */
2278 if (!ri_constraint_cache)
2279 ri_InitHashTables();
2280
2281 /*
2282 * Find or create a hash entry. If we find a valid one, just return it.
2283 */
2284 riinfo = (RI_ConstraintInfo *) hash_search(ri_constraint_cache,
2285 &constraintOid,
2286 HASH_ENTER, &found);
2287 if (!found)
2288 riinfo->valid = false;
2289 else if (riinfo->valid)
2290 return riinfo;
2291
2292 /*
2293 * Fetch the pg_constraint row so we can fill in the entry.
2294 */
2295 tup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constraintOid));
2296 if (!HeapTupleIsValid(tup)) /* should not happen */
2297 elog(ERROR, "cache lookup failed for constraint %u", constraintOid);
2298 conForm = (Form_pg_constraint) GETSTRUCT(tup);
2299
2300 if (conForm->contype != CONSTRAINT_FOREIGN) /* should not happen */
2301 elog(ERROR, "constraint %u is not a foreign key constraint",
2302 constraintOid);
2303
2304 /* And extract data */
2305 Assert(riinfo->constraint_id == constraintOid);
2306 if (OidIsValid(conForm->conparentid))
2307 riinfo->constraint_root_id =
2308 get_ri_constraint_root(conForm->conparentid);
2309 else
2310 riinfo->constraint_root_id = constraintOid;
2311 riinfo->oidHashValue = GetSysCacheHashValue1(CONSTROID,
2312 ObjectIdGetDatum(constraintOid));
2313 riinfo->rootHashValue = GetSysCacheHashValue1(CONSTROID,
2314 ObjectIdGetDatum(riinfo->constraint_root_id));
2315 memcpy(&riinfo->conname, &conForm->conname, sizeof(NameData));
2316 riinfo->pk_relid = conForm->confrelid;
2317 riinfo->fk_relid = conForm->conrelid;
2318 riinfo->confupdtype = conForm->confupdtype;
2319 riinfo->confdeltype = conForm->confdeltype;
2320 riinfo->confmatchtype = conForm->confmatchtype;
2321 riinfo->hasperiod = conForm->conperiod;
2322
2323 DeconstructFkConstraintRow(tup,
2324 &riinfo->nkeys,
2325 riinfo->fk_attnums,
2326 riinfo->pk_attnums,
2327 riinfo->pf_eq_oprs,
2328 riinfo->pp_eq_oprs,
2329 riinfo->ff_eq_oprs,
2330 &riinfo->ndelsetcols,
2331 riinfo->confdelsetcols);
2332
2333 /*
2334 * For temporal FKs, get the operators and functions we need. We ask the
2335 * opclass of the PK element for these. This all gets cached (as does the
2336 * generated plan), so there's no performance issue.
2337 */
2338 if (riinfo->hasperiod)
2339 {
2340 Oid opclass = get_index_column_opclass(conForm->conindid, riinfo->nkeys);
2341
2342 FindFKPeriodOpers(opclass,
2343 &riinfo->period_contained_by_oper,
2344 &riinfo->agged_period_contained_by_oper,
2345 &riinfo->period_intersect_oper);
2346 }
2347
2348 ReleaseSysCache(tup);
2349
2350 /*
2351 * For efficient processing of invalidation messages below, we keep a
2352 * doubly-linked count list of all currently valid entries.
2353 */
2354 dclist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);
2355
2356 riinfo->valid = true;
2357
2358 return riinfo;
2359 }
2360
2361 /*
2362 * get_ri_constraint_root
2363 * Returns the OID of the constraint's root parent
2364 */
2365 static Oid
2366 get_ri_constraint_root(Oid constrOid)
2367 {
2368 for (;;)
2369 {
2370 HeapTuple tuple;
2371 Oid constrParentOid;
2372
2373 tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constrOid));
2374 if (!HeapTupleIsValid(tuple))
2375 elog(ERROR, "cache lookup failed for constraint %u", constrOid);
2376 constrParentOid = ((Form_pg_constraint) GETSTRUCT(tuple))->conparentid;
2377 ReleaseSysCache(tuple);
2378 if (!OidIsValid(constrParentOid))
2379 break; /* we reached the root constraint */
2380 constrOid = constrParentOid;
2381 }
2382 return constrOid;
2383 }
2384
2385 /*
2386 * Callback for pg_constraint inval events
2387 *
2388 * While most syscache callbacks just flush all their entries, pg_constraint
2389 * gets enough update traffic that it's probably worth being smarter.
2390 * Invalidate any ri_constraint_cache entry associated with the syscache
2391 * entry with the specified hash value, or all entries if hashvalue == 0.
2392 *
2393 * Note: at the time a cache invalidation message is processed there may be
2394 * active references to the cache. Because of this we never remove entries
2395 * from the cache, but only mark them invalid, which is harmless to active
2396 * uses. (Any query using an entry should hold a lock sufficient to keep that
2397 * data from changing under it --- but we may get cache flushes anyway.)
2398 */
2399 static void
2400 InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
2401 {
2402 dlist_mutable_iter iter;
2403
2404 Assert(ri_constraint_cache != NULL);
2405
2406 /*
2407 * If the list of currently valid entries gets excessively large, we mark
2408 * them all invalid so we can empty the list. This arrangement avoids
2409 * O(N^2) behavior in situations where a session touches many foreign keys
2410 * and also does many ALTER TABLEs, such as a restore from pg_dump.
2411 */
2412 if (dclist_count(&ri_constraint_cache_valid_list) > 1000)
2413 hashvalue = 0; /* pretend it's a cache reset */
2414
2415 dclist_foreach_modify(iter, &ri_constraint_cache_valid_list)
2416 {
2417 RI_ConstraintInfo *riinfo = dclist_container(RI_ConstraintInfo,
2418 valid_link, iter.cur);
2419
2420 /*
2421 * We must invalidate not only entries directly matching the given
2422 * hash value, but also child entries, in case the invalidation
2423 * affects a root constraint.
2424 */
2425 if (hashvalue == 0 ||
2426 riinfo->oidHashValue == hashvalue ||
2427 riinfo->rootHashValue == hashvalue)
2428 {
2429 riinfo->valid = false;
2430 /* Remove invalidated entries from the list, too */
2431 dclist_delete_from(&ri_constraint_cache_valid_list, iter.cur);
2432 }
2433 }
2434 }
2435
2436
2437 /*
2438 * Prepare execution plan for a query to enforce an RI restriction
2439 */
2440 static SPIPlanPtr
2441 ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
2442 RI_QueryKey *qkey, Relation fk_rel, Relation pk_rel)
2443 {
2444 SPIPlanPtr qplan;
2445 Relation query_rel;
2446 Oid save_userid;
2447 int save_sec_context;
2448
2449 /*
2450 * Use the query type code to determine whether the query is run against
2451 * the PK or FK table; we'll do the check as that table's owner
2452 */
2453 if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK)
2454 query_rel = pk_rel;
2455 else
2456 query_rel = fk_rel;
2457
2458 /* Switch to proper UID to perform check as */
2459 GetUserIdAndSecContext(&save_userid, &save_sec_context);
2460 SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
2461 save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
2462 SECURITY_NOFORCE_RLS);
2463
2464 /* Create the plan */
2465 qplan = SPI_prepare(querystr, nargs, argtypes);
2466
2467 if (qplan == NULL)
2468 elog(ERROR, "SPI_prepare returned %s for %s", SPI_result_code_string(SPI_result), querystr);
2469
2470 /* Restore UID and security context */
2471 SetUserIdAndSecContext(save_userid, save_sec_context);
2472
2473 /* Save the plan */
2474 SPI_keepplan(qplan);
2475 ri_HashPreparedPlan(qkey, qplan);
2476
2477 return qplan;
2478 }
2479
2480 /*
2481 * Perform a query to enforce an RI restriction
2482 */
2483 static bool
2484 ri_PerformCheck(const RI_ConstraintInfo *riinfo,
2485 RI_QueryKey *qkey, SPIPlanPtr qplan,
2486 Relation fk_rel, Relation pk_rel,
2487 TupleTableSlot *oldslot, TupleTableSlot *newslot,
2488 bool is_restrict,
2489 bool detectNewRows, int expect_OK)
2490 {
2491 Relation query_rel,
2492 source_rel;
2493 bool source_is_pk;
2494 Snapshot test_snapshot;
2495 Snapshot crosscheck_snapshot;
2496 int limit;
2497 int spi_result;
2498 Oid save_userid;
2499 int save_sec_context;
2500 Datum vals[RI_MAX_NUMKEYS * 2];
2501 char nulls[RI_MAX_NUMKEYS * 2];
2502
2503 /*
2504 * Use the query type code to determine whether the query is run against
2505 * the PK or FK table; we'll do the check as that table's owner
2506 */
2507 if (qkey->constr_queryno <= RI_PLAN_LAST_ON_PK)
2508 query_rel = pk_rel;
2509 else
2510 query_rel = fk_rel;
2511
2512 /*
2513 * The values for the query are taken from the table on which the trigger
2514 * is called - it is normally the other one with respect to query_rel. An
2515 * exception is ri_Check_Pk_Match(), which uses the PK table for both (and
2516 * sets queryno to RI_PLAN_CHECK_LOOKUPPK_FROM_PK). We might eventually
2517 * need some less klugy way to determine this.
2518 */
2519 if (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK)
2520 {
2521 source_rel = fk_rel;
2522 source_is_pk = false;
2523 }
2524 else
2525 {
2526 source_rel = pk_rel;
2527 source_is_pk = true;
2528 }
2529
2530 /* Extract the parameters to be passed into the query */
2531 if (newslot)
2532 {
2533 ri_ExtractValues(source_rel, newslot, riinfo, source_is_pk,
2534 vals, nulls);
2535 if (oldslot)
2536 ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
2537 vals + riinfo->nkeys, nulls + riinfo->nkeys);
2538 }
2539 else
2540 {
2541 ri_ExtractValues(source_rel, oldslot, riinfo, source_is_pk,
2542 vals, nulls);
2543 }
2544
2545 /*
2546 * In READ COMMITTED mode, we just need to use an up-to-date regular
2547 * snapshot, and we will see all rows that could be interesting. But in
2548 * transaction-snapshot mode, we can't change the transaction snapshot. If
2549 * the caller passes detectNewRows == false then it's okay to do the query
2550 * with the transaction snapshot; otherwise we use a current snapshot, and
2551 * tell the executor to error out if it finds any rows under the current
2552 * snapshot that wouldn't be visible per the transaction snapshot. Note
2553 * that SPI_execute_snapshot will register the snapshots, so we don't need
2554 * to bother here.
2555 */
2556 if (IsolationUsesXactSnapshot() && detectNewRows)
2557 {
2558 CommandCounterIncrement(); /* be sure all my own work is visible */
2559 test_snapshot = GetLatestSnapshot();
2560 crosscheck_snapshot = GetTransactionSnapshot();
2561 }
2562 else
2563 {
2564 /* the default SPI behavior is okay */
2565 test_snapshot = InvalidSnapshot;
2566 crosscheck_snapshot = InvalidSnapshot;
2567 }
2568
2569 /*
2570 * If this is a select query (e.g., for a 'no action' or 'restrict'
2571 * trigger), we only need to see if there is a single row in the table,
2572 * matching the key. Otherwise, limit = 0 - because we want the query to
2573 * affect ALL the matching rows.
2574 */
2575 limit = (expect_OK == SPI_OK_SELECT) ? 1 : 0;
2576
2577 /* Switch to proper UID to perform check as */
2578 GetUserIdAndSecContext(&save_userid, &save_sec_context);
2579 SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
2580 save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
2581 SECURITY_NOFORCE_RLS);
2582
2583 /* Finally we can run the query. */
2584 spi_result = SPI_execute_snapshot(qplan,
2585 vals, nulls,
2586 test_snapshot, crosscheck_snapshot,
2587 false, false, limit);
2588
2589 /* Restore UID and security context */
2590 SetUserIdAndSecContext(save_userid, save_sec_context);
2591
2592 /* Check result */
2593 if (spi_result < 0)
2594 elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));
2595
2596 if (expect_OK >= 0 && spi_result != expect_OK)
2597 ereport(ERROR,
2598 (errcode(ERRCODE_INTERNAL_ERROR),
2599 errmsg("referential integrity query on \"%s\" from constraint \"%s\" on \"%s\" gave unexpected result",
2600 RelationGetRelationName(pk_rel),
2601 NameStr(riinfo->conname),
2602 RelationGetRelationName(fk_rel)),
2603 errhint("This is most likely due to a rule having rewritten the query.")));
2604
2605 /* XXX wouldn't it be clearer to do this part at the caller? */
2606 if (qkey->constr_queryno != RI_PLAN_CHECK_LOOKUPPK_FROM_PK &&
2607 expect_OK == SPI_OK_SELECT &&
2608 (SPI_processed == 0) == (qkey->constr_queryno == RI_PLAN_CHECK_LOOKUPPK))
2609 ri_ReportViolation(riinfo,
2610 pk_rel, fk_rel,
2611 newslot ? newslot : oldslot,
2612 NULL,
2613 qkey->constr_queryno, is_restrict, false);
2614
2615 return SPI_processed != 0;
2616 }
2617
2618 /*
2619 * Extract fields from a tuple into Datum/nulls arrays
2620 */
2621 static void
2622 ri_ExtractValues(Relation rel, TupleTableSlot *slot,
2623 const RI_ConstraintInfo *riinfo, bool rel_is_pk,
2624 Datum *vals, char *nulls)
2625 {
2626 const int16 *attnums;
2627 bool isnull;
2628
2629 if (rel_is_pk)
2630 attnums = riinfo->pk_attnums;
2631 else
2632 attnums = riinfo->fk_attnums;
2633
2634 for (int i = 0; i < riinfo->nkeys; i++)
2635 {
2636 vals[i] = slot_getattr(slot, attnums[i], &isnull);
2637 nulls[i] = isnull ? 'n' : ' ';
2638 }
2639 }
2640
2641 /*
2642 * Produce an error report
2643 *
2644 * If the failed constraint was on insert/update to the FK table,
2645 * we want the key names and values extracted from there, and the error
2646 * message to look like 'key blah is not present in PK'.
2647 * Otherwise, the attr names and values come from the PK table and the
2648 * message looks like 'key blah is still referenced from FK'.
2649 */
2650 static void
2651 ri_ReportViolation(const RI_ConstraintInfo *riinfo,
2652 Relation pk_rel, Relation fk_rel,
2653 TupleTableSlot *violatorslot, TupleDesc tupdesc,
2654 int queryno, bool is_restrict, bool partgone)
2655 {
2656 StringInfoData key_names;
2657 StringInfoData key_values;
2658 bool onfk;
2659 const int16 *attnums;
2660 Oid rel_oid;
2661 AclResult aclresult;
2662 bool has_perm = true;
2663
2664 /*
2665 * Determine which relation to complain about. If tupdesc wasn't passed
2666 * by caller, assume the violator tuple came from there.
2667 */
2668 onfk = (queryno == RI_PLAN_CHECK_LOOKUPPK);
2669 if (onfk)
2670 {
2671 attnums = riinfo->fk_attnums;
2672 rel_oid = fk_rel->rd_id;
2673 if (tupdesc == NULL)
2674 tupdesc = fk_rel->rd_att;
2675 }
2676 else
2677 {
2678 attnums = riinfo->pk_attnums;
2679 rel_oid = pk_rel->rd_id;
2680 if (tupdesc == NULL)
2681 tupdesc = pk_rel->rd_att;
2682 }
2683
2684 /*
2685 * Check permissions- if the user does not have access to view the data in
2686 * any of the key columns then we don't include the errdetail() below.
2687 *
2688 * Check if RLS is enabled on the relation first. If so, we don't return
2689 * any specifics to avoid leaking data.
2690 *
2691 * Check table-level permissions next and, failing that, column-level
2692 * privileges.
2693 *
2694 * When a partition at the referenced side is being detached/dropped, we
2695 * needn't check, since the user must be the table owner anyway.
2696 */
2697 if (partgone)
2698 has_perm = true;
2699 else if (check_enable_rls(rel_oid, InvalidOid, true) != RLS_ENABLED)
2700 {
2701 aclresult = pg_class_aclcheck(rel_oid, GetUserId(), ACL_SELECT);
2702 if (aclresult != ACLCHECK_OK)
2703 {
2704 /* Try for column-level permissions */
2705 for (int idx = 0; idx < riinfo->nkeys; idx++)
2706 {
2707 aclresult = pg_attribute_aclcheck(rel_oid, attnums[idx],
2708 GetUserId(),
2709 ACL_SELECT);
2710
2711 /* No access to the key */
2712 if (aclresult != ACLCHECK_OK)
2713 {
2714 has_perm = false;
2715 break;
2716 }
2717 }
2718 }
2719 }
2720 else
2721 has_perm = false;
2722
2723 if (has_perm)
2724 {
2725 /* Get printable versions of the keys involved */
2726 initStringInfo(&key_names);
2727 initStringInfo(&key_values);
2728 for (int idx = 0; idx < riinfo->nkeys; idx++)
2729 {
2730 int fnum = attnums[idx];
2731 Form_pg_attribute att = TupleDescAttr(tupdesc, fnum - 1);
2732 char *name,
2733 *val;
2734 Datum datum;
2735 bool isnull;
2736
2737 name = NameStr(att->attname);
2738
2739 datum = slot_getattr(violatorslot, fnum, &isnull);
2740 if (!isnull)
2741 {
2742 Oid foutoid;
2743 bool typisvarlena;
2744
2745 getTypeOutputInfo(att->atttypid, &foutoid, &typisvarlena);
2746 val = OidOutputFunctionCall(foutoid, datum);
2747 }
2748 else
2749 val = "null";
2750
2751 if (idx > 0)
2752 {
2753 appendStringInfoString(&key_names, ", ");
2754 appendStringInfoString(&key_values, ", ");
2755 }
2756 appendStringInfoString(&key_names, name);
2757 appendStringInfoString(&key_values, val);
2758 }
2759 }
2760
2761 if (partgone)
2762 ereport(ERROR,
2763 (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
2764 errmsg("removing partition \"%s\" violates foreign key constraint \"%s\"",
2765 RelationGetRelationName(pk_rel),
2766 NameStr(riinfo->conname)),
2767 errdetail("Key (%s)=(%s) is still referenced from table \"%s\".",
2768 key_names.data, key_values.data,
2769 RelationGetRelationName(fk_rel)),
2770 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
2771 else if (onfk)
2772 ereport(ERROR,
2773 (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
2774 errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
2775 RelationGetRelationName(fk_rel),
2776 NameStr(riinfo->conname)),
2777 has_perm ?
2778 errdetail("Key (%s)=(%s) is not present in table \"%s\".",
2779 key_names.data, key_values.data,
2780 RelationGetRelationName(pk_rel)) :
2781 errdetail("Key is not present in table \"%s\".",
2782 RelationGetRelationName(pk_rel)),
2783 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
2784 else if (is_restrict)
2785 ereport(ERROR,
2786 (errcode(ERRCODE_RESTRICT_VIOLATION),
2787 errmsg("update or delete on table \"%s\" violates RESTRICT setting of foreign key constraint \"%s\" on table \"%s\"",
2788 RelationGetRelationName(pk_rel),
2789 NameStr(riinfo->conname),
2790 RelationGetRelationName(fk_rel)),
2791 has_perm ?
2792 errdetail("Key (%s)=(%s) is referenced from table \"%s\".",
2793 key_names.data, key_values.data,
2794 RelationGetRelationName(fk_rel)) :
2795 errdetail("Key is referenced from table \"%s\".",
2796 RelationGetRelationName(fk_rel)),
2797 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
2798 else
2799 ereport(ERROR,
2800 (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
2801 errmsg("update or delete on table \"%s\" violates foreign key constraint \"%s\" on table \"%s\"",
2802 RelationGetRelationName(pk_rel),
2803 NameStr(riinfo->conname),
2804 RelationGetRelationName(fk_rel)),
2805 has_perm ?
2806 errdetail("Key (%s)=(%s) is still referenced from table \"%s\".",
2807 key_names.data, key_values.data,
2808 RelationGetRelationName(fk_rel)) :
2809 errdetail("Key is still referenced from table \"%s\".",
2810 RelationGetRelationName(fk_rel)),
2811 errtableconstraint(fk_rel, NameStr(riinfo->conname))));
2812 }
2813
2814
2815 /*
2816 * ri_NullCheck -
2817 *
2818 * Determine the NULL state of all key values in a tuple
2819 *
2820 * Returns one of RI_KEYS_ALL_NULL, RI_KEYS_NONE_NULL or RI_KEYS_SOME_NULL.
2821 */
2822 static int
2823 ri_NullCheck(TupleDesc tupDesc,
2824 TupleTableSlot *slot,
2825 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
2826 {
2827 const int16 *attnums;
2828 bool allnull = true;
2829 bool nonenull = true;
2830
2831 if (rel_is_pk)
2832 attnums = riinfo->pk_attnums;
2833 else
2834 attnums = riinfo->fk_attnums;
2835
2836 for (int i = 0; i < riinfo->nkeys; i++)
2837 {
2838 if (slot_attisnull(slot, attnums[i]))
2839 nonenull = false;
2840 else
2841 allnull = false;
2842 }
2843
2844 if (allnull)
2845 return RI_KEYS_ALL_NULL;
2846
2847 if (nonenull)
2848 return RI_KEYS_NONE_NULL;
2849
2850 return RI_KEYS_SOME_NULL;
2851 }
2852
2853
2854 /*
2855 * ri_InitHashTables -
2856 *
2857 * Initialize our internal hash tables.
2858 */
2859 static void
2860 ri_InitHashTables(void)
2861 {
2862 HASHCTL ctl;
2863
2864 ctl.keysize = sizeof(Oid);
2865 ctl.entrysize = sizeof(RI_ConstraintInfo);
2866 ri_constraint_cache = hash_create("RI constraint cache",
2867 RI_INIT_CONSTRAINTHASHSIZE,
2868 &ctl, HASH_ELEM | HASH_BLOBS);
2869
2870 /* Arrange to flush cache on pg_constraint changes */
2871 CacheRegisterSyscacheCallback(CONSTROID,
2872 InvalidateConstraintCacheCallBack,
2873 (Datum) 0);
2874
2875 ctl.keysize = sizeof(RI_QueryKey);
2876 ctl.entrysize = sizeof(RI_QueryHashEntry);
2877 ri_query_cache = hash_create("RI query cache",
2878 RI_INIT_QUERYHASHSIZE,
2879 &ctl, HASH_ELEM | HASH_BLOBS);
2880
2881 ctl.keysize = sizeof(RI_CompareKey);
2882 ctl.entrysize = sizeof(RI_CompareHashEntry);
2883 ri_compare_cache = hash_create("RI compare cache",
2884 RI_INIT_QUERYHASHSIZE,
2885 &ctl, HASH_ELEM | HASH_BLOBS);
2886 }
2887
2888
2889 /*
2890 * ri_FetchPreparedPlan -
2891 *
2892 * Lookup for a query key in our private hash table of prepared
2893 * and saved SPI execution plans. Return the plan if found or NULL.
2894 */
2895 static SPIPlanPtr
2896 ri_FetchPreparedPlan(RI_QueryKey *key)
2897 {
2898 RI_QueryHashEntry *entry;
2899 SPIPlanPtr plan;
2900
2901 /*
2902 * On the first call initialize the hashtable
2903 */
2904 if (!ri_query_cache)
2905 ri_InitHashTables();
2906
2907 /*
2908 * Lookup for the key
2909 */
2910 entry = (RI_QueryHashEntry *) hash_search(ri_query_cache,
2911 key,
2912 HASH_FIND, NULL);
2913 if (entry == NULL)
2914 return NULL;
2915
2916 /*
2917 * Check whether the plan is still valid. If it isn't, we don't want to
2918 * simply rely on plancache.c to regenerate it; rather we should start
2919 * from scratch and rebuild the query text too. This is to cover cases
2920 * such as table/column renames. We depend on the plancache machinery to
2921 * detect possible invalidations, though.
2922 *
2923 * CAUTION: this check is only trustworthy if the caller has already
2924 * locked both FK and PK rels.
2925 */
2926 plan = entry->plan;
2927 if (plan && SPI_plan_is_valid(plan))
2928 return plan;
2929
2930 /*
2931 * Otherwise we might as well flush the cached plan now, to free a little
2932 * memory space before we make a new one.
2933 */
2934 entry->plan = NULL;
2935 if (plan)
2936 SPI_freeplan(plan);
2937
2938 return NULL;
2939 }
2940
2941
2942 /*
2943 * ri_HashPreparedPlan -
2944 *
2945 * Add another plan to our private SPI query plan hashtable.
2946 */
2947 static void
2948 ri_HashPreparedPlan(RI_QueryKey *key, SPIPlanPtr plan)
2949 {
2950 RI_QueryHashEntry *entry;
2951 bool found;
2952
2953 /*
2954 * On the first call initialize the hashtable
2955 */
2956 if (!ri_query_cache)
2957 ri_InitHashTables();
2958
2959 /*
2960 * Add the new plan. We might be overwriting an entry previously found
2961 * invalid by ri_FetchPreparedPlan.
2962 */
2963 entry = (RI_QueryHashEntry *) hash_search(ri_query_cache,
2964 key,
2965 HASH_ENTER, &found);
2966 Assert(!found || entry->plan == NULL);
2967 entry->plan = plan;
2968 }
2969
2970
2971 /*
2972 * ri_KeysEqual -
2973 *
2974 * Check if all key values in OLD and NEW are "equivalent":
2975 * For normal FKs we check for equality.
2976 * For temporal FKs we check that the PK side is a superset of its old value,
2977 * or the FK side is a subset of its old value.
2978 *
2979 * Note: at some point we might wish to redefine this as checking for
2980 * "IS NOT DISTINCT" rather than "=", that is, allow two nulls to be
2981 * considered equal. Currently there is no need since all callers have
2982 * previously found at least one of the rows to contain no nulls.
2983 */
2984 static bool
2985 ri_KeysEqual(Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot,
2986 const RI_ConstraintInfo *riinfo, bool rel_is_pk)
2987 {
2988 const int16 *attnums;
2989
2990 if (rel_is_pk)
2991 attnums = riinfo->pk_attnums;
2992 else
2993 attnums = riinfo->fk_attnums;
2994
2995 /* XXX: could be worthwhile to fetch all necessary attrs at once */
2996 for (int i = 0; i < riinfo->nkeys; i++)
2997 {
2998 Datum oldvalue;
2999 Datum newvalue;
3000 bool isnull;
3001
3002 /*
3003 * Get one attribute's oldvalue. If it is NULL - they're not equal.
3004 */
3005 oldvalue = slot_getattr(oldslot, attnums[i], &isnull);
3006 if (isnull)
3007 return false;
3008
3009 /*
3010 * Get one attribute's newvalue. If it is NULL - they're not equal.
3011 */
3012 newvalue = slot_getattr(newslot, attnums[i], &isnull);
3013 if (isnull)
3014 return false;
3015
3016 if (rel_is_pk)
3017 {
3018 /*
3019 * If we are looking at the PK table, then do a bytewise
3020 * comparison. We must propagate PK changes if the value is
3021 * changed to one that "looks" different but would compare as
3022 * equal using the equality operator. This only makes a
3023 * difference for ON UPDATE CASCADE, but for consistency we treat
3024 * all changes to the PK the same.
3025 */
3026 CompactAttribute *att = TupleDescCompactAttr(oldslot->tts_tupleDescriptor, attnums[i] - 1);
3027
3028 if (!datum_image_eq(oldvalue, newvalue, att->attbyval, att->attlen))
3029 return false;
3030 }
3031 else
3032 {
3033 Oid eq_opr;
3034
3035 /*
3036 * When comparing the PERIOD columns we can skip the check
3037 * whenever the referencing column stayed equal or shrank, so test
3038 * with the contained-by operator instead.
3039 */
3040 if (riinfo->hasperiod && i == riinfo->nkeys - 1)
3041 eq_opr = riinfo->period_contained_by_oper;
3042 else
3043 eq_opr = riinfo->ff_eq_oprs[i];
3044
3045 /*
3046 * For the FK table, compare with the appropriate equality
3047 * operator. Changes that compare equal will still satisfy the
3048 * constraint after the update.
3049 */
3050 if (!ri_CompareWithCast(eq_opr, RIAttType(rel, attnums[i]), RIAttCollation(rel, attnums[i]),
3051 newvalue, oldvalue))
3052 return false;
3053 }
3054 }
3055
3056 return true;
3057 }
3058
3059
3060 /*
3061 * ri_CompareWithCast -
3062 *
3063 * Call the appropriate comparison operator for two values.
3064 * Normally this is equality, but for the PERIOD part of foreign keys
3065 * it is ContainedBy, so the order of lhs vs rhs is significant.
3066 * See below for how the collation is applied.
3067 *
3068 * NB: we have already checked that neither value is null.
3069 */
3070 static bool
3071 ri_CompareWithCast(Oid eq_opr, Oid typeid, Oid collid,
3072 Datum lhs, Datum rhs)
3073 {
3074 RI_CompareHashEntry *entry = ri_HashCompareOp(eq_opr, typeid);
3075
3076 /* Do we need to cast the values? */
3077 if (OidIsValid(entry->cast_func_finfo.fn_oid))
3078 {
3079 lhs = FunctionCall3(&entry->cast_func_finfo,
3080 lhs,
3081 Int32GetDatum(-1), /* typmod */
3082 BoolGetDatum(false)); /* implicit coercion */
3083 rhs = FunctionCall3(&entry->cast_func_finfo,
3084 rhs,
3085 Int32GetDatum(-1), /* typmod */
3086 BoolGetDatum(false)); /* implicit coercion */
3087 }
3088
3089 /*
3090 * Apply the comparison operator.
3091 *
3092 * Note: This function is part of a call stack that determines whether an
3093 * update to a row is significant enough that it needs checking or action
3094 * on the other side of a foreign-key constraint. Therefore, the
3095 * comparison here would need to be done with the collation of the *other*
3096 * table. For simplicity (e.g., we might not even have the other table
3097 * open), we'll use our own collation. This is fine because we require
3098 * that both collations have the same notion of equality (either they are
3099 * both deterministic or else they are both the same).
3100 *
3101 * With range/multirangetypes, the collation of the base type is stored as
3102 * part of the rangetype (pg_range.rngcollation), and always used, so
3103 * there is no danger of inconsistency even using a non-equals operator.
3104 * But if we support arbitrary types with PERIOD, we should perhaps just
3105 * always force a re-check.
3106 */
3107 return DatumGetBool(FunctionCall2Coll(&entry->eq_opr_finfo, collid, lhs, rhs));
3108 }
3109
3110 /*
3111 * ri_HashCompareOp -
3112 *
3113 * See if we know how to compare two values, and create a new hash entry
3114 * if not.
3115 */
3116 static RI_CompareHashEntry *
3117 ri_HashCompareOp(Oid eq_opr, Oid typeid)
3118 {
3119 RI_CompareKey key;
3120 RI_CompareHashEntry *entry;
3121 bool found;
3122
3123 /*
3124 * On the first call initialize the hashtable
3125 */
3126 if (!ri_compare_cache)
3127 ri_InitHashTables();
3128
3129 /*
3130 * Find or create a hash entry. Note we're assuming RI_CompareKey
3131 * contains no struct padding.
3132 */
3133 key.eq_opr = eq_opr;
3134 key.typeid = typeid;
3135 entry = (RI_CompareHashEntry *) hash_search(ri_compare_cache,
3136 &key,
3137 HASH_ENTER, &found);
3138 if (!found)
3139 entry->valid = false;
3140
3141 /*
3142 * If not already initialized, do so. Since we'll keep this hash entry
3143 * for the life of the backend, put any subsidiary info for the function
3144 * cache structs into TopMemoryContext.
3145 */
3146 if (!entry->valid)
3147 {
3148 Oid lefttype,
3149 righttype,
3150 castfunc;
3151 CoercionPathType pathtype;
3152
3153 /* We always need to know how to call the equality operator */
3154 fmgr_info_cxt(get_opcode(eq_opr), &entry->eq_opr_finfo,
3155 TopMemoryContext);
3156
3157 /*
3158 * If we chose to use a cast from FK to PK type, we may have to apply
3159 * the cast function to get to the operator's input type.
3160 *
3161 * XXX eventually it would be good to support array-coercion cases
3162 * here and in ri_CompareWithCast(). At the moment there is no point
3163 * because cases involving nonidentical array types will be rejected
3164 * at constraint creation time.
3165 *
3166 * XXX perhaps also consider supporting CoerceViaIO? No need at the
3167 * moment since that will never be generated for implicit coercions.
3168 */
3169 op_input_types(eq_opr, &lefttype, &righttype);
3170 Assert(lefttype == righttype);
3171 if (typeid == lefttype)
3172 castfunc = InvalidOid; /* simplest case */
3173 else
3174 {
3175 pathtype = find_coercion_pathway(lefttype, typeid,
3176 COERCION_IMPLICIT,
3177 &castfunc);
3178 if (pathtype != COERCION_PATH_FUNC &&
3179 pathtype != COERCION_PATH_RELABELTYPE)
3180 {
3181 /*
3182 * The declared input type of the eq_opr might be a
3183 * polymorphic type such as ANYARRAY or ANYENUM, or other
3184 * special cases such as RECORD; find_coercion_pathway
3185 * currently doesn't subsume these special cases.
3186 */
3187 if (!IsBinaryCoercible(typeid, lefttype))
3188 elog(ERROR, "no conversion function from %s to %s",
3189 format_type_be(typeid),
3190 format_type_be(lefttype));
3191 }
3192 }
3193 if (OidIsValid(castfunc))
3194 fmgr_info_cxt(castfunc, &entry->cast_func_finfo,
3195 TopMemoryContext);
3196 else
3197 entry->cast_func_finfo.fn_oid = InvalidOid;
3198 entry->valid = true;
3199 }
3200
3201 return entry;
3202 }
3203
3204
3205 /*
3206 * Given a trigger function OID, determine whether it is an RI trigger,
3207 * and if so whether it is attached to PK or FK relation.
3208 */
3209 int
3210 RI_FKey_trigger_type(Oid tgfoid)
3211 {
3212 switch (tgfoid)
3213 {
3214 case F_RI_FKEY_CASCADE_DEL:
3215 case F_RI_FKEY_CASCADE_UPD:
3216 case F_RI_FKEY_RESTRICT_DEL:
3217 case F_RI_FKEY_RESTRICT_UPD:
3218 case F_RI_FKEY_SETNULL_DEL:
3219 case F_RI_FKEY_SETNULL_UPD:
3220 case F_RI_FKEY_SETDEFAULT_DEL:
3221 case F_RI_FKEY_SETDEFAULT_UPD:
3222 case F_RI_FKEY_NOACTION_DEL:
3223 case F_RI_FKEY_NOACTION_UPD:
3224 return RI_TRIGGER_PK;
3225
3226 case F_RI_FKEY_CHECK_INS:
3227 case F_RI_FKEY_CHECK_UPD:
3228 return RI_TRIGGER_FK;
3229 }
3230
3231 return RI_TRIGGER_NONE;
3232 }
Postgresql Clone from git://git.postgresql.org/git/postgresql.git