TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * pg_constraint.c
4 : * routines to support manipulation of the pg_constraint relation
5 : *
6 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/catalog/pg_constraint.c
12 : *
13 : *-------------------------------------------------------------------------
14 : */
15 : #include "postgres.h"
16 :
17 : #include "access/genam.h"
18 : #include "access/htup_details.h"
19 : #include "access/sysattr.h"
20 : #include "access/table.h"
21 : #include "access/xact.h"
22 : #include "catalog/catalog.h"
23 : #include "catalog/dependency.h"
24 : #include "catalog/indexing.h"
25 : #include "catalog/objectaccess.h"
26 : #include "catalog/pg_constraint.h"
27 : #include "catalog/pg_operator.h"
28 : #include "catalog/pg_type.h"
29 : #include "commands/defrem.h"
30 : #include "commands/tablecmds.h"
31 : #include "utils/array.h"
32 : #include "utils/builtins.h"
33 : #include "utils/fmgroids.h"
34 : #include "utils/lsyscache.h"
35 : #include "utils/rel.h"
36 : #include "utils/syscache.h"
37 :
38 :
39 : /*
40 : * CreateConstraintEntry
41 : * Create a constraint table entry.
42 : *
43 : * Subsidiary records (such as triggers or indexes to implement the
44 : * constraint) are *not* created here. But we do make dependency links
45 : * from the constraint to the things it depends on.
46 : *
47 : * The new constraint's OID is returned.
48 : */
49 : Oid
50 CBC 44635 : CreateConstraintEntry(const char *constraintName,
51 : Oid constraintNamespace,
52 : char constraintType,
53 : bool isDeferrable,
54 : bool isDeferred,
55 : bool isValidated,
56 : Oid parentConstrId,
57 : Oid relId,
58 : const int16 *constraintKey,
59 : int constraintNKeys,
60 : int constraintNTotalKeys,
61 : Oid domainId,
62 : Oid indexRelId,
63 : Oid foreignRelId,
64 : const int16 *foreignKey,
65 : const Oid *pfEqOp,
66 : const Oid *ppEqOp,
67 : const Oid *ffEqOp,
68 : int foreignNKeys,
69 : char foreignUpdateType,
70 : char foreignDeleteType,
71 : const int16 *fkDeleteSetCols,
72 : int numFkDeleteSetCols,
73 : char foreignMatchType,
74 : const Oid *exclOp,
75 : Node *conExpr,
76 : const char *conBin,
77 : bool conIsLocal,
78 : int conInhCount,
79 : bool conNoInherit,
80 : bool is_internal)
81 : {
82 : Relation conDesc;
83 : Oid conOid;
84 : HeapTuple tup;
85 : bool nulls[Natts_pg_constraint];
86 : Datum values[Natts_pg_constraint];
87 : ArrayType *conkeyArray;
88 : ArrayType *confkeyArray;
89 : ArrayType *conpfeqopArray;
90 : ArrayType *conppeqopArray;
91 : ArrayType *conffeqopArray;
92 : ArrayType *conexclopArray;
93 : ArrayType *confdelsetcolsArray;
94 : NameData cname;
95 : int i;
96 : ObjectAddress conobject;
97 : ObjectAddresses *addrs_auto;
98 : ObjectAddresses *addrs_normal;
99 :
100 44635 : conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
101 :
102 44635 : Assert(constraintName);
103 44635 : namestrcpy(&cname, constraintName);
104 :
105 : /*
106 : * Convert C arrays into Postgres arrays.
107 : */
108 44635 : if (constraintNKeys > 0)
109 : {
110 : Datum *conkey;
111 :
112 43749 : conkey = (Datum *) palloc(constraintNKeys * sizeof(Datum));
113 109970 : for (i = 0; i < constraintNKeys; i++)
114 66221 : conkey[i] = Int16GetDatum(constraintKey[i]);
115 GNC 43749 : conkeyArray = construct_array_builtin(conkey, constraintNKeys, INT2OID);
116 : }
117 ECB : else
118 GIC 886 : conkeyArray = NULL;
119 ECB :
120 GIC 44635 : if (foreignNKeys > 0)
121 : {
122 : Datum *fkdatums;
123 ECB :
124 CBC 1553 : fkdatums = (Datum *) palloc(foreignNKeys * sizeof(Datum));
125 3445 : for (i = 0; i < foreignNKeys; i++)
126 1892 : fkdatums[i] = Int16GetDatum(foreignKey[i]);
127 GNC 1553 : confkeyArray = construct_array_builtin(fkdatums, foreignNKeys, INT2OID);
128 CBC 3445 : for (i = 0; i < foreignNKeys; i++)
129 1892 : fkdatums[i] = ObjectIdGetDatum(pfEqOp[i]);
130 GNC 1553 : conpfeqopArray = construct_array_builtin(fkdatums, foreignNKeys, OIDOID);
131 CBC 3445 : for (i = 0; i < foreignNKeys; i++)
132 1892 : fkdatums[i] = ObjectIdGetDatum(ppEqOp[i]);
133 GNC 1553 : conppeqopArray = construct_array_builtin(fkdatums, foreignNKeys, OIDOID);
134 CBC 3445 : for (i = 0; i < foreignNKeys; i++)
135 GIC 1892 : fkdatums[i] = ObjectIdGetDatum(ffEqOp[i]);
136 GNC 1553 : conffeqopArray = construct_array_builtin(fkdatums, foreignNKeys, OIDOID);
137 ECB :
138 GIC 1553 : if (numFkDeleteSetCols > 0)
139 : {
140 CBC 60 : for (i = 0; i < numFkDeleteSetCols; i++)
141 GIC 30 : fkdatums[i] = Int16GetDatum(fkDeleteSetCols[i]);
142 GNC 30 : confdelsetcolsArray = construct_array_builtin(fkdatums, numFkDeleteSetCols, INT2OID);
143 ECB : }
144 : else
145 CBC 1523 : confdelsetcolsArray = NULL;
146 ECB : }
147 : else
148 : {
149 GIC 43082 : confkeyArray = NULL;
150 CBC 43082 : conpfeqopArray = NULL;
151 GIC 43082 : conppeqopArray = NULL;
152 43082 : conffeqopArray = NULL;
153 43082 : confdelsetcolsArray = NULL;
154 ECB : }
155 :
156 CBC 44635 : if (exclOp != NULL)
157 ECB : {
158 : Datum *opdatums;
159 :
160 CBC 64 : opdatums = (Datum *) palloc(constraintNKeys * sizeof(Datum));
161 GIC 143 : for (i = 0; i < constraintNKeys; i++)
162 79 : opdatums[i] = ObjectIdGetDatum(exclOp[i]);
163 GNC 64 : conexclopArray = construct_array_builtin(opdatums, constraintNKeys, OIDOID);
164 ECB : }
165 : else
166 GIC 44571 : conexclopArray = NULL;
167 :
168 ECB : /* initialize nulls and values */
169 GIC 1205145 : for (i = 0; i < Natts_pg_constraint; i++)
170 ECB : {
171 CBC 1160510 : nulls[i] = false;
172 1160510 : values[i] = (Datum) NULL;
173 ECB : }
174 :
175 CBC 44635 : conOid = GetNewOidWithIndex(conDesc, ConstraintOidIndexId,
176 ECB : Anum_pg_constraint_oid);
177 CBC 44635 : values[Anum_pg_constraint_oid - 1] = ObjectIdGetDatum(conOid);
178 44635 : values[Anum_pg_constraint_conname - 1] = NameGetDatum(&cname);
179 44635 : values[Anum_pg_constraint_connamespace - 1] = ObjectIdGetDatum(constraintNamespace);
180 44635 : values[Anum_pg_constraint_contype - 1] = CharGetDatum(constraintType);
181 44635 : values[Anum_pg_constraint_condeferrable - 1] = BoolGetDatum(isDeferrable);
182 44635 : values[Anum_pg_constraint_condeferred - 1] = BoolGetDatum(isDeferred);
183 44635 : values[Anum_pg_constraint_convalidated - 1] = BoolGetDatum(isValidated);
184 44635 : values[Anum_pg_constraint_conrelid - 1] = ObjectIdGetDatum(relId);
185 44635 : values[Anum_pg_constraint_contypid - 1] = ObjectIdGetDatum(domainId);
186 44635 : values[Anum_pg_constraint_conindid - 1] = ObjectIdGetDatum(indexRelId);
187 44635 : values[Anum_pg_constraint_conparentid - 1] = ObjectIdGetDatum(parentConstrId);
188 GIC 44635 : values[Anum_pg_constraint_confrelid - 1] = ObjectIdGetDatum(foreignRelId);
189 CBC 44635 : values[Anum_pg_constraint_confupdtype - 1] = CharGetDatum(foreignUpdateType);
190 44635 : values[Anum_pg_constraint_confdeltype - 1] = CharGetDatum(foreignDeleteType);
191 GIC 44635 : values[Anum_pg_constraint_confmatchtype - 1] = CharGetDatum(foreignMatchType);
192 CBC 44635 : values[Anum_pg_constraint_conislocal - 1] = BoolGetDatum(conIsLocal);
193 GNC 44635 : values[Anum_pg_constraint_coninhcount - 1] = Int16GetDatum(conInhCount);
194 CBC 44635 : values[Anum_pg_constraint_connoinherit - 1] = BoolGetDatum(conNoInherit);
195 ECB :
196 GIC 44635 : if (conkeyArray)
197 CBC 43749 : values[Anum_pg_constraint_conkey - 1] = PointerGetDatum(conkeyArray);
198 : else
199 886 : nulls[Anum_pg_constraint_conkey - 1] = true;
200 ECB :
201 GIC 44635 : if (confkeyArray)
202 CBC 1553 : values[Anum_pg_constraint_confkey - 1] = PointerGetDatum(confkeyArray);
203 : else
204 43082 : nulls[Anum_pg_constraint_confkey - 1] = true;
205 ECB :
206 GIC 44635 : if (conpfeqopArray)
207 CBC 1553 : values[Anum_pg_constraint_conpfeqop - 1] = PointerGetDatum(conpfeqopArray);
208 : else
209 43082 : nulls[Anum_pg_constraint_conpfeqop - 1] = true;
210 ECB :
211 GIC 44635 : if (conppeqopArray)
212 CBC 1553 : values[Anum_pg_constraint_conppeqop - 1] = PointerGetDatum(conppeqopArray);
213 : else
214 43082 : nulls[Anum_pg_constraint_conppeqop - 1] = true;
215 ECB :
216 GIC 44635 : if (conffeqopArray)
217 CBC 1553 : values[Anum_pg_constraint_conffeqop - 1] = PointerGetDatum(conffeqopArray);
218 : else
219 43082 : nulls[Anum_pg_constraint_conffeqop - 1] = true;
220 ECB :
221 GIC 44635 : if (confdelsetcolsArray)
222 CBC 30 : values[Anum_pg_constraint_confdelsetcols - 1] = PointerGetDatum(confdelsetcolsArray);
223 : else
224 44605 : nulls[Anum_pg_constraint_confdelsetcols - 1] = true;
225 ECB :
226 GIC 44635 : if (conexclopArray)
227 CBC 64 : values[Anum_pg_constraint_conexclop - 1] = PointerGetDatum(conexclopArray);
228 : else
229 44571 : nulls[Anum_pg_constraint_conexclop - 1] = true;
230 :
231 44635 : if (conBin)
232 GIC 1780 : values[Anum_pg_constraint_conbin - 1] = CStringGetTextDatum(conBin);
233 ECB : else
234 GIC 42855 : nulls[Anum_pg_constraint_conbin - 1] = true;
235 ECB :
236 GIC 44635 : tup = heap_form_tuple(RelationGetDescr(conDesc), values, nulls);
237 :
238 CBC 44635 : CatalogTupleInsert(conDesc, tup);
239 :
240 44635 : ObjectAddressSet(conobject, ConstraintRelationId, conOid);
241 :
242 GIC 44635 : table_close(conDesc, RowExclusiveLock);
243 :
244 : /* Handle set of auto dependencies */
245 44635 : addrs_auto = new_object_addresses();
246 :
247 44635 : if (OidIsValid(relId))
248 ECB : {
249 : /*
250 : * Register auto dependency from constraint to owning relation, or to
251 : * specific column(s) if any are mentioned.
252 : */
253 : ObjectAddress relobject;
254 :
255 GIC 43831 : if (constraintNTotalKeys > 0)
256 : {
257 110124 : for (i = 0; i < constraintNTotalKeys; i++)
258 : {
259 CBC 66375 : ObjectAddressSubSet(relobject, RelationRelationId, relId,
260 ECB : constraintKey[i]);
261 GIC 66375 : add_exact_object_address(&relobject, addrs_auto);
262 : }
263 : }
264 ECB : else
265 : {
266 GIC 82 : ObjectAddressSet(relobject, RelationRelationId, relId);
267 82 : add_exact_object_address(&relobject, addrs_auto);
268 : }
269 : }
270 :
271 CBC 44635 : if (OidIsValid(domainId))
272 ECB : {
273 : /*
274 : * Register auto dependency from constraint to owning domain
275 : */
276 : ObjectAddress domobject;
277 :
278 GIC 804 : ObjectAddressSet(domobject, TypeRelationId, domainId);
279 804 : add_exact_object_address(&domobject, addrs_auto);
280 ECB : }
281 :
282 CBC 44635 : record_object_address_dependencies(&conobject, addrs_auto,
283 : DEPENDENCY_AUTO);
284 GIC 44635 : free_object_addresses(addrs_auto);
285 :
286 : /* Handle set of normal dependencies */
287 44635 : addrs_normal = new_object_addresses();
288 :
289 44635 : if (OidIsValid(foreignRelId))
290 ECB : {
291 : /*
292 : * Register normal dependency from constraint to foreign relation, or
293 : * to specific column(s) if any are mentioned.
294 : */
295 : ObjectAddress relobject;
296 :
297 GIC 1553 : if (foreignNKeys > 0)
298 : {
299 3445 : for (i = 0; i < foreignNKeys; i++)
300 : {
301 GBC 1892 : ObjectAddressSubSet(relobject, RelationRelationId,
302 EUB : foreignRelId, foreignKey[i]);
303 GIC 1892 : add_exact_object_address(&relobject, addrs_normal);
304 : }
305 : }
306 ECB : else
307 : {
308 UIC 0 : ObjectAddressSet(relobject, RelationRelationId, foreignRelId);
309 0 : add_exact_object_address(&relobject, addrs_normal);
310 : }
311 : }
312 :
313 GIC 44635 : if (OidIsValid(indexRelId) && constraintType == CONSTRAINT_FOREIGN)
314 : {
315 : /*
316 ECB : * Register normal dependency on the unique index that supports a
317 : * foreign-key constraint. (Note: for indexes associated with unique
318 : * or primary-key constraints, the dependency runs the other way, and
319 : * is not made here.)
320 : */
321 : ObjectAddress relobject;
322 :
323 GIC 1553 : ObjectAddressSet(relobject, RelationRelationId, indexRelId);
324 1553 : add_exact_object_address(&relobject, addrs_normal);
325 : }
326 :
327 44635 : if (foreignNKeys > 0)
328 : {
329 : /*
330 ECB : * Register normal dependencies on the equality operators that support
331 : * a foreign-key constraint. If the PK and FK types are the same then
332 : * all three operators for a column are the same; otherwise they are
333 : * different.
334 : */
335 : ObjectAddress oprobject;
336 :
337 CBC 1553 : oprobject.classId = OperatorRelationId;
338 GIC 1553 : oprobject.objectSubId = 0;
339 ECB :
340 CBC 3445 : for (i = 0; i < foreignNKeys; i++)
341 : {
342 1892 : oprobject.objectId = pfEqOp[i];
343 GIC 1892 : add_exact_object_address(&oprobject, addrs_normal);
344 CBC 1892 : if (ppEqOp[i] != pfEqOp[i])
345 ECB : {
346 GIC 28 : oprobject.objectId = ppEqOp[i];
347 28 : add_exact_object_address(&oprobject, addrs_normal);
348 : }
349 1892 : if (ffEqOp[i] != pfEqOp[i])
350 ECB : {
351 GIC 28 : oprobject.objectId = ffEqOp[i];
352 CBC 28 : add_exact_object_address(&oprobject, addrs_normal);
353 : }
354 : }
355 : }
356 :
357 GIC 44635 : record_object_address_dependencies(&conobject, addrs_normal,
358 : DEPENDENCY_NORMAL);
359 44635 : free_object_addresses(addrs_normal);
360 :
361 : /*
362 ECB : * We don't bother to register dependencies on the exclusion operators of
363 : * an exclusion constraint. We assume they are members of the opclass
364 : * supporting the index, so there's an indirect dependency via that. (This
365 : * would be pretty dicey for cross-type operators, but exclusion operators
366 : * can never be cross-type.)
367 : */
368 :
369 GIC 44635 : if (conExpr != NULL)
370 : {
371 : /*
372 : * Register dependencies from constraint to objects mentioned in CHECK
373 : * expression.
374 ECB : */
375 GIC 1780 : recordDependencyOnSingleRelExpr(&conobject, conExpr, relId,
376 : DEPENDENCY_NORMAL,
377 ECB : DEPENDENCY_NORMAL, false);
378 : }
379 :
380 : /* Post creation hook for new constraint */
381 GIC 44635 : InvokeObjectPostCreateHookArg(ConstraintRelationId, conOid, 0,
382 : is_internal);
383 :
384 44635 : return conOid;
385 : }
386 :
387 : /*
388 : * Test whether given name is currently used as a constraint name
389 : * for the given object (relation or domain).
390 : *
391 : * This is used to decide whether to accept a user-specified constraint name.
392 : * It is deliberately not the same test as ChooseConstraintName uses to decide
393 ECB : * whether an auto-generated name is OK: here, we will allow it unless there
394 : * is an identical constraint name in use *on the same object*.
395 : *
396 : * NB: Caller should hold exclusive lock on the given object, else
397 : * this test can be fooled by concurrent additions.
398 : */
399 : bool
400 GIC 6881 : ConstraintNameIsUsed(ConstraintCategory conCat, Oid objId,
401 ECB : const char *conname)
402 : {
403 : bool found;
404 : Relation conDesc;
405 : SysScanDesc conscan;
406 : ScanKeyData skey[3];
407 :
408 CBC 6881 : conDesc = table_open(ConstraintRelationId, AccessShareLock);
409 :
410 GIC 6881 : ScanKeyInit(&skey[0],
411 : Anum_pg_constraint_conrelid,
412 : BTEqualStrategyNumber, F_OIDEQ,
413 ECB : ObjectIdGetDatum((conCat == CONSTRAINT_RELATION)
414 : ? objId : InvalidOid));
415 GIC 6881 : ScanKeyInit(&skey[1],
416 : Anum_pg_constraint_contypid,
417 : BTEqualStrategyNumber, F_OIDEQ,
418 ECB : ObjectIdGetDatum((conCat == CONSTRAINT_DOMAIN)
419 : ? objId : InvalidOid));
420 GIC 6881 : ScanKeyInit(&skey[2],
421 : Anum_pg_constraint_conname,
422 ECB : BTEqualStrategyNumber, F_NAMEEQ,
423 : CStringGetDatum(conname));
424 :
425 CBC 6881 : conscan = systable_beginscan(conDesc, ConstraintRelidTypidNameIndexId,
426 : true, NULL, 3, skey);
427 ECB :
428 : /* There can be at most one matching row */
429 GIC 6881 : found = (HeapTupleIsValid(systable_getnext(conscan)));
430 :
431 6881 : systable_endscan(conscan);
432 6881 : table_close(conDesc, AccessShareLock);
433 :
434 6881 : return found;
435 : }
436 :
437 : /*
438 ECB : * Does any constraint of the given name exist in the given namespace?
439 : *
440 : * This is used for code that wants to match ChooseConstraintName's rule
441 : * that we should avoid autogenerating duplicate constraint names within a
442 : * namespace.
443 : */
444 : bool
445 CBC 3840 : ConstraintNameExists(const char *conname, Oid namespaceid)
446 : {
447 ECB : bool found;
448 : Relation conDesc;
449 : SysScanDesc conscan;
450 : ScanKeyData skey[2];
451 :
452 CBC 3840 : conDesc = table_open(ConstraintRelationId, AccessShareLock);
453 :
454 GIC 3840 : ScanKeyInit(&skey[0],
455 : Anum_pg_constraint_conname,
456 : BTEqualStrategyNumber, F_NAMEEQ,
457 ECB : CStringGetDatum(conname));
458 :
459 GIC 3840 : ScanKeyInit(&skey[1],
460 ECB : Anum_pg_constraint_connamespace,
461 : BTEqualStrategyNumber, F_OIDEQ,
462 : ObjectIdGetDatum(namespaceid));
463 :
464 GIC 3840 : conscan = systable_beginscan(conDesc, ConstraintNameNspIndexId, true,
465 ECB : NULL, 2, skey);
466 :
467 GIC 3840 : found = (HeapTupleIsValid(systable_getnext(conscan)));
468 :
469 3840 : systable_endscan(conscan);
470 3840 : table_close(conDesc, AccessShareLock);
471 :
472 3840 : return found;
473 : }
474 :
475 : /*
476 : * Select a nonconflicting name for a new constraint.
477 : *
478 : * The objective here is to choose a name that is unique within the
479 : * specified namespace. Postgres does not require this, but the SQL
480 : * spec does, and some apps depend on it. Therefore we avoid choosing
481 : * default names that so conflict.
482 : *
483 : * name1, name2, and label are used the same way as for makeObjectName(),
484 : * except that the label can't be NULL; digits will be appended to the label
485 : * if needed to create a name that is unique within the specified namespace.
486 : *
487 : * 'others' can be a list of string names already chosen within the current
488 : * command (but not yet reflected into the catalogs); we will not choose
489 : * a duplicate of one of these either.
490 : *
491 : * Note: it is theoretically possible to get a collision anyway, if someone
492 ECB : * else chooses the same name concurrently. This is fairly unlikely to be
493 : * a problem in practice, especially if one is holding an exclusive lock on
494 : * the relation identified by name1.
495 : *
496 : * Returns a palloc'd string.
497 : */
498 : char *
499 GIC 4637 : ChooseConstraintName(const char *name1, const char *name2,
500 : const char *label, Oid namespaceid,
501 : List *others)
502 : {
503 4637 : int pass = 0;
504 4637 : char *conname = NULL;
505 ECB : char modlabel[NAMEDATALEN];
506 : Relation conDesc;
507 : SysScanDesc conscan;
508 : ScanKeyData skey[2];
509 : bool found;
510 : ListCell *l;
511 :
512 CBC 4637 : conDesc = table_open(ConstraintRelationId, AccessShareLock);
513 :
514 ECB : /* try the unmodified label first */
515 GIC 4637 : strlcpy(modlabel, label, sizeof(modlabel));
516 ECB :
517 : for (;;)
518 : {
519 GIC 5400 : conname = makeObjectName(name1, name2, modlabel);
520 ECB :
521 CBC 5400 : found = false;
522 :
523 GIC 6603 : foreach(l, others)
524 : {
525 CBC 1212 : if (strcmp((char *) lfirst(l), conname) == 0)
526 : {
527 9 : found = true;
528 GIC 9 : break;
529 : }
530 : }
531 :
532 CBC 5400 : if (!found)
533 : {
534 GIC 5391 : ScanKeyInit(&skey[0],
535 : Anum_pg_constraint_conname,
536 : BTEqualStrategyNumber, F_NAMEEQ,
537 ECB : CStringGetDatum(conname));
538 :
539 GIC 5391 : ScanKeyInit(&skey[1],
540 ECB : Anum_pg_constraint_connamespace,
541 : BTEqualStrategyNumber, F_OIDEQ,
542 : ObjectIdGetDatum(namespaceid));
543 :
544 GIC 5391 : conscan = systable_beginscan(conDesc, ConstraintNameNspIndexId, true,
545 ECB : NULL, 2, skey);
546 :
547 GIC 5391 : found = (HeapTupleIsValid(systable_getnext(conscan)));
548 :
549 CBC 5391 : systable_endscan(conscan);
550 ECB : }
551 :
552 GIC 5400 : if (!found)
553 CBC 4637 : break;
554 :
555 ECB : /* found a conflict, so try a new name component */
556 GIC 763 : pfree(conname);
557 763 : snprintf(modlabel, sizeof(modlabel), "%s%d", label, ++pass);
558 : }
559 :
560 4637 : table_close(conDesc, AccessShareLock);
561 :
562 4637 : return conname;
563 : }
564 :
565 : /*
566 : * Find and return the pg_constraint tuple that implements a validated
567 : * NOT NULL constraint for the given column of the given relation.
568 : *
569 : * XXX This would be easier if we had pg_attribute.notnullconstr with the OID
570 : * of the constraint that implements the NOT NULL constraint for that column.
571 : * I'm not sure it's worth the catalog bloat and de-normalization, however.
572 : */
573 : HeapTuple
574 GNC 221 : findNotNullConstraintAttnum(Relation rel, AttrNumber attnum)
575 : {
576 : Relation pg_constraint;
577 : HeapTuple conTup,
578 221 : retval = NULL;
579 : SysScanDesc scan;
580 : ScanKeyData key;
581 :
582 221 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
583 221 : ScanKeyInit(&key,
584 : Anum_pg_constraint_conrelid,
585 : BTEqualStrategyNumber, F_OIDEQ,
586 : ObjectIdGetDatum(RelationGetRelid(rel)));
587 221 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId,
588 : true, NULL, 1, &key);
589 :
590 394 : while (HeapTupleIsValid(conTup = systable_getnext(scan)))
591 : {
592 259 : Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(conTup);
593 : AttrNumber conkey;
594 :
595 : /*
596 : * We're looking for a NOTNULL constraint that's marked validated,
597 : * with the column we're looking for as the sole element in conkey.
598 : */
599 259 : if (con->contype != CONSTRAINT_NOTNULL)
600 76 : continue;
601 183 : if (!con->convalidated)
602 UNC 0 : continue;
603 :
604 GNC 183 : conkey = extractNotNullColumn(conTup);
605 183 : if (conkey != attnum)
606 97 : continue;
607 :
608 : /* Found it */
609 86 : retval = heap_copytuple(conTup);
610 86 : break;
611 : }
612 :
613 221 : systable_endscan(scan);
614 221 : table_close(pg_constraint, AccessShareLock);
615 :
616 221 : return retval;
617 : }
618 :
619 : /*
620 : * Find and return the pg_constraint tuple that implements a validated
621 : * NOT NULL constraint for the given column of the given relation.
622 : */
623 : HeapTuple
624 80 : findNotNullConstraint(Relation rel, const char *colname)
625 : {
626 80 : AttrNumber attnum = get_attnum(RelationGetRelid(rel), colname);
627 :
628 80 : return findNotNullConstraintAttnum(rel, attnum);
629 : }
630 :
631 : /*
632 : * Given a pg_constraint tuple for a NOT NULL constraint, return the column
633 : * number it is for.
634 : */
635 : AttrNumber
636 1397 : extractNotNullColumn(HeapTuple constrTup)
637 : {
638 : AttrNumber colnum;
639 : Datum adatum;
640 : ArrayType *arr;
641 :
642 : /* only tuples for CHECK constraints should be given */
643 1397 : Assert(((Form_pg_constraint) GETSTRUCT(constrTup))->contype == CONSTRAINT_NOTNULL);
644 :
645 1397 : adatum = SysCacheGetAttrNotNull(CONSTROID, constrTup,
646 : Anum_pg_constraint_conkey);
647 1397 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
648 1397 : if (ARR_NDIM(arr) != 1 ||
649 1397 : ARR_HASNULL(arr) ||
650 1397 : ARR_ELEMTYPE(arr) != INT2OID ||
651 1397 : ARR_DIMS(arr)[0] != 1)
652 UNC 0 : elog(ERROR, "conkey is not a 1-D smallint array");
653 :
654 GNC 1397 : memcpy(&colnum, ARR_DATA_PTR(arr), sizeof(AttrNumber));
655 :
656 1397 : if ((Pointer) arr != DatumGetPointer(adatum))
657 1397 : pfree(arr); /* free de-toasted copy, if any */
658 :
659 1397 : return colnum;
660 : }
661 :
662 : /*
663 : * Delete a single constraint record.
664 ECB : */
665 : void
666 GIC 8799 : RemoveConstraintById(Oid conId)
667 : {
668 ECB : Relation conDesc;
669 : HeapTuple tup;
670 : Form_pg_constraint con;
671 :
672 CBC 8799 : conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
673 ECB :
674 GIC 8799 : tup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(conId));
675 8799 : if (!HeapTupleIsValid(tup)) /* should not happen */
676 UIC 0 : elog(ERROR, "cache lookup failed for constraint %u", conId);
677 CBC 8799 : con = (Form_pg_constraint) GETSTRUCT(tup);
678 :
679 : /*
680 ECB : * Special processing depending on what the constraint is for.
681 : */
682 CBC 8799 : if (OidIsValid(con->conrelid))
683 : {
684 : Relation rel;
685 :
686 : /*
687 : * If the constraint is for a relation, open and exclusive-lock the
688 : * relation it's for.
689 ECB : */
690 CBC 8686 : rel = table_open(con->conrelid, AccessExclusiveLock);
691 ECB :
692 EUB : /*
693 : * We need to update the relchecks count if it is a check constraint
694 ECB : * being dropped. This update will force backends to rebuild relcache
695 : * entries when we commit.
696 : */
697 GIC 8685 : if (con->contype == CONSTRAINT_CHECK)
698 : {
699 ECB : Relation pgrel;
700 : HeapTuple relTup;
701 : Form_pg_class classForm;
702 :
703 CBC 758 : pgrel = table_open(RelationRelationId, RowExclusiveLock);
704 758 : relTup = SearchSysCacheCopy1(RELOID,
705 : ObjectIdGetDatum(con->conrelid));
706 758 : if (!HeapTupleIsValid(relTup))
707 UIC 0 : elog(ERROR, "cache lookup failed for relation %u",
708 : con->conrelid);
709 GIC 758 : classForm = (Form_pg_class) GETSTRUCT(relTup);
710 :
711 758 : if (classForm->relchecks == 0) /* should not happen */
712 UIC 0 : elog(ERROR, "relation \"%s\" has relchecks = 0",
713 : RelationGetRelationName(rel));
714 CBC 758 : classForm->relchecks--;
715 :
716 758 : CatalogTupleUpdate(pgrel, &relTup->t_self, relTup);
717 :
718 758 : heap_freetuple(relTup);
719 :
720 GIC 758 : table_close(pgrel, RowExclusiveLock);
721 : }
722 :
723 : /* Keep lock on constraint's rel until end of xact */
724 8685 : table_close(rel, NoLock);
725 : }
726 CBC 113 : else if (OidIsValid(con->contypid))
727 : {
728 : /*
729 : * XXX for now, do nothing special when dropping a domain constraint
730 : *
731 : * Probably there should be some form of locking on the domain type,
732 : * but we have no such concept at the moment.
733 ECB : */
734 : }
735 : else
736 UIC 0 : elog(ERROR, "constraint %u is not of a known type", conId);
737 ECB :
738 : /* Fry the constraint itself */
739 CBC 8798 : CatalogTupleDelete(conDesc, &tup->t_self);
740 ECB :
741 : /* Clean up */
742 GBC 8798 : ReleaseSysCache(tup);
743 GIC 8798 : table_close(conDesc, RowExclusiveLock);
744 CBC 8798 : }
745 :
746 ECB : /*
747 : * RenameConstraintById
748 : * Rename a constraint.
749 : *
750 : * Note: this isn't intended to be a user-exposed function; it doesn't check
751 : * permissions etc. Currently this is only invoked when renaming an index
752 : * that is associated with a constraint, but it's made a little more general
753 : * than that with the expectation of someday having ALTER TABLE RENAME
754 : * CONSTRAINT.
755 : */
756 : void
757 GIC 45 : RenameConstraintById(Oid conId, const char *newname)
758 : {
759 : Relation conDesc;
760 : HeapTuple tuple;
761 : Form_pg_constraint con;
762 ECB :
763 GIC 45 : conDesc = table_open(ConstraintRelationId, RowExclusiveLock);
764 ECB :
765 CBC 45 : tuple = SearchSysCacheCopy1(CONSTROID, ObjectIdGetDatum(conId));
766 GBC 45 : if (!HeapTupleIsValid(tuple))
767 LBC 0 : elog(ERROR, "cache lookup failed for constraint %u", conId);
768 GIC 45 : con = (Form_pg_constraint) GETSTRUCT(tuple);
769 :
770 : /*
771 : * For user-friendliness, check whether the name is already in use.
772 ECB : */
773 GIC 87 : if (OidIsValid(con->conrelid) &&
774 42 : ConstraintNameIsUsed(CONSTRAINT_RELATION,
775 : con->conrelid,
776 : newname))
777 UIC 0 : ereport(ERROR,
778 : (errcode(ERRCODE_DUPLICATE_OBJECT),
779 : errmsg("constraint \"%s\" for relation \"%s\" already exists",
780 ECB : newname, get_rel_name(con->conrelid))));
781 GIC 48 : if (OidIsValid(con->contypid) &&
782 3 : ConstraintNameIsUsed(CONSTRAINT_DOMAIN,
783 : con->contypid,
784 : newname))
785 UIC 0 : ereport(ERROR,
786 : (errcode(ERRCODE_DUPLICATE_OBJECT),
787 ECB : errmsg("constraint \"%s\" for domain %s already exists",
788 : newname, format_type_be(con->contypid))));
789 :
790 : /* OK, do the rename --- tuple is a copy, so OK to scribble on it */
791 GIC 45 : namestrcpy(&(con->conname), newname);
792 :
793 CBC 45 : CatalogTupleUpdate(conDesc, &tuple->t_self, tuple);
794 ECB :
795 GIC 45 : InvokeObjectPostAlterHook(ConstraintRelationId, conId, 0);
796 ECB :
797 GBC 45 : heap_freetuple(tuple);
798 GIC 45 : table_close(conDesc, RowExclusiveLock);
799 CBC 45 : }
800 :
801 ECB : /*
802 EUB : * AlterConstraintNamespaces
803 : * Find any constraints belonging to the specified object,
804 ECB : * and move them to the specified new namespace.
805 : *
806 : * isType indicates whether the owning object is a type or a relation.
807 : */
808 : void
809 GIC 44 : AlterConstraintNamespaces(Oid ownerId, Oid oldNspId,
810 ECB : Oid newNspId, bool isType, ObjectAddresses *objsMoved)
811 : {
812 : Relation conRel;
813 : ScanKeyData key[2];
814 : SysScanDesc scan;
815 : HeapTuple tup;
816 :
817 GIC 44 : conRel = table_open(ConstraintRelationId, RowExclusiveLock);
818 :
819 44 : ScanKeyInit(&key[0],
820 : Anum_pg_constraint_conrelid,
821 : BTEqualStrategyNumber, F_OIDEQ,
822 : ObjectIdGetDatum(isType ? InvalidOid : ownerId));
823 44 : ScanKeyInit(&key[1],
824 : Anum_pg_constraint_contypid,
825 : BTEqualStrategyNumber, F_OIDEQ,
826 EUB : ObjectIdGetDatum(isType ? ownerId : InvalidOid));
827 :
828 GIC 44 : scan = systable_beginscan(conRel, ConstraintRelidTypidNameIndexId, true,
829 ECB : NULL, 2, key);
830 :
831 GIC 96 : while (HeapTupleIsValid((tup = systable_getnext(scan))))
832 ECB : {
833 CBC 52 : Form_pg_constraint conform = (Form_pg_constraint) GETSTRUCT(tup);
834 ECB : ObjectAddress thisobj;
835 :
836 GIC 52 : ObjectAddressSet(thisobj, ConstraintRelationId, conform->oid);
837 :
838 52 : if (object_address_present(&thisobj, objsMoved))
839 UIC 0 : continue;
840 :
841 : /* Don't update if the object is already part of the namespace */
842 GIC 52 : if (conform->connamespace == oldNspId && oldNspId != newNspId)
843 : {
844 37 : tup = heap_copytuple(tup);
845 37 : conform = (Form_pg_constraint) GETSTRUCT(tup);
846 :
847 CBC 37 : conform->connamespace = newNspId;
848 :
849 GIC 37 : CatalogTupleUpdate(conRel, &tup->t_self, tup);
850 :
851 : /*
852 : * Note: currently, the constraint will not have its own
853 ECB : * dependency on the namespace, so we don't need to do
854 : * changeDependencyFor().
855 : */
856 : }
857 EUB :
858 CBC 52 : InvokeObjectPostAlterHook(ConstraintRelationId, thisobj.objectId, 0);
859 :
860 GIC 52 : add_exact_object_address(&thisobj, objsMoved);
861 : }
862 :
863 CBC 44 : systable_endscan(scan);
864 ECB :
865 GIC 44 : table_close(conRel, RowExclusiveLock);
866 44 : }
867 EUB :
868 : /*
869 : * ConstraintSetParentConstraint
870 : * Set a partition's constraint as child of its parent constraint,
871 ECB : * or remove the linkage if parentConstrId is InvalidOid.
872 : *
873 : * This updates the constraint's pg_constraint row to show it as inherited, and
874 : * adds PARTITION dependencies to prevent the constraint from being deleted
875 EUB : * on its own. Alternatively, reverse that.
876 : */
877 : void
878 GIC 222 : ConstraintSetParentConstraint(Oid childConstrId,
879 : Oid parentConstrId,
880 : Oid childTableId)
881 ECB : {
882 : Relation constrRel;
883 : Form_pg_constraint constrForm;
884 : HeapTuple tuple,
885 : newtup;
886 : ObjectAddress depender;
887 : ObjectAddress referenced;
888 :
889 CBC 222 : constrRel = table_open(ConstraintRelationId, RowExclusiveLock);
890 GIC 222 : tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(childConstrId));
891 222 : if (!HeapTupleIsValid(tuple))
892 UIC 0 : elog(ERROR, "cache lookup failed for constraint %u", childConstrId);
893 GIC 222 : newtup = heap_copytuple(tuple);
894 222 : constrForm = (Form_pg_constraint) GETSTRUCT(newtup);
895 222 : if (OidIsValid(parentConstrId))
896 : {
897 : /* don't allow setting parent for a constraint that already has one */
898 143 : Assert(constrForm->coninhcount == 0);
899 CBC 143 : if (constrForm->conparentid != InvalidOid)
900 UIC 0 : elog(ERROR, "constraint %u already has a parent constraint",
901 : childConstrId);
902 :
903 GIC 143 : constrForm->conislocal = false;
904 143 : constrForm->coninhcount++;
905 GNC 143 : if (constrForm->coninhcount < 0)
906 UNC 0 : ereport(ERROR,
907 : errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
908 : errmsg("too many inheritance parents"));
909 GIC 143 : constrForm->conparentid = parentConstrId;
910 :
911 CBC 143 : CatalogTupleUpdate(constrRel, &tuple->t_self, newtup);
912 :
913 143 : ObjectAddressSet(depender, ConstraintRelationId, childConstrId);
914 :
915 GIC 143 : ObjectAddressSet(referenced, ConstraintRelationId, parentConstrId);
916 143 : recordDependencyOn(&depender, &referenced, DEPENDENCY_PARTITION_PRI);
917 ECB :
918 GIC 143 : ObjectAddressSet(referenced, RelationRelationId, childTableId);
919 143 : recordDependencyOn(&depender, &referenced, DEPENDENCY_PARTITION_SEC);
920 : }
921 : else
922 ECB : {
923 GIC 79 : constrForm->coninhcount--;
924 79 : constrForm->conislocal = true;
925 CBC 79 : constrForm->conparentid = InvalidOid;
926 :
927 ECB : /* Make sure there's no further inheritance. */
928 GIC 79 : Assert(constrForm->coninhcount == 0);
929 :
930 CBC 79 : CatalogTupleUpdate(constrRel, &tuple->t_self, newtup);
931 :
932 79 : deleteDependencyRecordsForClass(ConstraintRelationId, childConstrId,
933 EUB : ConstraintRelationId,
934 : DEPENDENCY_PARTITION_PRI);
935 GIC 79 : deleteDependencyRecordsForClass(ConstraintRelationId, childConstrId,
936 ECB : RelationRelationId,
937 : DEPENDENCY_PARTITION_SEC);
938 : }
939 :
940 GIC 222 : ReleaseSysCache(tuple);
941 CBC 222 : table_close(constrRel, RowExclusiveLock);
942 GIC 222 : }
943 ECB :
944 :
945 : /*
946 : * get_relation_constraint_oid
947 : * Find a constraint on the specified relation with the specified name.
948 : * Returns constraint's OID.
949 : */
950 : Oid
951 GIC 160 : get_relation_constraint_oid(Oid relid, const char *conname, bool missing_ok)
952 ECB : {
953 : Relation pg_constraint;
954 : HeapTuple tuple;
955 : SysScanDesc scan;
956 : ScanKeyData skey[3];
957 CBC 160 : Oid conOid = InvalidOid;
958 :
959 160 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
960 ECB :
961 GIC 160 : ScanKeyInit(&skey[0],
962 : Anum_pg_constraint_conrelid,
963 : BTEqualStrategyNumber, F_OIDEQ,
964 : ObjectIdGetDatum(relid));
965 160 : ScanKeyInit(&skey[1],
966 : Anum_pg_constraint_contypid,
967 : BTEqualStrategyNumber, F_OIDEQ,
968 : ObjectIdGetDatum(InvalidOid));
969 160 : ScanKeyInit(&skey[2],
970 : Anum_pg_constraint_conname,
971 : BTEqualStrategyNumber, F_NAMEEQ,
972 ECB : CStringGetDatum(conname));
973 :
974 GIC 160 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,
975 : NULL, 3, skey);
976 :
977 : /* There can be at most one matching row */
978 160 : if (HeapTupleIsValid(tuple = systable_getnext(scan)))
979 154 : conOid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;
980 :
981 160 : systable_endscan(scan);
982 :
983 ECB : /* If no such constraint exists, complain */
984 CBC 160 : if (!OidIsValid(conOid) && !missing_ok)
985 6 : ereport(ERROR,
986 EUB : (errcode(ERRCODE_UNDEFINED_OBJECT),
987 ECB : errmsg("constraint \"%s\" for table \"%s\" does not exist",
988 : conname, get_rel_name(relid))));
989 :
990 GIC 154 : table_close(pg_constraint, AccessShareLock);
991 :
992 CBC 154 : return conOid;
993 ECB : }
994 EUB :
995 : /*
996 : * get_relation_constraint_attnos
997 ECB : * Find a constraint on the specified relation with the specified name
998 : * and return the constrained columns.
999 : *
1000 EUB : * Returns a Bitmapset of the column attnos of the constrained columns, with
1001 : * attnos being offset by FirstLowInvalidHeapAttributeNumber so that system
1002 : * columns can be represented.
1003 ECB : *
1004 : * *constraintOid is set to the OID of the constraint, or InvalidOid on
1005 : * failure.
1006 : */
1007 : Bitmapset *
1008 GIC 24 : get_relation_constraint_attnos(Oid relid, const char *conname,
1009 ECB : bool missing_ok, Oid *constraintOid)
1010 : {
1011 GIC 24 : Bitmapset *conattnos = NULL;
1012 ECB : Relation pg_constraint;
1013 : HeapTuple tuple;
1014 : SysScanDesc scan;
1015 : ScanKeyData skey[3];
1016 :
1017 : /* Set *constraintOid, to avoid complaints about uninitialized vars */
1018 CBC 24 : *constraintOid = InvalidOid;
1019 ECB :
1020 GIC 24 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
1021 :
1022 CBC 24 : ScanKeyInit(&skey[0],
1023 : Anum_pg_constraint_conrelid,
1024 ECB : BTEqualStrategyNumber, F_OIDEQ,
1025 : ObjectIdGetDatum(relid));
1026 CBC 24 : ScanKeyInit(&skey[1],
1027 : Anum_pg_constraint_contypid,
1028 : BTEqualStrategyNumber, F_OIDEQ,
1029 ECB : ObjectIdGetDatum(InvalidOid));
1030 GIC 24 : ScanKeyInit(&skey[2],
1031 : Anum_pg_constraint_conname,
1032 : BTEqualStrategyNumber, F_NAMEEQ,
1033 : CStringGetDatum(conname));
1034 ECB :
1035 CBC 24 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,
1036 ECB : NULL, 3, skey);
1037 :
1038 : /* There can be at most one matching row */
1039 GIC 24 : if (HeapTupleIsValid(tuple = systable_getnext(scan)))
1040 : {
1041 : Datum adatum;
1042 : bool isNull;
1043 :
1044 24 : *constraintOid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;
1045 ECB :
1046 : /* Extract the conkey array, ie, attnums of constrained columns */
1047 GIC 24 : adatum = heap_getattr(tuple, Anum_pg_constraint_conkey,
1048 : RelationGetDescr(pg_constraint), &isNull);
1049 24 : if (!isNull)
1050 : {
1051 ECB : ArrayType *arr;
1052 : int numcols;
1053 : int16 *attnums;
1054 : int i;
1055 :
1056 GIC 24 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1057 24 : numcols = ARR_DIMS(arr)[0];
1058 24 : if (ARR_NDIM(arr) != 1 ||
1059 CBC 24 : numcols < 0 ||
1060 GIC 24 : ARR_HASNULL(arr) ||
1061 24 : ARR_ELEMTYPE(arr) != INT2OID)
1062 UIC 0 : elog(ERROR, "conkey is not a 1-D smallint array");
1063 CBC 24 : attnums = (int16 *) ARR_DATA_PTR(arr);
1064 :
1065 : /* Construct the result value */
1066 GIC 54 : for (i = 0; i < numcols; i++)
1067 : {
1068 CBC 30 : conattnos = bms_add_member(conattnos,
1069 GIC 30 : attnums[i] - FirstLowInvalidHeapAttributeNumber);
1070 : }
1071 : }
1072 ECB : }
1073 :
1074 GIC 24 : systable_endscan(scan);
1075 ECB :
1076 : /* If no such constraint exists, complain */
1077 GIC 24 : if (!OidIsValid(*constraintOid) && !missing_ok)
1078 LBC 0 : ereport(ERROR,
1079 ECB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1080 : errmsg("constraint \"%s\" for table \"%s\" does not exist",
1081 : conname, get_rel_name(relid))));
1082 :
1083 GIC 24 : table_close(pg_constraint, AccessShareLock);
1084 ECB :
1085 GIC 24 : return conattnos;
1086 ECB : }
1087 :
1088 : /*
1089 : * Return the OID of the constraint enforced by the given index in the
1090 : * given relation; or InvalidOid if no such index is catalogued.
1091 : *
1092 : * Much like get_constraint_index, this function is concerned only with the
1093 : * one constraint that "owns" the given index. Therefore, constraints of
1094 : * types other than unique, primary-key, and exclusion are ignored.
1095 : */
1096 : Oid
1097 GIC 517 : get_relation_idx_constraint_oid(Oid relationId, Oid indexId)
1098 : {
1099 : Relation pg_constraint;
1100 : SysScanDesc scan;
1101 : ScanKeyData key;
1102 ECB : HeapTuple tuple;
1103 GIC 517 : Oid constraintId = InvalidOid;
1104 :
1105 CBC 517 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
1106 :
1107 GIC 517 : ScanKeyInit(&key,
1108 : Anum_pg_constraint_conrelid,
1109 : BTEqualStrategyNumber,
1110 : F_OIDEQ,
1111 : ObjectIdGetDatum(relationId));
1112 CBC 517 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId,
1113 : true, NULL, 1, &key);
1114 619 : while ((tuple = systable_getnext(scan)) != NULL)
1115 : {
1116 ECB : Form_pg_constraint constrForm;
1117 :
1118 GIC 405 : constrForm = (Form_pg_constraint) GETSTRUCT(tuple);
1119 :
1120 ECB : /* See above */
1121 GIC 405 : if (constrForm->contype != CONSTRAINT_PRIMARY &&
1122 141 : constrForm->contype != CONSTRAINT_UNIQUE &&
1123 99 : constrForm->contype != CONSTRAINT_EXCLUSION)
1124 CBC 99 : continue;
1125 :
1126 GIC 306 : if (constrForm->conindid == indexId)
1127 : {
1128 303 : constraintId = constrForm->oid;
1129 CBC 303 : break;
1130 : }
1131 : }
1132 GIC 517 : systable_endscan(scan);
1133 ECB :
1134 GIC 517 : table_close(pg_constraint, AccessShareLock);
1135 517 : return constraintId;
1136 : }
1137 :
1138 ECB : /*
1139 : * get_domain_constraint_oid
1140 : * Find a constraint on the specified domain with the specified name.
1141 : * Returns constraint's OID.
1142 : */
1143 : Oid
1144 GIC 29 : get_domain_constraint_oid(Oid typid, const char *conname, bool missing_ok)
1145 : {
1146 : Relation pg_constraint;
1147 : HeapTuple tuple;
1148 : SysScanDesc scan;
1149 : ScanKeyData skey[3];
1150 CBC 29 : Oid conOid = InvalidOid;
1151 ECB :
1152 CBC 29 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
1153 ECB :
1154 CBC 29 : ScanKeyInit(&skey[0],
1155 ECB : Anum_pg_constraint_conrelid,
1156 EUB : BTEqualStrategyNumber, F_OIDEQ,
1157 ECB : ObjectIdGetDatum(InvalidOid));
1158 GIC 29 : ScanKeyInit(&skey[1],
1159 : Anum_pg_constraint_contypid,
1160 ECB : BTEqualStrategyNumber, F_OIDEQ,
1161 : ObjectIdGetDatum(typid));
1162 CBC 29 : ScanKeyInit(&skey[2],
1163 ECB : Anum_pg_constraint_conname,
1164 : BTEqualStrategyNumber, F_NAMEEQ,
1165 : CStringGetDatum(conname));
1166 :
1167 GIC 29 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,
1168 ECB : NULL, 3, skey);
1169 :
1170 : /* There can be at most one matching row */
1171 CBC 29 : if (HeapTupleIsValid(tuple = systable_getnext(scan)))
1172 GBC 26 : conOid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;
1173 :
1174 GIC 29 : systable_endscan(scan);
1175 :
1176 : /* If no such constraint exists, complain */
1177 CBC 29 : if (!OidIsValid(conOid) && !missing_ok)
1178 GIC 3 : ereport(ERROR,
1179 ECB : (errcode(ERRCODE_UNDEFINED_OBJECT),
1180 : errmsg("constraint \"%s\" for domain %s does not exist",
1181 : conname, format_type_be(typid))));
1182 :
1183 GIC 26 : table_close(pg_constraint, AccessShareLock);
1184 :
1185 26 : return conOid;
1186 : }
1187 :
1188 : /*
1189 : * get_primary_key_attnos
1190 : * Identify the columns in a relation's primary key, if any.
1191 ECB : *
1192 : * Returns a Bitmapset of the column attnos of the primary key's columns,
1193 : * with attnos being offset by FirstLowInvalidHeapAttributeNumber so that
1194 : * system columns can be represented.
1195 : *
1196 : * If there is no primary key, return NULL. We also return NULL if the pkey
1197 : * constraint is deferrable and deferrableOk is false.
1198 : *
1199 : * *constraintOid is set to the OID of the pkey constraint, or InvalidOid
1200 : * on failure.
1201 : */
1202 : Bitmapset *
1203 GIC 442 : get_primary_key_attnos(Oid relid, bool deferrableOk, Oid *constraintOid)
1204 : {
1205 442 : Bitmapset *pkattnos = NULL;
1206 ECB : Relation pg_constraint;
1207 : HeapTuple tuple;
1208 : SysScanDesc scan;
1209 : ScanKeyData skey[1];
1210 :
1211 : /* Set *constraintOid, to avoid complaints about uninitialized vars */
1212 CBC 442 : *constraintOid = InvalidOid;
1213 :
1214 : /* Scan pg_constraint for constraints of the target rel */
1215 442 : pg_constraint = table_open(ConstraintRelationId, AccessShareLock);
1216 ECB :
1217 CBC 442 : ScanKeyInit(&skey[0],
1218 ECB : Anum_pg_constraint_conrelid,
1219 : BTEqualStrategyNumber, F_OIDEQ,
1220 : ObjectIdGetDatum(relid));
1221 :
1222 CBC 442 : scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,
1223 ECB : NULL, 1, skey);
1224 :
1225 GIC 523 : while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1226 ECB : {
1227 GIC 240 : Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
1228 ECB : Datum adatum;
1229 : bool isNull;
1230 : ArrayType *arr;
1231 : int16 *attnums;
1232 : int numkeys;
1233 : int i;
1234 :
1235 : /* Skip constraints that are not PRIMARY KEYs */
1236 GIC 240 : if (con->contype != CONSTRAINT_PRIMARY)
1237 81 : continue;
1238 ECB :
1239 : /*
1240 : * If the primary key is deferrable, but we've been instructed to
1241 : * ignore deferrable constraints, then we might as well give up
1242 : * searching, since there can only be a single primary key on a table.
1243 : */
1244 CBC 159 : if (con->condeferrable && !deferrableOk)
1245 GIC 159 : break;
1246 ECB :
1247 : /* Extract the conkey array, ie, attnums of PK's columns */
1248 CBC 156 : adatum = heap_getattr(tuple, Anum_pg_constraint_conkey,
1249 : RelationGetDescr(pg_constraint), &isNull);
1250 GIC 156 : if (isNull)
1251 UIC 0 : elog(ERROR, "null conkey for constraint %u",
1252 ECB : ((Form_pg_constraint) GETSTRUCT(tuple))->oid);
1253 GIC 156 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1254 156 : numkeys = ARR_DIMS(arr)[0];
1255 156 : if (ARR_NDIM(arr) != 1 ||
1256 CBC 156 : numkeys < 0 ||
1257 GIC 156 : ARR_HASNULL(arr) ||
1258 156 : ARR_ELEMTYPE(arr) != INT2OID)
1259 UIC 0 : elog(ERROR, "conkey is not a 1-D smallint array");
1260 GIC 156 : attnums = (int16 *) ARR_DATA_PTR(arr);
1261 ECB :
1262 : /* Construct the result value */
1263 GIC 348 : for (i = 0; i < numkeys; i++)
1264 : {
1265 CBC 192 : pkattnos = bms_add_member(pkattnos,
1266 192 : attnums[i] - FirstLowInvalidHeapAttributeNumber);
1267 : }
1268 156 : *constraintOid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;
1269 :
1270 : /* No need to search further */
1271 156 : break;
1272 ECB : }
1273 :
1274 GIC 442 : systable_endscan(scan);
1275 :
1276 442 : table_close(pg_constraint, AccessShareLock);
1277 ECB :
1278 GIC 442 : return pkattnos;
1279 ECB : }
1280 :
1281 : /*
1282 : * Extract data from the pg_constraint tuple of a foreign-key constraint.
1283 : *
1284 : * All arguments save the first are output arguments. All output arguments
1285 : * other than numfks, conkey and confkey can be passed as NULL if caller
1286 : * doesn't need them.
1287 : */
1288 : void
1289 GIC 3481 : DeconstructFkConstraintRow(HeapTuple tuple, int *numfks,
1290 : AttrNumber *conkey, AttrNumber *confkey,
1291 : Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs,
1292 : int *num_fk_del_set_cols, AttrNumber *fk_del_set_cols)
1293 : {
1294 : Datum adatum;
1295 : bool isNull;
1296 ECB : ArrayType *arr;
1297 : int numkeys;
1298 :
1299 : /*
1300 : * We expect the arrays to be 1-D arrays of the right types; verify that.
1301 : * We don't need to use deconstruct_array() since the array data is just
1302 : * going to look like a C array of values.
1303 : */
1304 GNC 3481 : adatum = SysCacheGetAttrNotNull(CONSTROID, tuple,
1305 : Anum_pg_constraint_conkey);
1306 CBC 3481 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1307 GIC 3481 : if (ARR_NDIM(arr) != 1 ||
1308 3481 : ARR_HASNULL(arr) ||
1309 3481 : ARR_ELEMTYPE(arr) != INT2OID)
1310 UIC 0 : elog(ERROR, "conkey is not a 1-D smallint array");
1311 CBC 3481 : numkeys = ARR_DIMS(arr)[0];
1312 GIC 3481 : if (numkeys <= 0 || numkeys > INDEX_MAX_KEYS)
1313 UIC 0 : elog(ERROR, "foreign key constraint cannot have %d columns", numkeys);
1314 CBC 3481 : memcpy(conkey, ARR_DATA_PTR(arr), numkeys * sizeof(int16));
1315 GIC 3481 : if ((Pointer) arr != DatumGetPointer(adatum))
1316 CBC 3481 : pfree(arr); /* free de-toasted copy, if any */
1317 :
1318 GNC 3481 : adatum = SysCacheGetAttrNotNull(CONSTROID, tuple,
1319 : Anum_pg_constraint_confkey);
1320 GIC 3481 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1321 3481 : if (ARR_NDIM(arr) != 1 ||
1322 3481 : ARR_DIMS(arr)[0] != numkeys ||
1323 CBC 3481 : ARR_HASNULL(arr) ||
1324 3481 : ARR_ELEMTYPE(arr) != INT2OID)
1325 UIC 0 : elog(ERROR, "confkey is not a 1-D smallint array");
1326 GIC 3481 : memcpy(confkey, ARR_DATA_PTR(arr), numkeys * sizeof(int16));
1327 3481 : if ((Pointer) arr != DatumGetPointer(adatum))
1328 3481 : pfree(arr); /* free de-toasted copy, if any */
1329 :
1330 3481 : if (pf_eq_oprs)
1331 ECB : {
1332 GNC 3481 : adatum = SysCacheGetAttrNotNull(CONSTROID, tuple,
1333 : Anum_pg_constraint_conpfeqop);
1334 GIC 3481 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1335 ECB : /* see TryReuseForeignKey if you change the test below */
1336 GBC 3481 : if (ARR_NDIM(arr) != 1 ||
1337 GIC 3481 : ARR_DIMS(arr)[0] != numkeys ||
1338 CBC 3481 : ARR_HASNULL(arr) ||
1339 3481 : ARR_ELEMTYPE(arr) != OIDOID)
1340 LBC 0 : elog(ERROR, "conpfeqop is not a 1-D Oid array");
1341 CBC 3481 : memcpy(pf_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));
1342 3481 : if ((Pointer) arr != DatumGetPointer(adatum))
1343 3481 : pfree(arr); /* free de-toasted copy, if any */
1344 EUB : }
1345 ECB :
1346 GIC 3481 : if (pp_eq_oprs)
1347 : {
1348 GNC 2173 : adatum = SysCacheGetAttrNotNull(CONSTROID, tuple,
1349 : Anum_pg_constraint_conppeqop);
1350 GIC 2173 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1351 CBC 2173 : if (ARR_NDIM(arr) != 1 ||
1352 GIC 2173 : ARR_DIMS(arr)[0] != numkeys ||
1353 2173 : ARR_HASNULL(arr) ||
1354 CBC 2173 : ARR_ELEMTYPE(arr) != OIDOID)
1355 UIC 0 : elog(ERROR, "conppeqop is not a 1-D Oid array");
1356 GIC 2173 : memcpy(pp_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));
1357 CBC 2173 : if ((Pointer) arr != DatumGetPointer(adatum))
1358 GIC 2173 : pfree(arr); /* free de-toasted copy, if any */
1359 ECB : }
1360 :
1361 CBC 3481 : if (ff_eq_oprs)
1362 : {
1363 GNC 2173 : adatum = SysCacheGetAttrNotNull(CONSTROID, tuple,
1364 : Anum_pg_constraint_conffeqop);
1365 GIC 2173 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1366 2173 : if (ARR_NDIM(arr) != 1 ||
1367 2173 : ARR_DIMS(arr)[0] != numkeys ||
1368 2173 : ARR_HASNULL(arr) ||
1369 2173 : ARR_ELEMTYPE(arr) != OIDOID)
1370 LBC 0 : elog(ERROR, "conffeqop is not a 1-D Oid array");
1371 GIC 2173 : memcpy(ff_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));
1372 2173 : if ((Pointer) arr != DatumGetPointer(adatum))
1373 2173 : pfree(arr); /* free de-toasted copy, if any */
1374 : }
1375 :
1376 3481 : if (fk_del_set_cols)
1377 : {
1378 2173 : adatum = SysCacheGetAttr(CONSTROID, tuple,
1379 : Anum_pg_constraint_confdelsetcols, &isNull);
1380 2173 : if (isNull)
1381 : {
1382 2134 : *num_fk_del_set_cols = 0;
1383 : }
1384 : else
1385 ECB : {
1386 : int num_delete_cols;
1387 :
1388 CBC 39 : arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */
1389 39 : if (ARR_NDIM(arr) != 1 ||
1390 39 : ARR_HASNULL(arr) ||
1391 GBC 39 : ARR_ELEMTYPE(arr) != INT2OID)
1392 LBC 0 : elog(ERROR, "confdelsetcols is not a 1-D smallint array");
1393 CBC 39 : num_delete_cols = ARR_DIMS(arr)[0];
1394 GBC 39 : memcpy(fk_del_set_cols, ARR_DATA_PTR(arr), num_delete_cols * sizeof(int16));
1395 CBC 39 : if ((Pointer) arr != DatumGetPointer(adatum))
1396 39 : pfree(arr); /* free de-toasted copy, if any */
1397 ECB :
1398 GIC 39 : *num_fk_del_set_cols = num_delete_cols;
1399 ECB : }
1400 : }
1401 :
1402 CBC 3481 : *numfks = numkeys;
1403 3481 : }
1404 ECB :
1405 : /*
1406 EUB : * Determine whether a relation can be proven functionally dependent on
1407 ECB : * a set of grouping columns. If so, return true and add the pg_constraint
1408 : * OIDs of the constraints needed for the proof to the *constraintDeps list.
1409 : *
1410 : * grouping_columns is a list of grouping expressions, in which columns of
1411 : * the rel of interest are Vars with the indicated varno/varlevelsup.
1412 : *
1413 : * Currently we only check to see if the rel has a primary key that is a
1414 : * subset of the grouping_columns. We could also use plain unique constraints
1415 : * if all their columns are known not null, but there's a problem: we need
1416 : * to be able to represent the not-null-ness as part of the constraints added
1417 : * to *constraintDeps. FIXME whenever not-null constraints get represented
1418 : * in pg_constraint.
1419 : */
1420 : bool
1421 GBC 98 : check_functional_grouping(Oid relid,
1422 ECB : Index varno, Index varlevelsup,
1423 : List *grouping_columns,
1424 : List **constraintDeps)
1425 : {
1426 : Bitmapset *pkattnos;
1427 : Bitmapset *groupbyattnos;
1428 : Oid constraintOid;
1429 : ListCell *gl;
1430 :
1431 : /* If the rel has no PK, then we can't prove functional dependency */
1432 CBC 98 : pkattnos = get_primary_key_attnos(relid, false, &constraintOid);
1433 98 : if (pkattnos == NULL)
1434 21 : return false;
1435 ECB :
1436 EUB : /* Identify all the rel's columns that appear in grouping_columns */
1437 CBC 77 : groupbyattnos = NULL;
1438 175 : foreach(gl, grouping_columns)
1439 ECB : {
1440 GIC 98 : Var *gvar = (Var *) lfirst(gl);
1441 :
1442 CBC 98 : if (IsA(gvar, Var) &&
1443 GIC 98 : gvar->varno == varno &&
1444 CBC 77 : gvar->varlevelsup == varlevelsup)
1445 GIC 77 : groupbyattnos = bms_add_member(groupbyattnos,
1446 CBC 77 : gvar->varattno - FirstLowInvalidHeapAttributeNumber);
1447 ECB : }
1448 :
1449 CBC 77 : if (bms_is_subset(pkattnos, groupbyattnos))
1450 ECB : {
1451 EUB : /* The PK is a subset of grouping_columns, so we win */
1452 CBC 56 : *constraintDeps = lappend_oid(*constraintDeps, constraintOid);
1453 56 : return true;
1454 ECB : }
1455 :
1456 GIC 21 : return false;
1457 ECB : }
|
