1 /*-------------------------------------------------------------------------
2 *
3 * heap.c
4 * code to create and destroy POSTGRES heap relations
5 *
6 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
7 * Portions Copyright (c) 1994, Regents of the University of California
8 *
9 *
10 * IDENTIFICATION
11 * src/backend/catalog/heap.c
12 *
13 *
14 * INTERFACE ROUTINES
15 * heap_create() - Create an uncataloged heap relation
16 * heap_create_with_catalog() - Create a cataloged relation
17 * heap_drop_with_catalog() - Removes named relation from catalogs
18 *
19 * NOTES
20 * this code taken from access/heap/create.c, which contains
21 * the old heap_create_with_catalog, amcreate, and amdestroy.
22 * those routines will soon call these routines using the function
23 * manager,
24 * just like the poorly named "NewXXX" routines do. The
25 * "New" routines are all going to die soon, once and for all!
26 * -cim 1/13/91
27 *
28 *-------------------------------------------------------------------------
29 */
30 #include "postgres.h"
31
32 #include "access/htup_details.h"
33 #include "access/multixact.h"
34 #include "access/sysattr.h"
35 #include "access/transam.h"
36 #include "access/xact.h"
37 #include "access/xlog.h"
38 #include "catalog/binary_upgrade.h"
39 #include "catalog/catalog.h"
40 #include "catalog/dependency.h"
41 #include "catalog/heap.h"
42 #include "catalog/index.h"
43 #include "catalog/objectaccess.h"
44 #include "catalog/partition.h"
45 #include "catalog/pg_attrdef.h"
46 #include "catalog/pg_collation.h"
47 #include "catalog/pg_constraint.h"
48 #include "catalog/pg_foreign_table.h"
49 #include "catalog/pg_inherits.h"
50 #include "catalog/pg_namespace.h"
51 #include "catalog/pg_opclass.h"
52 #include "catalog/pg_partitioned_table.h"
53 #include "catalog/pg_statistic.h"
54 #include "catalog/pg_subscription_rel.h"
55 #include "catalog/pg_tablespace.h"
56 #include "catalog/pg_type.h"
57 #include "catalog/storage.h"
58 #include "catalog/storage_xlog.h"
59 #include "commands/tablecmds.h"
60 #include "commands/typecmds.h"
61 #include "executor/executor.h"
62 #include "miscadmin.h"
63 #include "nodes/nodeFuncs.h"
64 #include "optimizer/clauses.h"
65 #include "optimizer/var.h"
66 #include "optimizer/planner.h"
67 #include "parser/parse_coerce.h"
68 #include "parser/parse_collate.h"
69 #include "parser/parse_expr.h"
70 #include "parser/parse_relation.h"
71 #include "storage/lmgr.h"
72 #include "storage/predicate.h"
73 #include "storage/smgr.h"
74 #include "utils/acl.h"
75 #include "utils/builtins.h"
76 #include "utils/datum.h"
77 #include "utils/fmgroids.h"
78 #include "utils/inval.h"
79 #include "utils/lsyscache.h"
80 #include "utils/partcache.h"
81 #include "utils/rel.h"
82 #include "utils/ruleutils.h"
83 #include "utils/snapmgr.h"
84 #include "utils/syscache.h"
85 #include "utils/tqual.h"
86
87
88 /* Potentially set by pg_upgrade_support functions */
89 Oid binary_upgrade_next_heap_pg_class_oid = InvalidOid;
90 Oid binary_upgrade_next_toast_pg_class_oid = InvalidOid;
91
92 static void AddNewRelationTuple(Relation pg_class_desc,
93 Relation new_rel_desc,
94 Oid new_rel_oid,
95 Oid new_type_oid,
96 Oid reloftype,
97 Oid relowner,
98 char relkind,
99 Datum relacl,
100 Datum reloptions);
101 static ObjectAddress AddNewRelationType(const char *typeName,
102 Oid typeNamespace,
103 Oid new_rel_oid,
104 char new_rel_kind,
105 Oid ownerid,
106 Oid new_row_type,
107 Oid new_array_type);
108 static void RelationRemoveInheritance(Oid relid);
109 static Oid StoreRelCheck(Relation rel, const char *ccname, Node *expr,
110 bool is_validated, bool is_local, int inhcount,
111 bool is_no_inherit, bool is_internal);
112 static void StoreConstraints(Relation rel, List *cooked_constraints,
113 bool is_internal);
114 static bool MergeWithExistingConstraint(Relation rel, const char *ccname, Node *expr,
115 bool allow_merge, bool is_local,
116 bool is_initially_valid,
117 bool is_no_inherit);
118 static void SetRelationNumChecks(Relation rel, int numchecks);
119 static Node *cookConstraint(ParseState *pstate,
120 Node *raw_constraint,
121 char *relname);
122 static List *insert_ordered_unique_oid(List *list, Oid datum);
123
124
125 /* ----------------------------------------------------------------
126 * XXX UGLY HARD CODED BADNESS FOLLOWS XXX
127 *
128 * these should all be moved to someplace in the lib/catalog
129 * module, if not obliterated first.
130 * ----------------------------------------------------------------
131 */
132
133
134 /*
135 * Note:
136 * Should the system special case these attributes in the future?
137 * Advantage: consume much less space in the ATTRIBUTE relation.
138 * Disadvantage: special cases will be all over the place.
139 */
140
141 /*
142 * The initializers below do not include trailing variable length fields,
143 * but that's OK - we're never going to reference anything beyond the
144 * fixed-size portion of the structure anyway.
145 */
146
147 static FormData_pg_attribute a1 = {
148 0, {"ctid"}, TIDOID, 0, sizeof(ItemPointerData),
149 SelfItemPointerAttributeNumber, 0, -1, -1,
150 false, 'p', 's', true, false, false, '\0', false, true, 0
151 };
152
153 static FormData_pg_attribute a2 = {
154 0, {"oid"}, OIDOID, 0, sizeof(Oid),
155 ObjectIdAttributeNumber, 0, -1, -1,
156 true, 'p', 'i', true, false, false, '\0', false, true, 0
157 };
158
159 static FormData_pg_attribute a3 = {
160 0, {"xmin"}, XIDOID, 0, sizeof(TransactionId),
161 MinTransactionIdAttributeNumber, 0, -1, -1,
162 true, 'p', 'i', true, false, false, '\0', false, true, 0
163 };
164
165 static FormData_pg_attribute a4 = {
166 0, {"cmin"}, CIDOID, 0, sizeof(CommandId),
167 MinCommandIdAttributeNumber, 0, -1, -1,
168 true, 'p', 'i', true, false, false, '\0', false, true, 0
169 };
170
171 static FormData_pg_attribute a5 = {
172 0, {"xmax"}, XIDOID, 0, sizeof(TransactionId),
173 MaxTransactionIdAttributeNumber, 0, -1, -1,
174 true, 'p', 'i', true, false, false, '\0', false, true, 0
175 };
176
177 static FormData_pg_attribute a6 = {
178 0, {"cmax"}, CIDOID, 0, sizeof(CommandId),
179 MaxCommandIdAttributeNumber, 0, -1, -1,
180 true, 'p', 'i', true, false, false, '\0', false, true, 0
181 };
182
183 /*
184 * We decided to call this attribute "tableoid" rather than say
185 * "classoid" on the basis that in the future there may be more than one
186 * table of a particular class/type. In any case table is still the word
187 * used in SQL.
188 */
189 static FormData_pg_attribute a7 = {
190 0, {"tableoid"}, OIDOID, 0, sizeof(Oid),
191 TableOidAttributeNumber, 0, -1, -1,
192 true, 'p', 'i', true, false, false, '\0', false, true, 0
193 };
194
195 static const Form_pg_attribute SysAtt[] = {&a1, &a2, &a3, &a4, &a5, &a6, &a7};
196
197 /*
198 * This function returns a Form_pg_attribute pointer for a system attribute.
199 * Note that we elog if the presented attno is invalid, which would only
200 * happen if there's a problem upstream.
201 */
202 Form_pg_attribute
SystemAttributeDefinition(AttrNumber attno,bool relhasoids)203 SystemAttributeDefinition(AttrNumber attno, bool relhasoids)
204 {
205 if (attno >= 0 || attno < -(int) lengthof(SysAtt))
206 elog(ERROR, "invalid system attribute number %d", attno);
207 if (attno == ObjectIdAttributeNumber && !relhasoids)
208 elog(ERROR, "invalid system attribute number %d", attno);
209 return SysAtt[-attno - 1];
210 }
211
212 /*
213 * If the given name is a system attribute name, return a Form_pg_attribute
214 * pointer for a prototype definition. If not, return NULL.
215 */
216 Form_pg_attribute
SystemAttributeByName(const char * attname,bool relhasoids)217 SystemAttributeByName(const char *attname, bool relhasoids)
218 {
219 int j;
220
221 for (j = 0; j < (int) lengthof(SysAtt); j++)
222 {
223 Form_pg_attribute att = SysAtt[j];
224
225 if (relhasoids || att->attnum != ObjectIdAttributeNumber)
226 {
227 if (strcmp(NameStr(att->attname), attname) == 0)
228 return att;
229 }
230 }
231
232 return NULL;
233 }
234
235
236 /* ----------------------------------------------------------------
237 * XXX END OF UGLY HARD CODED BADNESS XXX
238 * ---------------------------------------------------------------- */
239
240
241 /* ----------------------------------------------------------------
242 * heap_create - Create an uncataloged heap relation
243 *
244 * Note API change: the caller must now always provide the OID
245 * to use for the relation. The relfilenode may (and, normally,
246 * should) be left unspecified.
247 *
248 * rel->rd_rel is initialized by RelationBuildLocalRelation,
249 * and is mostly zeroes at return.
250 * ----------------------------------------------------------------
251 */
252 Relation
heap_create(const char * relname,Oid relnamespace,Oid reltablespace,Oid relid,Oid relfilenode,TupleDesc tupDesc,char relkind,char relpersistence,bool shared_relation,bool mapped_relation,bool allow_system_table_mods)253 heap_create(const char *relname,
254 Oid relnamespace,
255 Oid reltablespace,
256 Oid relid,
257 Oid relfilenode,
258 TupleDesc tupDesc,
259 char relkind,
260 char relpersistence,
261 bool shared_relation,
262 bool mapped_relation,
263 bool allow_system_table_mods)
264 {
265 bool create_storage;
266 Relation rel;
267
268 /* The caller must have provided an OID for the relation. */
269 Assert(OidIsValid(relid));
270
271 /*
272 * Don't allow creating relations in pg_catalog directly, even though it
273 * is allowed to move user defined relations there. Semantics with search
274 * paths including pg_catalog are too confusing for now.
275 *
276 * But allow creating indexes on relations in pg_catalog even if
277 * allow_system_table_mods = off, upper layers already guarantee it's on a
278 * user defined relation, not a system one.
279 */
280 if (!allow_system_table_mods &&
281 ((IsSystemNamespace(relnamespace) && relkind != RELKIND_INDEX) ||
282 IsToastNamespace(relnamespace)) &&
283 IsNormalProcessingMode())
284 ereport(ERROR,
285 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
286 errmsg("permission denied to create \"%s.%s\"",
287 get_namespace_name(relnamespace), relname),
288 errdetail("System catalog modifications are currently disallowed.")));
289
290 /*
291 * Decide if we need storage or not, and handle a couple other special
292 * cases for particular relkinds.
293 */
294 switch (relkind)
295 {
296 case RELKIND_VIEW:
297 case RELKIND_COMPOSITE_TYPE:
298 case RELKIND_FOREIGN_TABLE:
299 case RELKIND_PARTITIONED_TABLE:
300 create_storage = false;
301
302 /*
303 * Force reltablespace to zero if the relation has no physical
304 * storage. This is mainly just for cleanliness' sake.
305 */
306 reltablespace = InvalidOid;
307 break;
308
309 case RELKIND_PARTITIONED_INDEX:
310 /*
311 * Preserve tablespace so that it's used as tablespace for indexes
312 * on future partitions.
313 */
314 create_storage = false;
315 break;
316
317 case RELKIND_SEQUENCE:
318 create_storage = true;
319
320 /*
321 * Force reltablespace to zero for sequences, since we don't
322 * support moving them around into different tablespaces.
323 */
324 reltablespace = InvalidOid;
325 break;
326 default:
327 create_storage = true;
328 break;
329 }
330
331 /*
332 * Unless otherwise requested, the physical ID (relfilenode) is initially
333 * the same as the logical ID (OID). When the caller did specify a
334 * relfilenode, it already exists; do not attempt to create it.
335 */
336 if (OidIsValid(relfilenode))
337 create_storage = false;
338 else
339 relfilenode = relid;
340
341 /*
342 * Never allow a pg_class entry to explicitly specify the database's
343 * default tablespace in reltablespace; force it to zero instead. This
344 * ensures that if the database is cloned with a different default
345 * tablespace, the pg_class entry will still match where CREATE DATABASE
346 * will put the physically copied relation.
347 *
348 * Yes, this is a bit of a hack.
349 */
350 if (reltablespace == MyDatabaseTableSpace)
351 reltablespace = InvalidOid;
352
353 /*
354 * build the relcache entry.
355 */
356 rel = RelationBuildLocalRelation(relname,
357 relnamespace,
358 tupDesc,
359 relid,
360 relfilenode,
361 reltablespace,
362 shared_relation,
363 mapped_relation,
364 relpersistence,
365 relkind);
366
367 /*
368 * Have the storage manager create the relation's disk file, if needed.
369 *
370 * We only create the main fork here, other forks will be created on
371 * demand.
372 */
373 if (create_storage)
374 {
375 RelationOpenSmgr(rel);
376 RelationCreateStorage(rel->rd_node, relpersistence);
377 }
378
379 /*
380 * If a tablespace is specified, removal of that tablespace is normally
381 * protected by the existence of a physical file; but for relations with
382 * no files, add a pg_shdepend entry to account for that.
383 */
384 if (!create_storage && reltablespace != InvalidOid)
385 recordDependencyOnTablespace(RelationRelationId, relid,
386 reltablespace);
387
388 return rel;
389 }
390
391 /* ----------------------------------------------------------------
392 * heap_create_with_catalog - Create a cataloged relation
393 *
394 * this is done in multiple steps:
395 *
396 * 1) CheckAttributeNamesTypes() is used to make certain the tuple
397 * descriptor contains a valid set of attribute names and types
398 *
399 * 2) pg_class is opened and get_relname_relid()
400 * performs a scan to ensure that no relation with the
401 * same name already exists.
402 *
403 * 3) heap_create() is called to create the new relation on disk.
404 *
405 * 4) TypeCreate() is called to define a new type corresponding
406 * to the new relation.
407 *
408 * 5) AddNewRelationTuple() is called to register the
409 * relation in pg_class.
410 *
411 * 6) AddNewAttributeTuples() is called to register the
412 * new relation's schema in pg_attribute.
413 *
414 * 7) StoreConstraints is called () - vadim 08/22/97
415 *
416 * 8) the relations are closed and the new relation's oid
417 * is returned.
418 *
419 * ----------------------------------------------------------------
420 */
421
422 /* --------------------------------
423 * CheckAttributeNamesTypes
424 *
425 * this is used to make certain the tuple descriptor contains a
426 * valid set of attribute names and datatypes. a problem simply
427 * generates ereport(ERROR) which aborts the current transaction.
428 * --------------------------------
429 */
430 void
CheckAttributeNamesTypes(TupleDesc tupdesc,char relkind,bool allow_system_table_mods)431 CheckAttributeNamesTypes(TupleDesc tupdesc, char relkind,
432 bool allow_system_table_mods)
433 {
434 int i;
435 int j;
436 int natts = tupdesc->natts;
437
438 /* Sanity check on column count */
439 if (natts < 0 || natts > MaxHeapAttributeNumber)
440 ereport(ERROR,
441 (errcode(ERRCODE_TOO_MANY_COLUMNS),
442 errmsg("tables can have at most %d columns",
443 MaxHeapAttributeNumber)));
444
445 /*
446 * first check for collision with system attribute names
447 *
448 * Skip this for a view or type relation, since those don't have system
449 * attributes.
450 */
451 if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
452 {
453 for (i = 0; i < natts; i++)
454 {
455 Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
456
457 if (SystemAttributeByName(NameStr(attr->attname),
458 tupdesc->tdhasoid) != NULL)
459 ereport(ERROR,
460 (errcode(ERRCODE_DUPLICATE_COLUMN),
461 errmsg("column name \"%s\" conflicts with a system column name",
462 NameStr(attr->attname))));
463 }
464 }
465
466 /*
467 * next check for repeated attribute names
468 */
469 for (i = 1; i < natts; i++)
470 {
471 for (j = 0; j < i; j++)
472 {
473 if (strcmp(NameStr(TupleDescAttr(tupdesc, j)->attname),
474 NameStr(TupleDescAttr(tupdesc, i)->attname)) == 0)
475 ereport(ERROR,
476 (errcode(ERRCODE_DUPLICATE_COLUMN),
477 errmsg("column name \"%s\" specified more than once",
478 NameStr(TupleDescAttr(tupdesc, j)->attname))));
479 }
480 }
481
482 /*
483 * next check the attribute types
484 */
485 for (i = 0; i < natts; i++)
486 {
487 CheckAttributeType(NameStr(TupleDescAttr(tupdesc, i)->attname),
488 TupleDescAttr(tupdesc, i)->atttypid,
489 TupleDescAttr(tupdesc, i)->attcollation,
490 NIL, /* assume we're creating a new rowtype */
491 allow_system_table_mods);
492 }
493 }
494
495 /* --------------------------------
496 * CheckAttributeType
497 *
498 * Verify that the proposed datatype of an attribute is legal.
499 * This is needed mainly because there are types (and pseudo-types)
500 * in the catalogs that we do not support as elements of real tuples.
501 * We also check some other properties required of a table column.
502 *
503 * If the attribute is being proposed for addition to an existing table or
504 * composite type, pass a one-element list of the rowtype OID as
505 * containing_rowtypes. When checking a to-be-created rowtype, it's
506 * sufficient to pass NIL, because there could not be any recursive reference
507 * to a not-yet-existing rowtype.
508 * --------------------------------
509 */
510 void
CheckAttributeType(const char * attname,Oid atttypid,Oid attcollation,List * containing_rowtypes,bool allow_system_table_mods)511 CheckAttributeType(const char *attname,
512 Oid atttypid, Oid attcollation,
513 List *containing_rowtypes,
514 bool allow_system_table_mods)
515 {
516 char att_typtype = get_typtype(atttypid);
517 Oid att_typelem;
518
519 if (att_typtype == TYPTYPE_PSEUDO)
520 {
521 /*
522 * Refuse any attempt to create a pseudo-type column, except for a
523 * special hack for pg_statistic: allow ANYARRAY when modifying system
524 * catalogs (this allows creating pg_statistic and cloning it during
525 * VACUUM FULL)
526 */
527 if (atttypid != ANYARRAYOID || !allow_system_table_mods)
528 ereport(ERROR,
529 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
530 errmsg("column \"%s\" has pseudo-type %s",
531 attname, format_type_be(atttypid))));
532 }
533 else if (att_typtype == TYPTYPE_DOMAIN)
534 {
535 /*
536 * If it's a domain, recurse to check its base type.
537 */
538 CheckAttributeType(attname, getBaseType(atttypid), attcollation,
539 containing_rowtypes,
540 allow_system_table_mods);
541 }
542 else if (att_typtype == TYPTYPE_COMPOSITE)
543 {
544 /*
545 * For a composite type, recurse into its attributes.
546 */
547 Relation relation;
548 TupleDesc tupdesc;
549 int i;
550
551 /*
552 * Check for self-containment. Eventually we might be able to allow
553 * this (just return without complaint, if so) but it's not clear how
554 * many other places would require anti-recursion defenses before it
555 * would be safe to allow tables to contain their own rowtype.
556 */
557 if (list_member_oid(containing_rowtypes, atttypid))
558 ereport(ERROR,
559 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
560 errmsg("composite type %s cannot be made a member of itself",
561 format_type_be(atttypid))));
562
563 containing_rowtypes = lcons_oid(atttypid, containing_rowtypes);
564
565 relation = relation_open(get_typ_typrelid(atttypid), AccessShareLock);
566
567 tupdesc = RelationGetDescr(relation);
568
569 for (i = 0; i < tupdesc->natts; i++)
570 {
571 Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
572
573 if (attr->attisdropped)
574 continue;
575 CheckAttributeType(NameStr(attr->attname),
576 attr->atttypid, attr->attcollation,
577 containing_rowtypes,
578 allow_system_table_mods);
579 }
580
581 relation_close(relation, AccessShareLock);
582
583 containing_rowtypes = list_delete_first(containing_rowtypes);
584 }
585 else if (att_typtype == TYPTYPE_RANGE)
586 {
587 /*
588 * If it's a range, recurse to check its subtype.
589 */
590 CheckAttributeType(attname, get_range_subtype(atttypid),
591 get_range_collation(atttypid),
592 containing_rowtypes,
593 allow_system_table_mods);
594 }
595 else if (OidIsValid((att_typelem = get_element_type(atttypid))))
596 {
597 /*
598 * Must recurse into array types, too, in case they are composite.
599 */
600 CheckAttributeType(attname, att_typelem, attcollation,
601 containing_rowtypes,
602 allow_system_table_mods);
603 }
604
605 /*
606 * This might not be strictly invalid per SQL standard, but it is pretty
607 * useless, and it cannot be dumped, so we must disallow it.
608 */
609 if (!OidIsValid(attcollation) && type_is_collatable(atttypid))
610 ereport(ERROR,
611 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
612 errmsg("no collation was derived for column \"%s\" with collatable type %s",
613 attname, format_type_be(atttypid)),
614 errhint("Use the COLLATE clause to set the collation explicitly.")));
615 }
616
617 /*
618 * InsertPgAttributeTuple
619 * Construct and insert a new tuple in pg_attribute.
620 *
621 * Caller has already opened and locked pg_attribute. new_attribute is the
622 * attribute to insert (but we ignore attacl and attoptions, which are always
623 * initialized to NULL).
624 *
625 * indstate is the index state for CatalogTupleInsertWithInfo. It can be
626 * passed as NULL, in which case we'll fetch the necessary info. (Don't do
627 * this when inserting multiple attributes, because it's a tad more
628 * expensive.)
629 */
630 void
InsertPgAttributeTuple(Relation pg_attribute_rel,Form_pg_attribute new_attribute,CatalogIndexState indstate)631 InsertPgAttributeTuple(Relation pg_attribute_rel,
632 Form_pg_attribute new_attribute,
633 CatalogIndexState indstate)
634 {
635 Datum values[Natts_pg_attribute];
636 bool nulls[Natts_pg_attribute];
637 HeapTuple tup;
638
639 /* This is a tad tedious, but way cleaner than what we used to do... */
640 memset(values, 0, sizeof(values));
641 memset(nulls, false, sizeof(nulls));
642
643 values[Anum_pg_attribute_attrelid - 1] = ObjectIdGetDatum(new_attribute->attrelid);
644 values[Anum_pg_attribute_attname - 1] = NameGetDatum(&new_attribute->attname);
645 values[Anum_pg_attribute_atttypid - 1] = ObjectIdGetDatum(new_attribute->atttypid);
646 values[Anum_pg_attribute_attstattarget - 1] = Int32GetDatum(new_attribute->attstattarget);
647 values[Anum_pg_attribute_attlen - 1] = Int16GetDatum(new_attribute->attlen);
648 values[Anum_pg_attribute_attnum - 1] = Int16GetDatum(new_attribute->attnum);
649 values[Anum_pg_attribute_attndims - 1] = Int32GetDatum(new_attribute->attndims);
650 values[Anum_pg_attribute_attcacheoff - 1] = Int32GetDatum(new_attribute->attcacheoff);
651 values[Anum_pg_attribute_atttypmod - 1] = Int32GetDatum(new_attribute->atttypmod);
652 values[Anum_pg_attribute_attbyval - 1] = BoolGetDatum(new_attribute->attbyval);
653 values[Anum_pg_attribute_attstorage - 1] = CharGetDatum(new_attribute->attstorage);
654 values[Anum_pg_attribute_attalign - 1] = CharGetDatum(new_attribute->attalign);
655 values[Anum_pg_attribute_attnotnull - 1] = BoolGetDatum(new_attribute->attnotnull);
656 values[Anum_pg_attribute_atthasdef - 1] = BoolGetDatum(new_attribute->atthasdef);
657 values[Anum_pg_attribute_atthasmissing - 1] = BoolGetDatum(new_attribute->atthasmissing);
658 values[Anum_pg_attribute_attidentity - 1] = CharGetDatum(new_attribute->attidentity);
659 values[Anum_pg_attribute_attisdropped - 1] = BoolGetDatum(new_attribute->attisdropped);
660 values[Anum_pg_attribute_attislocal - 1] = BoolGetDatum(new_attribute->attislocal);
661 values[Anum_pg_attribute_attinhcount - 1] = Int32GetDatum(new_attribute->attinhcount);
662 values[Anum_pg_attribute_attcollation - 1] = ObjectIdGetDatum(new_attribute->attcollation);
663
664 /* start out with empty permissions and empty options */
665 nulls[Anum_pg_attribute_attacl - 1] = true;
666 nulls[Anum_pg_attribute_attoptions - 1] = true;
667 nulls[Anum_pg_attribute_attfdwoptions - 1] = true;
668 nulls[Anum_pg_attribute_attmissingval - 1] = true;
669
670 tup = heap_form_tuple(RelationGetDescr(pg_attribute_rel), values, nulls);
671
672 /* finally insert the new tuple, update the indexes, and clean up */
673 if (indstate != NULL)
674 CatalogTupleInsertWithInfo(pg_attribute_rel, tup, indstate);
675 else
676 CatalogTupleInsert(pg_attribute_rel, tup);
677
678 heap_freetuple(tup);
679 }
680
681 /* --------------------------------
682 * AddNewAttributeTuples
683 *
684 * this registers the new relation's schema by adding
685 * tuples to pg_attribute.
686 * --------------------------------
687 */
688 static void
AddNewAttributeTuples(Oid new_rel_oid,TupleDesc tupdesc,char relkind,bool oidislocal,int oidinhcount)689 AddNewAttributeTuples(Oid new_rel_oid,
690 TupleDesc tupdesc,
691 char relkind,
692 bool oidislocal,
693 int oidinhcount)
694 {
695 Form_pg_attribute attr;
696 int i;
697 Relation rel;
698 CatalogIndexState indstate;
699 int natts = tupdesc->natts;
700 ObjectAddress myself,
701 referenced;
702
703 /*
704 * open pg_attribute and its indexes.
705 */
706 rel = heap_open(AttributeRelationId, RowExclusiveLock);
707
708 indstate = CatalogOpenIndexes(rel);
709
710 /*
711 * First we add the user attributes. This is also a convenient place to
712 * add dependencies on their datatypes and collations.
713 */
714 for (i = 0; i < natts; i++)
715 {
716 attr = TupleDescAttr(tupdesc, i);
717 /* Fill in the correct relation OID */
718 attr->attrelid = new_rel_oid;
719 /* Make sure these are OK, too */
720 attr->attstattarget = -1;
721 attr->attcacheoff = -1;
722
723 InsertPgAttributeTuple(rel, attr, indstate);
724
725 /* Add dependency info */
726 myself.classId = RelationRelationId;
727 myself.objectId = new_rel_oid;
728 myself.objectSubId = i + 1;
729 referenced.classId = TypeRelationId;
730 referenced.objectId = attr->atttypid;
731 referenced.objectSubId = 0;
732 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
733
734 /* The default collation is pinned, so don't bother recording it */
735 if (OidIsValid(attr->attcollation) &&
736 attr->attcollation != DEFAULT_COLLATION_OID)
737 {
738 referenced.classId = CollationRelationId;
739 referenced.objectId = attr->attcollation;
740 referenced.objectSubId = 0;
741 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
742 }
743 }
744
745 /*
746 * Next we add the system attributes. Skip OID if rel has no OIDs. Skip
747 * all for a view or type relation. We don't bother with making datatype
748 * dependencies here, since presumably all these types are pinned.
749 */
750 if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
751 {
752 for (i = 0; i < (int) lengthof(SysAtt); i++)
753 {
754 FormData_pg_attribute attStruct;
755
756 /* skip OID where appropriate */
757 if (!tupdesc->tdhasoid &&
758 SysAtt[i]->attnum == ObjectIdAttributeNumber)
759 continue;
760
761 memcpy(&attStruct, (char *) SysAtt[i], sizeof(FormData_pg_attribute));
762
763 /* Fill in the correct relation OID in the copied tuple */
764 attStruct.attrelid = new_rel_oid;
765
766 /* Fill in correct inheritance info for the OID column */
767 if (attStruct.attnum == ObjectIdAttributeNumber)
768 {
769 attStruct.attislocal = oidislocal;
770 attStruct.attinhcount = oidinhcount;
771 }
772
773 InsertPgAttributeTuple(rel, &attStruct, indstate);
774 }
775 }
776
777 /*
778 * clean up
779 */
780 CatalogCloseIndexes(indstate);
781
782 heap_close(rel, RowExclusiveLock);
783 }
784
785 /* --------------------------------
786 * InsertPgClassTuple
787 *
788 * Construct and insert a new tuple in pg_class.
789 *
790 * Caller has already opened and locked pg_class.
791 * Tuple data is taken from new_rel_desc->rd_rel, except for the
792 * variable-width fields which are not present in a cached reldesc.
793 * relacl and reloptions are passed in Datum form (to avoid having
794 * to reference the data types in heap.h). Pass (Datum) 0 to set them
795 * to NULL.
796 * --------------------------------
797 */
798 void
InsertPgClassTuple(Relation pg_class_desc,Relation new_rel_desc,Oid new_rel_oid,Datum relacl,Datum reloptions)799 InsertPgClassTuple(Relation pg_class_desc,
800 Relation new_rel_desc,
801 Oid new_rel_oid,
802 Datum relacl,
803 Datum reloptions)
804 {
805 Form_pg_class rd_rel = new_rel_desc->rd_rel;
806 Datum values[Natts_pg_class];
807 bool nulls[Natts_pg_class];
808 HeapTuple tup;
809
810 /* This is a tad tedious, but way cleaner than what we used to do... */
811 memset(values, 0, sizeof(values));
812 memset(nulls, false, sizeof(nulls));
813
814 values[Anum_pg_class_relname - 1] = NameGetDatum(&rd_rel->relname);
815 values[Anum_pg_class_relnamespace - 1] = ObjectIdGetDatum(rd_rel->relnamespace);
816 values[Anum_pg_class_reltype - 1] = ObjectIdGetDatum(rd_rel->reltype);
817 values[Anum_pg_class_reloftype - 1] = ObjectIdGetDatum(rd_rel->reloftype);
818 values[Anum_pg_class_relowner - 1] = ObjectIdGetDatum(rd_rel->relowner);
819 values[Anum_pg_class_relam - 1] = ObjectIdGetDatum(rd_rel->relam);
820 values[Anum_pg_class_relfilenode - 1] = ObjectIdGetDatum(rd_rel->relfilenode);
821 values[Anum_pg_class_reltablespace - 1] = ObjectIdGetDatum(rd_rel->reltablespace);
822 values[Anum_pg_class_relpages - 1] = Int32GetDatum(rd_rel->relpages);
823 values[Anum_pg_class_reltuples - 1] = Float4GetDatum(rd_rel->reltuples);
824 values[Anum_pg_class_relallvisible - 1] = Int32GetDatum(rd_rel->relallvisible);
825 values[Anum_pg_class_reltoastrelid - 1] = ObjectIdGetDatum(rd_rel->reltoastrelid);
826 values[Anum_pg_class_relhasindex - 1] = BoolGetDatum(rd_rel->relhasindex);
827 values[Anum_pg_class_relisshared - 1] = BoolGetDatum(rd_rel->relisshared);
828 values[Anum_pg_class_relpersistence - 1] = CharGetDatum(rd_rel->relpersistence);
829 values[Anum_pg_class_relkind - 1] = CharGetDatum(rd_rel->relkind);
830 values[Anum_pg_class_relnatts - 1] = Int16GetDatum(rd_rel->relnatts);
831 values[Anum_pg_class_relchecks - 1] = Int16GetDatum(rd_rel->relchecks);
832 values[Anum_pg_class_relhasoids - 1] = BoolGetDatum(rd_rel->relhasoids);
833 values[Anum_pg_class_relhasrules - 1] = BoolGetDatum(rd_rel->relhasrules);
834 values[Anum_pg_class_relhastriggers - 1] = BoolGetDatum(rd_rel->relhastriggers);
835 values[Anum_pg_class_relrowsecurity - 1] = BoolGetDatum(rd_rel->relrowsecurity);
836 values[Anum_pg_class_relforcerowsecurity - 1] = BoolGetDatum(rd_rel->relforcerowsecurity);
837 values[Anum_pg_class_relhassubclass - 1] = BoolGetDatum(rd_rel->relhassubclass);
838 values[Anum_pg_class_relispopulated - 1] = BoolGetDatum(rd_rel->relispopulated);
839 values[Anum_pg_class_relreplident - 1] = CharGetDatum(rd_rel->relreplident);
840 values[Anum_pg_class_relispartition - 1] = BoolGetDatum(rd_rel->relispartition);
841 values[Anum_pg_class_relrewrite - 1] = ObjectIdGetDatum(rd_rel->relrewrite);
842 values[Anum_pg_class_relfrozenxid - 1] = TransactionIdGetDatum(rd_rel->relfrozenxid);
843 values[Anum_pg_class_relminmxid - 1] = MultiXactIdGetDatum(rd_rel->relminmxid);
844 if (relacl != (Datum) 0)
845 values[Anum_pg_class_relacl - 1] = relacl;
846 else
847 nulls[Anum_pg_class_relacl - 1] = true;
848 if (reloptions != (Datum) 0)
849 values[Anum_pg_class_reloptions - 1] = reloptions;
850 else
851 nulls[Anum_pg_class_reloptions - 1] = true;
852
853 /* relpartbound is set by updating this tuple, if necessary */
854 nulls[Anum_pg_class_relpartbound - 1] = true;
855
856 tup = heap_form_tuple(RelationGetDescr(pg_class_desc), values, nulls);
857
858 /*
859 * The new tuple must have the oid already chosen for the rel. Sure would
860 * be embarrassing to do this sort of thing in polite company.
861 */
862 HeapTupleSetOid(tup, new_rel_oid);
863
864 /* finally insert the new tuple, update the indexes, and clean up */
865 CatalogTupleInsert(pg_class_desc, tup);
866
867 heap_freetuple(tup);
868 }
869
870 /* --------------------------------
871 * AddNewRelationTuple
872 *
873 * this registers the new relation in the catalogs by
874 * adding a tuple to pg_class.
875 * --------------------------------
876 */
877 static void
AddNewRelationTuple(Relation pg_class_desc,Relation new_rel_desc,Oid new_rel_oid,Oid new_type_oid,Oid reloftype,Oid relowner,char relkind,Datum relacl,Datum reloptions)878 AddNewRelationTuple(Relation pg_class_desc,
879 Relation new_rel_desc,
880 Oid new_rel_oid,
881 Oid new_type_oid,
882 Oid reloftype,
883 Oid relowner,
884 char relkind,
885 Datum relacl,
886 Datum reloptions)
887 {
888 Form_pg_class new_rel_reltup;
889
890 /*
891 * first we update some of the information in our uncataloged relation's
892 * relation descriptor.
893 */
894 new_rel_reltup = new_rel_desc->rd_rel;
895
896 switch (relkind)
897 {
898 case RELKIND_RELATION:
899 case RELKIND_MATVIEW:
900 case RELKIND_INDEX:
901 case RELKIND_TOASTVALUE:
902 /* The relation is real, but as yet empty */
903 new_rel_reltup->relpages = 0;
904 new_rel_reltup->reltuples = 0;
905 new_rel_reltup->relallvisible = 0;
906 break;
907 case RELKIND_SEQUENCE:
908 /* Sequences always have a known size */
909 new_rel_reltup->relpages = 1;
910 new_rel_reltup->reltuples = 1;
911 new_rel_reltup->relallvisible = 0;
912 break;
913 default:
914 /* Views, etc, have no disk storage */
915 new_rel_reltup->relpages = 0;
916 new_rel_reltup->reltuples = 0;
917 new_rel_reltup->relallvisible = 0;
918 break;
919 }
920
921 /* Initialize relfrozenxid and relminmxid */
922 if (relkind == RELKIND_RELATION ||
923 relkind == RELKIND_MATVIEW ||
924 relkind == RELKIND_TOASTVALUE)
925 {
926 /*
927 * Initialize to the minimum XID that could put tuples in the table.
928 * We know that no xacts older than RecentXmin are still running, so
929 * that will do.
930 */
931 new_rel_reltup->relfrozenxid = RecentXmin;
932
933 /*
934 * Similarly, initialize the minimum Multixact to the first value that
935 * could possibly be stored in tuples in the table. Running
936 * transactions could reuse values from their local cache, so we are
937 * careful to consider all currently running multis.
938 *
939 * XXX this could be refined further, but is it worth the hassle?
940 */
941 new_rel_reltup->relminmxid = GetOldestMultiXactId();
942 }
943 else
944 {
945 /*
946 * Other relation types will not contain XIDs, so set relfrozenxid to
947 * InvalidTransactionId. (Note: a sequence does contain a tuple, but
948 * we force its xmin to be FrozenTransactionId always; see
949 * commands/sequence.c.)
950 */
951 new_rel_reltup->relfrozenxid = InvalidTransactionId;
952 new_rel_reltup->relminmxid = InvalidMultiXactId;
953 }
954
955 new_rel_reltup->relowner = relowner;
956 new_rel_reltup->reltype = new_type_oid;
957 new_rel_reltup->reloftype = reloftype;
958
959 /* relispartition is always set by updating this tuple later */
960 new_rel_reltup->relispartition = false;
961
962 new_rel_desc->rd_att->tdtypeid = new_type_oid;
963
964 /* Now build and insert the tuple */
965 InsertPgClassTuple(pg_class_desc, new_rel_desc, new_rel_oid,
966 relacl, reloptions);
967 }
968
969
970 /* --------------------------------
971 * AddNewRelationType -
972 *
973 * define a composite type corresponding to the new relation
974 * --------------------------------
975 */
976 static ObjectAddress
AddNewRelationType(const char * typeName,Oid typeNamespace,Oid new_rel_oid,char new_rel_kind,Oid ownerid,Oid new_row_type,Oid new_array_type)977 AddNewRelationType(const char *typeName,
978 Oid typeNamespace,
979 Oid new_rel_oid,
980 char new_rel_kind,
981 Oid ownerid,
982 Oid new_row_type,
983 Oid new_array_type)
984 {
985 return
986 TypeCreate(new_row_type, /* optional predetermined OID */
987 typeName, /* type name */
988 typeNamespace, /* type namespace */
989 new_rel_oid, /* relation oid */
990 new_rel_kind, /* relation kind */
991 ownerid, /* owner's ID */
992 -1, /* internal size (varlena) */
993 TYPTYPE_COMPOSITE, /* type-type (composite) */
994 TYPCATEGORY_COMPOSITE, /* type-category (ditto) */
995 false, /* composite types are never preferred */
996 DEFAULT_TYPDELIM, /* default array delimiter */
997 F_RECORD_IN, /* input procedure */
998 F_RECORD_OUT, /* output procedure */
999 F_RECORD_RECV, /* receive procedure */
1000 F_RECORD_SEND, /* send procedure */
1001 InvalidOid, /* typmodin procedure - none */
1002 InvalidOid, /* typmodout procedure - none */
1003 InvalidOid, /* analyze procedure - default */
1004 InvalidOid, /* array element type - irrelevant */
1005 false, /* this is not an array type */
1006 new_array_type, /* array type if any */
1007 InvalidOid, /* domain base type - irrelevant */
1008 NULL, /* default value - none */
1009 NULL, /* default binary representation */
1010 false, /* passed by reference */
1011 'd', /* alignment - must be the largest! */
1012 'x', /* fully TOASTable */
1013 -1, /* typmod */
1014 0, /* array dimensions for typBaseType */
1015 false, /* Type NOT NULL */
1016 InvalidOid); /* rowtypes never have a collation */
1017 }
1018
1019 /* --------------------------------
1020 * heap_create_with_catalog
1021 *
1022 * creates a new cataloged relation. see comments above.
1023 *
1024 * Arguments:
1025 * relname: name to give to new rel
1026 * relnamespace: OID of namespace it goes in
1027 * reltablespace: OID of tablespace it goes in
1028 * relid: OID to assign to new rel, or InvalidOid to select a new OID
1029 * reltypeid: OID to assign to rel's rowtype, or InvalidOid to select one
1030 * reloftypeid: if a typed table, OID of underlying type; else InvalidOid
1031 * ownerid: OID of new rel's owner
1032 * tupdesc: tuple descriptor (source of column definitions)
1033 * cooked_constraints: list of precooked check constraints and defaults
1034 * relkind: relkind for new rel
1035 * relpersistence: rel's persistence status (permanent, temp, or unlogged)
1036 * shared_relation: true if it's to be a shared relation
1037 * mapped_relation: true if the relation will use the relfilenode map
1038 * oidislocal: true if oid column (if any) should be marked attislocal
1039 * oidinhcount: attinhcount to assign to oid column (if any)
1040 * oncommit: ON COMMIT marking (only relevant if it's a temp table)
1041 * reloptions: reloptions in Datum form, or (Datum) 0 if none
1042 * use_user_acl: true if should look for user-defined default permissions;
1043 * if false, relacl is always set NULL
1044 * allow_system_table_mods: true to allow creation in system namespaces
1045 * is_internal: is this a system-generated catalog?
1046 *
1047 * Output parameters:
1048 * typaddress: if not null, gets the object address of the new pg_type entry
1049 *
1050 * Returns the OID of the new relation
1051 * --------------------------------
1052 */
1053 Oid
heap_create_with_catalog(const char * relname,Oid relnamespace,Oid reltablespace,Oid relid,Oid reltypeid,Oid reloftypeid,Oid ownerid,TupleDesc tupdesc,List * cooked_constraints,char relkind,char relpersistence,bool shared_relation,bool mapped_relation,bool oidislocal,int oidinhcount,OnCommitAction oncommit,Datum reloptions,bool use_user_acl,bool allow_system_table_mods,bool is_internal,Oid relrewrite,ObjectAddress * typaddress)1054 heap_create_with_catalog(const char *relname,
1055 Oid relnamespace,
1056 Oid reltablespace,
1057 Oid relid,
1058 Oid reltypeid,
1059 Oid reloftypeid,
1060 Oid ownerid,
1061 TupleDesc tupdesc,
1062 List *cooked_constraints,
1063 char relkind,
1064 char relpersistence,
1065 bool shared_relation,
1066 bool mapped_relation,
1067 bool oidislocal,
1068 int oidinhcount,
1069 OnCommitAction oncommit,
1070 Datum reloptions,
1071 bool use_user_acl,
1072 bool allow_system_table_mods,
1073 bool is_internal,
1074 Oid relrewrite,
1075 ObjectAddress *typaddress)
1076 {
1077 Relation pg_class_desc;
1078 Relation new_rel_desc;
1079 Acl *relacl;
1080 Oid existing_relid;
1081 Oid old_type_oid;
1082 Oid new_type_oid;
1083 ObjectAddress new_type_addr;
1084 Oid new_array_oid = InvalidOid;
1085
1086 pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
1087
1088 /*
1089 * sanity checks
1090 */
1091 Assert(IsNormalProcessingMode() || IsBootstrapProcessingMode());
1092
1093 CheckAttributeNamesTypes(tupdesc, relkind, allow_system_table_mods);
1094
1095 /*
1096 * This would fail later on anyway, if the relation already exists. But
1097 * by catching it here we can emit a nicer error message.
1098 */
1099 existing_relid = get_relname_relid(relname, relnamespace);
1100 if (existing_relid != InvalidOid)
1101 ereport(ERROR,
1102 (errcode(ERRCODE_DUPLICATE_TABLE),
1103 errmsg("relation \"%s\" already exists", relname)));
1104
1105 /*
1106 * Since we are going to create a rowtype as well, also check for
1107 * collision with an existing type name. If there is one and it's an
1108 * autogenerated array, we can rename it out of the way; otherwise we can
1109 * at least give a good error message.
1110 */
1111 old_type_oid = GetSysCacheOid2(TYPENAMENSP,
1112 CStringGetDatum(relname),
1113 ObjectIdGetDatum(relnamespace));
1114 if (OidIsValid(old_type_oid))
1115 {
1116 if (!moveArrayTypeName(old_type_oid, relname, relnamespace))
1117 ereport(ERROR,
1118 (errcode(ERRCODE_DUPLICATE_OBJECT),
1119 errmsg("type \"%s\" already exists", relname),
1120 errhint("A relation has an associated type of the same name, "
1121 "so you must use a name that doesn't conflict "
1122 "with any existing type.")));
1123 }
1124
1125 /*
1126 * Shared relations must be in pg_global (last-ditch check)
1127 */
1128 if (shared_relation && reltablespace != GLOBALTABLESPACE_OID)
1129 elog(ERROR, "shared relations must be placed in pg_global tablespace");
1130
1131 /*
1132 * Allocate an OID for the relation, unless we were told what to use.
1133 *
1134 * The OID will be the relfilenode as well, so make sure it doesn't
1135 * collide with either pg_class OIDs or existing physical files.
1136 */
1137 if (!OidIsValid(relid))
1138 {
1139 /* Use binary-upgrade override for pg_class.oid/relfilenode? */
1140 if (IsBinaryUpgrade &&
1141 (relkind == RELKIND_RELATION || relkind == RELKIND_SEQUENCE ||
1142 relkind == RELKIND_VIEW || relkind == RELKIND_MATVIEW ||
1143 relkind == RELKIND_COMPOSITE_TYPE || relkind == RELKIND_FOREIGN_TABLE ||
1144 relkind == RELKIND_PARTITIONED_TABLE))
1145 {
1146 if (!OidIsValid(binary_upgrade_next_heap_pg_class_oid))
1147 ereport(ERROR,
1148 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1149 errmsg("pg_class heap OID value not set when in binary upgrade mode")));
1150
1151 relid = binary_upgrade_next_heap_pg_class_oid;
1152 binary_upgrade_next_heap_pg_class_oid = InvalidOid;
1153 }
1154 /* There might be no TOAST table, so we have to test for it. */
1155 else if (IsBinaryUpgrade &&
1156 OidIsValid(binary_upgrade_next_toast_pg_class_oid) &&
1157 relkind == RELKIND_TOASTVALUE)
1158 {
1159 relid = binary_upgrade_next_toast_pg_class_oid;
1160 binary_upgrade_next_toast_pg_class_oid = InvalidOid;
1161 }
1162 else
1163 relid = GetNewRelFileNode(reltablespace, pg_class_desc,
1164 relpersistence);
1165 }
1166
1167 /*
1168 * Determine the relation's initial permissions.
1169 */
1170 if (use_user_acl)
1171 {
1172 switch (relkind)
1173 {
1174 case RELKIND_RELATION:
1175 case RELKIND_VIEW:
1176 case RELKIND_MATVIEW:
1177 case RELKIND_FOREIGN_TABLE:
1178 case RELKIND_PARTITIONED_TABLE:
1179 relacl = get_user_default_acl(OBJECT_TABLE, ownerid,
1180 relnamespace);
1181 break;
1182 case RELKIND_SEQUENCE:
1183 relacl = get_user_default_acl(OBJECT_SEQUENCE, ownerid,
1184 relnamespace);
1185 break;
1186 default:
1187 relacl = NULL;
1188 break;
1189 }
1190 }
1191 else
1192 relacl = NULL;
1193
1194 /*
1195 * Create the relcache entry (mostly dummy at this point) and the physical
1196 * disk file. (If we fail further down, it's the smgr's responsibility to
1197 * remove the disk file again.)
1198 */
1199 new_rel_desc = heap_create(relname,
1200 relnamespace,
1201 reltablespace,
1202 relid,
1203 InvalidOid,
1204 tupdesc,
1205 relkind,
1206 relpersistence,
1207 shared_relation,
1208 mapped_relation,
1209 allow_system_table_mods);
1210
1211 Assert(relid == RelationGetRelid(new_rel_desc));
1212
1213 new_rel_desc->rd_rel->relrewrite = relrewrite;
1214
1215 /*
1216 * Decide whether to create an array type over the relation's rowtype. We
1217 * do not create any array types for system catalogs (ie, those made
1218 * during initdb). We do not create them where the use of a relation as
1219 * such is an implementation detail: toast tables, sequences and indexes.
1220 */
1221 if (IsUnderPostmaster && (relkind == RELKIND_RELATION ||
1222 relkind == RELKIND_VIEW ||
1223 relkind == RELKIND_MATVIEW ||
1224 relkind == RELKIND_FOREIGN_TABLE ||
1225 relkind == RELKIND_COMPOSITE_TYPE ||
1226 relkind == RELKIND_PARTITIONED_TABLE))
1227 new_array_oid = AssignTypeArrayOid();
1228
1229 /*
1230 * Since defining a relation also defines a complex type, we add a new
1231 * system type corresponding to the new relation. The OID of the type can
1232 * be preselected by the caller, but if reltypeid is InvalidOid, we'll
1233 * generate a new OID for it.
1234 *
1235 * NOTE: we could get a unique-index failure here, in case someone else is
1236 * creating the same type name in parallel but hadn't committed yet when
1237 * we checked for a duplicate name above.
1238 */
1239 new_type_addr = AddNewRelationType(relname,
1240 relnamespace,
1241 relid,
1242 relkind,
1243 ownerid,
1244 reltypeid,
1245 new_array_oid);
1246 new_type_oid = new_type_addr.objectId;
1247 if (typaddress)
1248 *typaddress = new_type_addr;
1249
1250 /*
1251 * Now make the array type if wanted.
1252 */
1253 if (OidIsValid(new_array_oid))
1254 {
1255 char *relarrayname;
1256
1257 relarrayname = makeArrayTypeName(relname, relnamespace);
1258
1259 TypeCreate(new_array_oid, /* force the type's OID to this */
1260 relarrayname, /* Array type name */
1261 relnamespace, /* Same namespace as parent */
1262 InvalidOid, /* Not composite, no relationOid */
1263 0, /* relkind, also N/A here */
1264 ownerid, /* owner's ID */
1265 -1, /* Internal size (varlena) */
1266 TYPTYPE_BASE, /* Not composite - typelem is */
1267 TYPCATEGORY_ARRAY, /* type-category (array) */
1268 false, /* array types are never preferred */
1269 DEFAULT_TYPDELIM, /* default array delimiter */
1270 F_ARRAY_IN, /* array input proc */
1271 F_ARRAY_OUT, /* array output proc */
1272 F_ARRAY_RECV, /* array recv (bin) proc */
1273 F_ARRAY_SEND, /* array send (bin) proc */
1274 InvalidOid, /* typmodin procedure - none */
1275 InvalidOid, /* typmodout procedure - none */
1276 F_ARRAY_TYPANALYZE, /* array analyze procedure */
1277 new_type_oid, /* array element type - the rowtype */
1278 true, /* yes, this is an array type */
1279 InvalidOid, /* this has no array type */
1280 InvalidOid, /* domain base type - irrelevant */
1281 NULL, /* default value - none */
1282 NULL, /* default binary representation */
1283 false, /* passed by reference */
1284 'd', /* alignment - must be the largest! */
1285 'x', /* fully TOASTable */
1286 -1, /* typmod */
1287 0, /* array dimensions for typBaseType */
1288 false, /* Type NOT NULL */
1289 InvalidOid); /* rowtypes never have a collation */
1290
1291 pfree(relarrayname);
1292 }
1293
1294 /*
1295 * now create an entry in pg_class for the relation.
1296 *
1297 * NOTE: we could get a unique-index failure here, in case someone else is
1298 * creating the same relation name in parallel but hadn't committed yet
1299 * when we checked for a duplicate name above.
1300 */
1301 AddNewRelationTuple(pg_class_desc,
1302 new_rel_desc,
1303 relid,
1304 new_type_oid,
1305 reloftypeid,
1306 ownerid,
1307 relkind,
1308 PointerGetDatum(relacl),
1309 reloptions);
1310
1311 /*
1312 * now add tuples to pg_attribute for the attributes in our new relation.
1313 */
1314 AddNewAttributeTuples(relid, new_rel_desc->rd_att, relkind,
1315 oidislocal, oidinhcount);
1316
1317 /*
1318 * Make a dependency link to force the relation to be deleted if its
1319 * namespace is. Also make a dependency link to its owner, as well as
1320 * dependencies for any roles mentioned in the default ACL.
1321 *
1322 * For composite types, these dependencies are tracked for the pg_type
1323 * entry, so we needn't record them here. Likewise, TOAST tables don't
1324 * need a namespace dependency (they live in a pinned namespace) nor an
1325 * owner dependency (they depend indirectly through the parent table), nor
1326 * should they have any ACL entries. The same applies for extension
1327 * dependencies.
1328 *
1329 * Also, skip this in bootstrap mode, since we don't make dependencies
1330 * while bootstrapping.
1331 */
1332 if (relkind != RELKIND_COMPOSITE_TYPE &&
1333 relkind != RELKIND_TOASTVALUE &&
1334 !IsBootstrapProcessingMode())
1335 {
1336 ObjectAddress myself,
1337 referenced;
1338
1339 myself.classId = RelationRelationId;
1340 myself.objectId = relid;
1341 myself.objectSubId = 0;
1342
1343 referenced.classId = NamespaceRelationId;
1344 referenced.objectId = relnamespace;
1345 referenced.objectSubId = 0;
1346 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
1347
1348 recordDependencyOnOwner(RelationRelationId, relid, ownerid);
1349
1350 recordDependencyOnNewAcl(RelationRelationId, relid, 0, ownerid, relacl);
1351
1352 recordDependencyOnCurrentExtension(&myself, false);
1353
1354 if (reloftypeid)
1355 {
1356 referenced.classId = TypeRelationId;
1357 referenced.objectId = reloftypeid;
1358 referenced.objectSubId = 0;
1359 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
1360 }
1361 }
1362
1363 /* Post creation hook for new relation */
1364 InvokeObjectPostCreateHookArg(RelationRelationId, relid, 0, is_internal);
1365
1366 /*
1367 * Store any supplied constraints and defaults.
1368 *
1369 * NB: this may do a CommandCounterIncrement and rebuild the relcache
1370 * entry, so the relation must be valid and self-consistent at this point.
1371 * In particular, there are not yet constraints and defaults anywhere.
1372 */
1373 StoreConstraints(new_rel_desc, cooked_constraints, is_internal);
1374
1375 /*
1376 * If there's a special on-commit action, remember it
1377 */
1378 if (oncommit != ONCOMMIT_NOOP)
1379 register_on_commit_action(relid, oncommit);
1380
1381 /*
1382 * Unlogged objects need an init fork, except for partitioned tables which
1383 * have no storage at all.
1384 */
1385 if (relpersistence == RELPERSISTENCE_UNLOGGED &&
1386 relkind != RELKIND_PARTITIONED_TABLE)
1387 heap_create_init_fork(new_rel_desc);
1388
1389 /*
1390 * ok, the relation has been cataloged, so close our relations and return
1391 * the OID of the newly created relation.
1392 */
1393 heap_close(new_rel_desc, NoLock); /* do not unlock till end of xact */
1394 heap_close(pg_class_desc, RowExclusiveLock);
1395
1396 return relid;
1397 }
1398
1399 /*
1400 * Set up an init fork for an unlogged table so that it can be correctly
1401 * reinitialized on restart. An immediate sync is required even if the
1402 * page has been logged, because the write did not go through
1403 * shared_buffers and therefore a concurrent checkpoint may have moved
1404 * the redo pointer past our xlog record. Recovery may as well remove it
1405 * while replaying, for example, XLOG_DBASE_CREATE or XLOG_TBLSPC_CREATE
1406 * record. Therefore, logging is necessary even if wal_level=minimal.
1407 */
1408 void
heap_create_init_fork(Relation rel)1409 heap_create_init_fork(Relation rel)
1410 {
1411 Assert(rel->rd_rel->relkind == RELKIND_RELATION ||
1412 rel->rd_rel->relkind == RELKIND_MATVIEW ||
1413 rel->rd_rel->relkind == RELKIND_TOASTVALUE);
1414 RelationOpenSmgr(rel);
1415 smgrcreate(rel->rd_smgr, INIT_FORKNUM, false);
1416 log_smgrcreate(&rel->rd_smgr->smgr_rnode.node, INIT_FORKNUM);
1417 smgrimmedsync(rel->rd_smgr, INIT_FORKNUM);
1418 }
1419
1420 /*
1421 * RelationRemoveInheritance
1422 *
1423 * Formerly, this routine checked for child relations and aborted the
1424 * deletion if any were found. Now we rely on the dependency mechanism
1425 * to check for or delete child relations. By the time we get here,
1426 * there are no children and we need only remove any pg_inherits rows
1427 * linking this relation to its parent(s).
1428 */
1429 static void
RelationRemoveInheritance(Oid relid)1430 RelationRemoveInheritance(Oid relid)
1431 {
1432 Relation catalogRelation;
1433 SysScanDesc scan;
1434 ScanKeyData key;
1435 HeapTuple tuple;
1436
1437 catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
1438
1439 ScanKeyInit(&key,
1440 Anum_pg_inherits_inhrelid,
1441 BTEqualStrategyNumber, F_OIDEQ,
1442 ObjectIdGetDatum(relid));
1443
1444 scan = systable_beginscan(catalogRelation, InheritsRelidSeqnoIndexId, true,
1445 NULL, 1, &key);
1446
1447 while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1448 CatalogTupleDelete(catalogRelation, &tuple->t_self);
1449
1450 systable_endscan(scan);
1451 heap_close(catalogRelation, RowExclusiveLock);
1452 }
1453
1454 /*
1455 * DeleteRelationTuple
1456 *
1457 * Remove pg_class row for the given relid.
1458 *
1459 * Note: this is shared by relation deletion and index deletion. It's
1460 * not intended for use anyplace else.
1461 */
1462 void
DeleteRelationTuple(Oid relid)1463 DeleteRelationTuple(Oid relid)
1464 {
1465 Relation pg_class_desc;
1466 HeapTuple tup;
1467
1468 /* Grab an appropriate lock on the pg_class relation */
1469 pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
1470
1471 tup = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
1472 if (!HeapTupleIsValid(tup))
1473 elog(ERROR, "cache lookup failed for relation %u", relid);
1474
1475 /* delete the relation tuple from pg_class, and finish up */
1476 CatalogTupleDelete(pg_class_desc, &tup->t_self);
1477
1478 ReleaseSysCache(tup);
1479
1480 heap_close(pg_class_desc, RowExclusiveLock);
1481 }
1482
1483 /*
1484 * DeleteAttributeTuples
1485 *
1486 * Remove pg_attribute rows for the given relid.
1487 *
1488 * Note: this is shared by relation deletion and index deletion. It's
1489 * not intended for use anyplace else.
1490 */
1491 void
DeleteAttributeTuples(Oid relid)1492 DeleteAttributeTuples(Oid relid)
1493 {
1494 Relation attrel;
1495 SysScanDesc scan;
1496 ScanKeyData key[1];
1497 HeapTuple atttup;
1498
1499 /* Grab an appropriate lock on the pg_attribute relation */
1500 attrel = heap_open(AttributeRelationId, RowExclusiveLock);
1501
1502 /* Use the index to scan only attributes of the target relation */
1503 ScanKeyInit(&key[0],
1504 Anum_pg_attribute_attrelid,
1505 BTEqualStrategyNumber, F_OIDEQ,
1506 ObjectIdGetDatum(relid));
1507
1508 scan = systable_beginscan(attrel, AttributeRelidNumIndexId, true,
1509 NULL, 1, key);
1510
1511 /* Delete all the matching tuples */
1512 while ((atttup = systable_getnext(scan)) != NULL)
1513 CatalogTupleDelete(attrel, &atttup->t_self);
1514
1515 /* Clean up after the scan */
1516 systable_endscan(scan);
1517 heap_close(attrel, RowExclusiveLock);
1518 }
1519
1520 /*
1521 * DeleteSystemAttributeTuples
1522 *
1523 * Remove pg_attribute rows for system columns of the given relid.
1524 *
1525 * Note: this is only used when converting a table to a view. Views don't
1526 * have system columns, so we should remove them from pg_attribute.
1527 */
1528 void
DeleteSystemAttributeTuples(Oid relid)1529 DeleteSystemAttributeTuples(Oid relid)
1530 {
1531 Relation attrel;
1532 SysScanDesc scan;
1533 ScanKeyData key[2];
1534 HeapTuple atttup;
1535
1536 /* Grab an appropriate lock on the pg_attribute relation */
1537 attrel = heap_open(AttributeRelationId, RowExclusiveLock);
1538
1539 /* Use the index to scan only system attributes of the target relation */
1540 ScanKeyInit(&key[0],
1541 Anum_pg_attribute_attrelid,
1542 BTEqualStrategyNumber, F_OIDEQ,
1543 ObjectIdGetDatum(relid));
1544 ScanKeyInit(&key[1],
1545 Anum_pg_attribute_attnum,
1546 BTLessEqualStrategyNumber, F_INT2LE,
1547 Int16GetDatum(0));
1548
1549 scan = systable_beginscan(attrel, AttributeRelidNumIndexId, true,
1550 NULL, 2, key);
1551
1552 /* Delete all the matching tuples */
1553 while ((atttup = systable_getnext(scan)) != NULL)
1554 CatalogTupleDelete(attrel, &atttup->t_self);
1555
1556 /* Clean up after the scan */
1557 systable_endscan(scan);
1558 heap_close(attrel, RowExclusiveLock);
1559 }
1560
1561 /*
1562 * RemoveAttributeById
1563 *
1564 * This is the guts of ALTER TABLE DROP COLUMN: actually mark the attribute
1565 * deleted in pg_attribute. We also remove pg_statistic entries for it.
1566 * (Everything else needed, such as getting rid of any pg_attrdef entry,
1567 * is handled by dependency.c.)
1568 */
1569 void
RemoveAttributeById(Oid relid,AttrNumber attnum)1570 RemoveAttributeById(Oid relid, AttrNumber attnum)
1571 {
1572 Relation rel;
1573 Relation attr_rel;
1574 HeapTuple tuple;
1575 Form_pg_attribute attStruct;
1576 char newattname[NAMEDATALEN];
1577
1578 /*
1579 * Grab an exclusive lock on the target table, which we will NOT release
1580 * until end of transaction. (In the simple case where we are directly
1581 * dropping this column, AlterTableDropColumn already did this ... but
1582 * when cascading from a drop of some other object, we may not have any
1583 * lock.)
1584 */
1585 rel = relation_open(relid, AccessExclusiveLock);
1586
1587 attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1588
1589 tuple = SearchSysCacheCopy2(ATTNUM,
1590 ObjectIdGetDatum(relid),
1591 Int16GetDatum(attnum));
1592 if (!HeapTupleIsValid(tuple)) /* shouldn't happen */
1593 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1594 attnum, relid);
1595 attStruct = (Form_pg_attribute) GETSTRUCT(tuple);
1596
1597 if (attnum < 0)
1598 {
1599 /* System attribute (probably OID) ... just delete the row */
1600
1601 CatalogTupleDelete(attr_rel, &tuple->t_self);
1602 }
1603 else
1604 {
1605 /* Dropping user attributes is lots harder */
1606
1607 /* Mark the attribute as dropped */
1608 attStruct->attisdropped = true;
1609
1610 /*
1611 * Set the type OID to invalid. A dropped attribute's type link
1612 * cannot be relied on (once the attribute is dropped, the type might
1613 * be too). Fortunately we do not need the type row --- the only
1614 * really essential information is the type's typlen and typalign,
1615 * which are preserved in the attribute's attlen and attalign. We set
1616 * atttypid to zero here as a means of catching code that incorrectly
1617 * expects it to be valid.
1618 */
1619 attStruct->atttypid = InvalidOid;
1620
1621 /* Remove any NOT NULL constraint the column may have */
1622 attStruct->attnotnull = false;
1623
1624 /* We don't want to keep stats for it anymore */
1625 attStruct->attstattarget = 0;
1626
1627 /*
1628 * Change the column name to something that isn't likely to conflict
1629 */
1630 snprintf(newattname, sizeof(newattname),
1631 "........pg.dropped.%d........", attnum);
1632 namestrcpy(&(attStruct->attname), newattname);
1633
1634 /* clear the missing value if any */
1635 if (attStruct->atthasmissing)
1636 {
1637 Datum valuesAtt[Natts_pg_attribute];
1638 bool nullsAtt[Natts_pg_attribute];
1639 bool replacesAtt[Natts_pg_attribute];
1640
1641 /* update the tuple - set atthasmissing and attmissingval */
1642 MemSet(valuesAtt, 0, sizeof(valuesAtt));
1643 MemSet(nullsAtt, false, sizeof(nullsAtt));
1644 MemSet(replacesAtt, false, sizeof(replacesAtt));
1645
1646 valuesAtt[Anum_pg_attribute_atthasmissing - 1] =
1647 BoolGetDatum(false);
1648 replacesAtt[Anum_pg_attribute_atthasmissing - 1] = true;
1649 valuesAtt[Anum_pg_attribute_attmissingval - 1] = (Datum) 0;
1650 nullsAtt[Anum_pg_attribute_attmissingval - 1] = true;
1651 replacesAtt[Anum_pg_attribute_attmissingval - 1] = true;
1652
1653 tuple = heap_modify_tuple(tuple, RelationGetDescr(attr_rel),
1654 valuesAtt, nullsAtt, replacesAtt);
1655 }
1656
1657 CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
1658 }
1659
1660 /*
1661 * Because updating the pg_attribute row will trigger a relcache flush for
1662 * the target relation, we need not do anything else to notify other
1663 * backends of the change.
1664 */
1665
1666 heap_close(attr_rel, RowExclusiveLock);
1667
1668 if (attnum > 0)
1669 RemoveStatistics(relid, attnum);
1670
1671 relation_close(rel, NoLock);
1672 }
1673
1674 /*
1675 * RemoveAttrDefault
1676 *
1677 * If the specified relation/attribute has a default, remove it.
1678 * (If no default, raise error if complain is true, else return quietly.)
1679 */
1680 void
RemoveAttrDefault(Oid relid,AttrNumber attnum,DropBehavior behavior,bool complain,bool internal)1681 RemoveAttrDefault(Oid relid, AttrNumber attnum,
1682 DropBehavior behavior, bool complain, bool internal)
1683 {
1684 Relation attrdef_rel;
1685 ScanKeyData scankeys[2];
1686 SysScanDesc scan;
1687 HeapTuple tuple;
1688 bool found = false;
1689
1690 attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1691
1692 ScanKeyInit(&scankeys[0],
1693 Anum_pg_attrdef_adrelid,
1694 BTEqualStrategyNumber, F_OIDEQ,
1695 ObjectIdGetDatum(relid));
1696 ScanKeyInit(&scankeys[1],
1697 Anum_pg_attrdef_adnum,
1698 BTEqualStrategyNumber, F_INT2EQ,
1699 Int16GetDatum(attnum));
1700
1701 scan = systable_beginscan(attrdef_rel, AttrDefaultIndexId, true,
1702 NULL, 2, scankeys);
1703
1704 /* There should be at most one matching tuple, but we loop anyway */
1705 while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1706 {
1707 ObjectAddress object;
1708
1709 object.classId = AttrDefaultRelationId;
1710 object.objectId = HeapTupleGetOid(tuple);
1711 object.objectSubId = 0;
1712
1713 performDeletion(&object, behavior,
1714 internal ? PERFORM_DELETION_INTERNAL : 0);
1715
1716 found = true;
1717 }
1718
1719 systable_endscan(scan);
1720 heap_close(attrdef_rel, RowExclusiveLock);
1721
1722 if (complain && !found)
1723 elog(ERROR, "could not find attrdef tuple for relation %u attnum %d",
1724 relid, attnum);
1725 }
1726
1727 /*
1728 * RemoveAttrDefaultById
1729 *
1730 * Remove a pg_attrdef entry specified by OID. This is the guts of
1731 * attribute-default removal. Note it should be called via performDeletion,
1732 * not directly.
1733 */
1734 void
RemoveAttrDefaultById(Oid attrdefId)1735 RemoveAttrDefaultById(Oid attrdefId)
1736 {
1737 Relation attrdef_rel;
1738 Relation attr_rel;
1739 Relation myrel;
1740 ScanKeyData scankeys[1];
1741 SysScanDesc scan;
1742 HeapTuple tuple;
1743 Oid myrelid;
1744 AttrNumber myattnum;
1745
1746 /* Grab an appropriate lock on the pg_attrdef relation */
1747 attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1748
1749 /* Find the pg_attrdef tuple */
1750 ScanKeyInit(&scankeys[0],
1751 ObjectIdAttributeNumber,
1752 BTEqualStrategyNumber, F_OIDEQ,
1753 ObjectIdGetDatum(attrdefId));
1754
1755 scan = systable_beginscan(attrdef_rel, AttrDefaultOidIndexId, true,
1756 NULL, 1, scankeys);
1757
1758 tuple = systable_getnext(scan);
1759 if (!HeapTupleIsValid(tuple))
1760 elog(ERROR, "could not find tuple for attrdef %u", attrdefId);
1761
1762 myrelid = ((Form_pg_attrdef) GETSTRUCT(tuple))->adrelid;
1763 myattnum = ((Form_pg_attrdef) GETSTRUCT(tuple))->adnum;
1764
1765 /* Get an exclusive lock on the relation owning the attribute */
1766 myrel = relation_open(myrelid, AccessExclusiveLock);
1767
1768 /* Now we can delete the pg_attrdef row */
1769 CatalogTupleDelete(attrdef_rel, &tuple->t_self);
1770
1771 systable_endscan(scan);
1772 heap_close(attrdef_rel, RowExclusiveLock);
1773
1774 /* Fix the pg_attribute row */
1775 attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1776
1777 tuple = SearchSysCacheCopy2(ATTNUM,
1778 ObjectIdGetDatum(myrelid),
1779 Int16GetDatum(myattnum));
1780 if (!HeapTupleIsValid(tuple)) /* shouldn't happen */
1781 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1782 myattnum, myrelid);
1783
1784 ((Form_pg_attribute) GETSTRUCT(tuple))->atthasdef = false;
1785
1786 CatalogTupleUpdate(attr_rel, &tuple->t_self, tuple);
1787
1788 /*
1789 * Our update of the pg_attribute row will force a relcache rebuild, so
1790 * there's nothing else to do here.
1791 */
1792 heap_close(attr_rel, RowExclusiveLock);
1793
1794 /* Keep lock on attribute's rel until end of xact */
1795 relation_close(myrel, NoLock);
1796 }
1797
1798 /*
1799 * heap_drop_with_catalog - removes specified relation from catalogs
1800 *
1801 * Note that this routine is not responsible for dropping objects that are
1802 * linked to the pg_class entry via dependencies (for example, indexes and
1803 * constraints). Those are deleted by the dependency-tracing logic in
1804 * dependency.c before control gets here. In general, therefore, this routine
1805 * should never be called directly; go through performDeletion() instead.
1806 */
1807 void
heap_drop_with_catalog(Oid relid)1808 heap_drop_with_catalog(Oid relid)
1809 {
1810 Relation rel;
1811 HeapTuple tuple;
1812 Oid parentOid = InvalidOid,
1813 defaultPartOid = InvalidOid;
1814
1815 /*
1816 * To drop a partition safely, we must grab exclusive lock on its parent,
1817 * because another backend might be about to execute a query on the parent
1818 * table. If it relies on previously cached partition descriptor, then it
1819 * could attempt to access the just-dropped relation as its partition. We
1820 * must therefore take a table lock strong enough to prevent all queries
1821 * on the table from proceeding until we commit and send out a
1822 * shared-cache-inval notice that will make them update their partition
1823 * descriptors.
1824 */
1825 tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
1826 if (!HeapTupleIsValid(tuple))
1827 elog(ERROR, "cache lookup failed for relation %u", relid);
1828 if (((Form_pg_class) GETSTRUCT(tuple))->relispartition)
1829 {
1830 parentOid = get_partition_parent(relid);
1831 LockRelationOid(parentOid, AccessExclusiveLock);
1832
1833 /*
1834 * If this is not the default partition, dropping it will change the
1835 * default partition's partition constraint, so we must lock it.
1836 */
1837 defaultPartOid = get_default_partition_oid(parentOid);
1838 if (OidIsValid(defaultPartOid) && relid != defaultPartOid)
1839 LockRelationOid(defaultPartOid, AccessExclusiveLock);
1840 }
1841
1842 ReleaseSysCache(tuple);
1843
1844 /*
1845 * Open and lock the relation.
1846 */
1847 rel = relation_open(relid, AccessExclusiveLock);
1848
1849 /*
1850 * There can no longer be anyone *else* touching the relation, but we
1851 * might still have open queries or cursors, or pending trigger events, in
1852 * our own session.
1853 */
1854 CheckTableNotInUse(rel, "DROP TABLE");
1855
1856 /*
1857 * This effectively deletes all rows in the table, and may be done in a
1858 * serializable transaction. In that case we must record a rw-conflict in
1859 * to this transaction from each transaction holding a predicate lock on
1860 * the table.
1861 */
1862 CheckTableForSerializableConflictIn(rel);
1863
1864 /*
1865 * Delete pg_foreign_table tuple first.
1866 */
1867 if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
1868 {
1869 Relation rel;
1870 HeapTuple tuple;
1871
1872 rel = heap_open(ForeignTableRelationId, RowExclusiveLock);
1873
1874 tuple = SearchSysCache1(FOREIGNTABLEREL, ObjectIdGetDatum(relid));
1875 if (!HeapTupleIsValid(tuple))
1876 elog(ERROR, "cache lookup failed for foreign table %u", relid);
1877
1878 CatalogTupleDelete(rel, &tuple->t_self);
1879
1880 ReleaseSysCache(tuple);
1881 heap_close(rel, RowExclusiveLock);
1882 }
1883
1884 /*
1885 * If a partitioned table, delete the pg_partitioned_table tuple.
1886 */
1887 if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
1888 RemovePartitionKeyByRelId(relid);
1889
1890 /*
1891 * If the relation being dropped is the default partition itself,
1892 * invalidate its entry in pg_partitioned_table.
1893 */
1894 if (relid == defaultPartOid)
1895 update_default_partition_oid(parentOid, InvalidOid);
1896
1897 /*
1898 * Schedule unlinking of the relation's physical files at commit.
1899 */
1900 if (rel->rd_rel->relkind != RELKIND_VIEW &&
1901 rel->rd_rel->relkind != RELKIND_COMPOSITE_TYPE &&
1902 rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
1903 rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
1904 {
1905 RelationDropStorage(rel);
1906 }
1907
1908 /*
1909 * Close relcache entry, but *keep* AccessExclusiveLock on the relation
1910 * until transaction commit. This ensures no one else will try to do
1911 * something with the doomed relation.
1912 */
1913 relation_close(rel, NoLock);
1914
1915 /*
1916 * Remove any associated relation synchronization states.
1917 */
1918 RemoveSubscriptionRel(InvalidOid, relid);
1919
1920 /*
1921 * Forget any ON COMMIT action for the rel
1922 */
1923 remove_on_commit_action(relid);
1924
1925 /*
1926 * Flush the relation from the relcache. We want to do this before
1927 * starting to remove catalog entries, just to be certain that no relcache
1928 * entry rebuild will happen partway through. (That should not really
1929 * matter, since we don't do CommandCounterIncrement here, but let's be
1930 * safe.)
1931 */
1932 RelationForgetRelation(relid);
1933
1934 /*
1935 * remove inheritance information
1936 */
1937 RelationRemoveInheritance(relid);
1938
1939 /*
1940 * delete statistics
1941 */
1942 RemoveStatistics(relid, 0);
1943
1944 /*
1945 * delete attribute tuples
1946 */
1947 DeleteAttributeTuples(relid);
1948
1949 /*
1950 * delete relation tuple
1951 */
1952 DeleteRelationTuple(relid);
1953
1954 if (OidIsValid(parentOid))
1955 {
1956 /*
1957 * If this is not the default partition, the partition constraint of
1958 * the default partition has changed to include the portion of the key
1959 * space previously covered by the dropped partition.
1960 */
1961 if (OidIsValid(defaultPartOid) && relid != defaultPartOid)
1962 CacheInvalidateRelcacheByRelid(defaultPartOid);
1963
1964 /*
1965 * Invalidate the parent's relcache so that the partition is no longer
1966 * included in its partition descriptor.
1967 */
1968 CacheInvalidateRelcacheByRelid(parentOid);
1969 /* keep the lock */
1970 }
1971 }
1972
1973
1974 /*
1975 * RelationClearMissing
1976 *
1977 * Set atthasmissing and attmissingval to false/null for all attributes
1978 * where they are currently set. This can be safely and usefully done if
1979 * the table is rewritten (e.g. by VACUUM FULL or CLUSTER) where we know there
1980 * are no rows left with less than a full complement of attributes.
1981 *
1982 * The caller must have an AccessExclusive lock on the relation.
1983 */
1984 void
RelationClearMissing(Relation rel)1985 RelationClearMissing(Relation rel)
1986 {
1987 Relation attr_rel;
1988 Oid relid = RelationGetRelid(rel);
1989 int natts = RelationGetNumberOfAttributes(rel);
1990 int attnum;
1991 Datum repl_val[Natts_pg_attribute];
1992 bool repl_null[Natts_pg_attribute];
1993 bool repl_repl[Natts_pg_attribute];
1994 Form_pg_attribute attrtuple;
1995 HeapTuple tuple,
1996 newtuple;
1997
1998 memset(repl_val, 0, sizeof(repl_val));
1999 memset(repl_null, false, sizeof(repl_null));
2000 memset(repl_repl, false, sizeof(repl_repl));
2001
2002 repl_val[Anum_pg_attribute_atthasmissing - 1] = BoolGetDatum(false);
2003 repl_null[Anum_pg_attribute_attmissingval - 1] = true;
2004
2005 repl_repl[Anum_pg_attribute_atthasmissing - 1] = true;
2006 repl_repl[Anum_pg_attribute_attmissingval - 1] = true;
2007
2008
2009 /* Get a lock on pg_attribute */
2010 attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
2011
2012 /* process each non-system attribute, including any dropped columns */
2013 for (attnum = 1; attnum <= natts; attnum++)
2014 {
2015 tuple = SearchSysCache2(ATTNUM,
2016 ObjectIdGetDatum(relid),
2017 Int16GetDatum(attnum));
2018 if (!HeapTupleIsValid(tuple)) /* shouldn't happen */
2019 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
2020 attnum, relid);
2021
2022 attrtuple = (Form_pg_attribute) GETSTRUCT(tuple);
2023
2024 /* ignore any where atthasmissing is not true */
2025 if (attrtuple->atthasmissing)
2026 {
2027 newtuple = heap_modify_tuple(tuple, RelationGetDescr(attr_rel),
2028 repl_val, repl_null, repl_repl);
2029
2030 CatalogTupleUpdate(attr_rel, &newtuple->t_self, newtuple);
2031
2032 heap_freetuple(newtuple);
2033 }
2034
2035 ReleaseSysCache(tuple);
2036 }
2037
2038 /*
2039 * Our update of the pg_attribute rows will force a relcache rebuild, so
2040 * there's nothing else to do here.
2041 */
2042 heap_close(attr_rel, RowExclusiveLock);
2043 }
2044
2045 /*
2046 * SetAttrMissing
2047 *
2048 * Set the missing value of a single attribute. This should only be used by
2049 * binary upgrade. Takes an AccessExclusive lock on the relation owning the
2050 * attribute.
2051 */
2052 void
SetAttrMissing(Oid relid,char * attname,char * value)2053 SetAttrMissing(Oid relid, char *attname, char *value)
2054 {
2055 Datum valuesAtt[Natts_pg_attribute];
2056 bool nullsAtt[Natts_pg_attribute];
2057 bool replacesAtt[Natts_pg_attribute];
2058 Datum missingval;
2059 Form_pg_attribute attStruct;
2060 Relation attrrel,
2061 tablerel;
2062 HeapTuple atttup,
2063 newtup;
2064
2065 /* lock the table the attribute belongs to */
2066 tablerel = heap_open(relid, AccessExclusiveLock);
2067
2068 /* Don't do anything unless it's a plain table */
2069 if (tablerel->rd_rel->relkind != RELKIND_RELATION)
2070 {
2071 heap_close(tablerel, AccessExclusiveLock);
2072 return;
2073 }
2074
2075 /* Lock the attribute row and get the data */
2076 attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
2077 atttup = SearchSysCacheAttName(relid, attname);
2078 if (!HeapTupleIsValid(atttup))
2079 elog(ERROR, "cache lookup failed for attribute %s of relation %u",
2080 attname, relid);
2081 attStruct = (Form_pg_attribute) GETSTRUCT(atttup);
2082
2083 /* get an array value from the value string */
2084 missingval = OidFunctionCall3(F_ARRAY_IN,
2085 CStringGetDatum(value),
2086 ObjectIdGetDatum(attStruct->atttypid),
2087 Int32GetDatum(attStruct->atttypmod));
2088
2089 /* update the tuple - set atthasmissing and attmissingval */
2090 MemSet(valuesAtt, 0, sizeof(valuesAtt));
2091 MemSet(nullsAtt, false, sizeof(nullsAtt));
2092 MemSet(replacesAtt, false, sizeof(replacesAtt));
2093
2094 valuesAtt[Anum_pg_attribute_atthasmissing - 1] = BoolGetDatum(true);
2095 replacesAtt[Anum_pg_attribute_atthasmissing - 1] = true;
2096 valuesAtt[Anum_pg_attribute_attmissingval - 1] = missingval;
2097 replacesAtt[Anum_pg_attribute_attmissingval - 1] = true;
2098
2099 newtup = heap_modify_tuple(atttup, RelationGetDescr(attrrel),
2100 valuesAtt, nullsAtt, replacesAtt);
2101 CatalogTupleUpdate(attrrel, &newtup->t_self, newtup);
2102
2103 /* clean up */
2104 ReleaseSysCache(atttup);
2105 heap_close(attrrel, RowExclusiveLock);
2106 heap_close(tablerel, AccessExclusiveLock);
2107 }
2108
2109 /*
2110 * Store a default expression for column attnum of relation rel.
2111 *
2112 * Returns the OID of the new pg_attrdef tuple.
2113 *
2114 * add_column_mode must be true if we are storing the default for a new
2115 * attribute, and false if it's for an already existing attribute. The reason
2116 * for this is that the missing value must never be updated after it is set,
2117 * which can only be when a column is added to the table. Otherwise we would
2118 * in effect be changing existing tuples.
2119 */
2120 Oid
StoreAttrDefault(Relation rel,AttrNumber attnum,Node * expr,bool is_internal,bool add_column_mode)2121 StoreAttrDefault(Relation rel, AttrNumber attnum,
2122 Node *expr, bool is_internal, bool add_column_mode)
2123 {
2124 char *adbin;
2125 char *adsrc;
2126 Relation adrel;
2127 HeapTuple tuple;
2128 Datum values[4];
2129 static bool nulls[4] = {false, false, false, false};
2130 Relation attrrel;
2131 HeapTuple atttup;
2132 Form_pg_attribute attStruct;
2133 Oid attrdefOid;
2134 ObjectAddress colobject,
2135 defobject;
2136
2137 /*
2138 * Flatten expression to string form for storage.
2139 */
2140 adbin = nodeToString(expr);
2141
2142 /*
2143 * Also deparse it to form the mostly-obsolete adsrc field.
2144 */
2145 adsrc = deparse_expression(expr,
2146 deparse_context_for(RelationGetRelationName(rel),
2147 RelationGetRelid(rel)),
2148 false, false);
2149
2150 /*
2151 * Make the pg_attrdef entry.
2152 */
2153 values[Anum_pg_attrdef_adrelid - 1] = RelationGetRelid(rel);
2154 values[Anum_pg_attrdef_adnum - 1] = attnum;
2155 values[Anum_pg_attrdef_adbin - 1] = CStringGetTextDatum(adbin);
2156 values[Anum_pg_attrdef_adsrc - 1] = CStringGetTextDatum(adsrc);
2157
2158 adrel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
2159
2160 tuple = heap_form_tuple(adrel->rd_att, values, nulls);
2161 attrdefOid = CatalogTupleInsert(adrel, tuple);
2162
2163 defobject.classId = AttrDefaultRelationId;
2164 defobject.objectId = attrdefOid;
2165 defobject.objectSubId = 0;
2166
2167 heap_close(adrel, RowExclusiveLock);
2168
2169 /* now can free some of the stuff allocated above */
2170 pfree(DatumGetPointer(values[Anum_pg_attrdef_adbin - 1]));
2171 pfree(DatumGetPointer(values[Anum_pg_attrdef_adsrc - 1]));
2172 heap_freetuple(tuple);
2173 pfree(adbin);
2174 pfree(adsrc);
2175
2176 /*
2177 * Update the pg_attribute entry for the column to show that a default
2178 * exists.
2179 */
2180 attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
2181 atttup = SearchSysCacheCopy2(ATTNUM,
2182 ObjectIdGetDatum(RelationGetRelid(rel)),
2183 Int16GetDatum(attnum));
2184 if (!HeapTupleIsValid(atttup))
2185 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
2186 attnum, RelationGetRelid(rel));
2187 attStruct = (Form_pg_attribute) GETSTRUCT(atttup);
2188 if (!attStruct->atthasdef)
2189 {
2190 Form_pg_attribute defAttStruct;
2191
2192 ExprState *exprState;
2193 Expr *expr2 = (Expr *) expr;
2194 EState *estate = NULL;
2195 ExprContext *econtext;
2196 Datum valuesAtt[Natts_pg_attribute];
2197 bool nullsAtt[Natts_pg_attribute];
2198 bool replacesAtt[Natts_pg_attribute];
2199 Datum missingval = (Datum) 0;
2200 bool missingIsNull = true;
2201
2202 MemSet(valuesAtt, 0, sizeof(valuesAtt));
2203 MemSet(nullsAtt, false, sizeof(nullsAtt));
2204 MemSet(replacesAtt, false, sizeof(replacesAtt));
2205 valuesAtt[Anum_pg_attribute_atthasdef - 1] = true;
2206 replacesAtt[Anum_pg_attribute_atthasdef - 1] = true;
2207
2208 if (rel->rd_rel->relkind == RELKIND_RELATION && add_column_mode)
2209 {
2210 expr2 = expression_planner(expr2);
2211 estate = CreateExecutorState();
2212 exprState = ExecPrepareExpr(expr2, estate);
2213 econtext = GetPerTupleExprContext(estate);
2214
2215 missingval = ExecEvalExpr(exprState, econtext,
2216 &missingIsNull);
2217
2218 FreeExecutorState(estate);
2219
2220 defAttStruct = TupleDescAttr(rel->rd_att, attnum - 1);
2221
2222 if (missingIsNull)
2223 {
2224 /* if the default evaluates to NULL, just store a NULL array */
2225 missingval = (Datum) 0;
2226 }
2227 else
2228 {
2229 /* otherwise make a one-element array of the value */
2230 missingval = PointerGetDatum(
2231 construct_array(&missingval,
2232 1,
2233 defAttStruct->atttypid,
2234 defAttStruct->attlen,
2235 defAttStruct->attbyval,
2236 defAttStruct->attalign));
2237 }
2238
2239 valuesAtt[Anum_pg_attribute_atthasmissing - 1] = !missingIsNull;
2240 replacesAtt[Anum_pg_attribute_atthasmissing - 1] = true;
2241 valuesAtt[Anum_pg_attribute_attmissingval - 1] = missingval;
2242 replacesAtt[Anum_pg_attribute_attmissingval - 1] = true;
2243 nullsAtt[Anum_pg_attribute_attmissingval - 1] = missingIsNull;
2244 }
2245 atttup = heap_modify_tuple(atttup, RelationGetDescr(attrrel),
2246 valuesAtt, nullsAtt, replacesAtt);
2247
2248 CatalogTupleUpdate(attrrel, &atttup->t_self, atttup);
2249
2250 if (!missingIsNull)
2251 pfree(DatumGetPointer(missingval));
2252
2253 }
2254 heap_close(attrrel, RowExclusiveLock);
2255 heap_freetuple(atttup);
2256
2257 /*
2258 * Make a dependency so that the pg_attrdef entry goes away if the column
2259 * (or whole table) is deleted.
2260 */
2261 colobject.classId = RelationRelationId;
2262 colobject.objectId = RelationGetRelid(rel);
2263 colobject.objectSubId = attnum;
2264
2265 recordDependencyOn(&defobject, &colobject, DEPENDENCY_AUTO);
2266
2267 /*
2268 * Record dependencies on objects used in the expression, too.
2269 */
2270 recordDependencyOnExpr(&defobject, expr, NIL, DEPENDENCY_NORMAL);
2271
2272 /*
2273 * Post creation hook for attribute defaults.
2274 *
2275 * XXX. ALTER TABLE ALTER COLUMN SET/DROP DEFAULT is implemented with a
2276 * couple of deletion/creation of the attribute's default entry, so the
2277 * callee should check existence of an older version of this entry if it
2278 * needs to distinguish.
2279 */
2280 InvokeObjectPostCreateHookArg(AttrDefaultRelationId,
2281 RelationGetRelid(rel), attnum, is_internal);
2282
2283 return attrdefOid;
2284 }
2285
2286 /*
2287 * Store a check-constraint expression for the given relation.
2288 *
2289 * Caller is responsible for updating the count of constraints
2290 * in the pg_class entry for the relation.
2291 *
2292 * The OID of the new constraint is returned.
2293 */
2294 static Oid
StoreRelCheck(Relation rel,const char * ccname,Node * expr,bool is_validated,bool is_local,int inhcount,bool is_no_inherit,bool is_internal)2295 StoreRelCheck(Relation rel, const char *ccname, Node *expr,
2296 bool is_validated, bool is_local, int inhcount,
2297 bool is_no_inherit, bool is_internal)
2298 {
2299 char *ccbin;
2300 char *ccsrc;
2301 List *varList;
2302 int keycount;
2303 int16 *attNos;
2304 Oid constrOid;
2305
2306 /*
2307 * Flatten expression to string form for storage.
2308 */
2309 ccbin = nodeToString(expr);
2310
2311 /*
2312 * Also deparse it to form the mostly-obsolete consrc field.
2313 */
2314 ccsrc = deparse_expression(expr,
2315 deparse_context_for(RelationGetRelationName(rel),
2316 RelationGetRelid(rel)),
2317 false, false);
2318
2319 /*
2320 * Find columns of rel that are used in expr
2321 *
2322 * NB: pull_var_clause is okay here only because we don't allow subselects
2323 * in check constraints; it would fail to examine the contents of
2324 * subselects.
2325 */
2326 varList = pull_var_clause(expr, 0);
2327 keycount = list_length(varList);
2328
2329 if (keycount > 0)
2330 {
2331 ListCell *vl;
2332 int i = 0;
2333
2334 attNos = (int16 *) palloc(keycount * sizeof(int16));
2335 foreach(vl, varList)
2336 {
2337 Var *var = (Var *) lfirst(vl);
2338 int j;
2339
2340 for (j = 0; j < i; j++)
2341 if (attNos[j] == var->varattno)
2342 break;
2343 if (j == i)
2344 attNos[i++] = var->varattno;
2345 }
2346 keycount = i;
2347 }
2348 else
2349 attNos = NULL;
2350
2351 /*
2352 * Partitioned tables do not contain any rows themselves, so a NO INHERIT
2353 * constraint makes no sense.
2354 */
2355 if (is_no_inherit &&
2356 rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2357 ereport(ERROR,
2358 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
2359 errmsg("cannot add NO INHERIT constraint to partitioned table \"%s\"",
2360 RelationGetRelationName(rel))));
2361
2362 /*
2363 * Create the Check Constraint
2364 */
2365 constrOid =
2366 CreateConstraintEntry(ccname, /* Constraint Name */
2367 RelationGetNamespace(rel), /* namespace */
2368 CONSTRAINT_CHECK, /* Constraint Type */
2369 false, /* Is Deferrable */
2370 false, /* Is Deferred */
2371 is_validated,
2372 InvalidOid, /* no parent constraint */
2373 RelationGetRelid(rel), /* relation */
2374 attNos, /* attrs in the constraint */
2375 keycount, /* # key attrs in the constraint */
2376 keycount, /* # total attrs in the constraint */
2377 InvalidOid, /* not a domain constraint */
2378 InvalidOid, /* no associated index */
2379 InvalidOid, /* Foreign key fields */
2380 NULL,
2381 NULL,
2382 NULL,
2383 NULL,
2384 0,
2385 ' ',
2386 ' ',
2387 ' ',
2388 NULL, /* not an exclusion constraint */
2389 expr, /* Tree form of check constraint */
2390 ccbin, /* Binary form of check constraint */
2391 ccsrc, /* Source form of check constraint */
2392 is_local, /* conislocal */
2393 inhcount, /* coninhcount */
2394 is_no_inherit, /* connoinherit */
2395 is_internal); /* internally constructed? */
2396
2397 pfree(ccbin);
2398 pfree(ccsrc);
2399
2400 return constrOid;
2401 }
2402
2403 /*
2404 * Store defaults and constraints (passed as a list of CookedConstraint).
2405 *
2406 * Each CookedConstraint struct is modified to store the new catalog tuple OID.
2407 *
2408 * NOTE: only pre-cooked expressions will be passed this way, which is to
2409 * say expressions inherited from an existing relation. Newly parsed
2410 * expressions can be added later, by direct calls to StoreAttrDefault
2411 * and StoreRelCheck (see AddRelationNewConstraints()).
2412 */
2413 static void
StoreConstraints(Relation rel,List * cooked_constraints,bool is_internal)2414 StoreConstraints(Relation rel, List *cooked_constraints, bool is_internal)
2415 {
2416 int numchecks = 0;
2417 ListCell *lc;
2418
2419 if (cooked_constraints == NIL)
2420 return; /* nothing to do */
2421
2422 /*
2423 * Deparsing of constraint expressions will fail unless the just-created
2424 * pg_attribute tuples for this relation are made visible. So, bump the
2425 * command counter. CAUTION: this will cause a relcache entry rebuild.
2426 */
2427 CommandCounterIncrement();
2428
2429 foreach(lc, cooked_constraints)
2430 {
2431 CookedConstraint *con = (CookedConstraint *) lfirst(lc);
2432
2433 switch (con->contype)
2434 {
2435 case CONSTR_DEFAULT:
2436 con->conoid = StoreAttrDefault(rel, con->attnum, con->expr,
2437 is_internal, false);
2438 break;
2439 case CONSTR_CHECK:
2440 con->conoid =
2441 StoreRelCheck(rel, con->name, con->expr,
2442 !con->skip_validation, con->is_local,
2443 con->inhcount, con->is_no_inherit,
2444 is_internal);
2445 numchecks++;
2446 break;
2447 default:
2448 elog(ERROR, "unrecognized constraint type: %d",
2449 (int) con->contype);
2450 }
2451 }
2452
2453 if (numchecks > 0)
2454 SetRelationNumChecks(rel, numchecks);
2455 }
2456
2457 /*
2458 * AddRelationNewConstraints
2459 *
2460 * Add new column default expressions and/or constraint check expressions
2461 * to an existing relation. This is defined to do both for efficiency in
2462 * DefineRelation, but of course you can do just one or the other by passing
2463 * empty lists.
2464 *
2465 * rel: relation to be modified
2466 * newColDefaults: list of RawColumnDefault structures
2467 * newConstraints: list of Constraint nodes
2468 * allow_merge: true if check constraints may be merged with existing ones
2469 * is_local: true if definition is local, false if it's inherited
2470 * is_internal: true if result of some internal process, not a user request
2471 *
2472 * All entries in newColDefaults will be processed. Entries in newConstraints
2473 * will be processed only if they are CONSTR_CHECK type.
2474 *
2475 * Returns a list of CookedConstraint nodes that shows the cooked form of
2476 * the default and constraint expressions added to the relation.
2477 *
2478 * NB: caller should have opened rel with AccessExclusiveLock, and should
2479 * hold that lock till end of transaction. Also, we assume the caller has
2480 * done a CommandCounterIncrement if necessary to make the relation's catalog
2481 * tuples visible.
2482 */
2483 List *
AddRelationNewConstraints(Relation rel,List * newColDefaults,List * newConstraints,bool allow_merge,bool is_local,bool is_internal)2484 AddRelationNewConstraints(Relation rel,
2485 List *newColDefaults,
2486 List *newConstraints,
2487 bool allow_merge,
2488 bool is_local,
2489 bool is_internal)
2490 {
2491 List *cookedConstraints = NIL;
2492 TupleDesc tupleDesc;
2493 TupleConstr *oldconstr;
2494 int numoldchecks;
2495 ParseState *pstate;
2496 RangeTblEntry *rte;
2497 int numchecks;
2498 List *checknames;
2499 ListCell *cell;
2500 Node *expr;
2501 CookedConstraint *cooked;
2502
2503 /*
2504 * Get info about existing constraints.
2505 */
2506 tupleDesc = RelationGetDescr(rel);
2507 oldconstr = tupleDesc->constr;
2508 if (oldconstr)
2509 numoldchecks = oldconstr->num_check;
2510 else
2511 numoldchecks = 0;
2512
2513 /*
2514 * Create a dummy ParseState and insert the target relation as its sole
2515 * rangetable entry. We need a ParseState for transformExpr.
2516 */
2517 pstate = make_parsestate(NULL);
2518 rte = addRangeTableEntryForRelation(pstate,
2519 rel,
2520 NULL,
2521 false,
2522 true);
2523 addRTEtoQuery(pstate, rte, true, true, true);
2524
2525 /*
2526 * Process column default expressions.
2527 */
2528 foreach(cell, newColDefaults)
2529 {
2530 RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell);
2531 Form_pg_attribute atp = TupleDescAttr(rel->rd_att, colDef->attnum - 1);
2532 Oid defOid;
2533
2534 expr = cookDefault(pstate, colDef->raw_default,
2535 atp->atttypid, atp->atttypmod,
2536 NameStr(atp->attname));
2537
2538 /*
2539 * If the expression is just a NULL constant, we do not bother to make
2540 * an explicit pg_attrdef entry, since the default behavior is
2541 * equivalent.
2542 *
2543 * Note a nonobvious property of this test: if the column is of a
2544 * domain type, what we'll get is not a bare null Const but a
2545 * CoerceToDomain expr, so we will not discard the default. This is
2546 * critical because the column default needs to be retained to
2547 * override any default that the domain might have.
2548 */
2549 if (expr == NULL ||
2550 (IsA(expr, Const) &&((Const *) expr)->constisnull))
2551 continue;
2552
2553 /* If the DEFAULT is volatile we cannot use a missing value */
2554 if (colDef->missingMode && contain_volatile_functions((Node *) expr))
2555 colDef->missingMode = false;
2556
2557 defOid = StoreAttrDefault(rel, colDef->attnum, expr, is_internal,
2558 colDef->missingMode);
2559
2560 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
2561 cooked->contype = CONSTR_DEFAULT;
2562 cooked->conoid = defOid;
2563 cooked->name = NULL;
2564 cooked->attnum = colDef->attnum;
2565 cooked->expr = expr;
2566 cooked->skip_validation = false;
2567 cooked->is_local = is_local;
2568 cooked->inhcount = is_local ? 0 : 1;
2569 cooked->is_no_inherit = false;
2570 cookedConstraints = lappend(cookedConstraints, cooked);
2571 }
2572
2573 /*
2574 * Process constraint expressions.
2575 */
2576 numchecks = numoldchecks;
2577 checknames = NIL;
2578 foreach(cell, newConstraints)
2579 {
2580 Constraint *cdef = (Constraint *) lfirst(cell);
2581 char *ccname;
2582 Oid constrOid;
2583
2584 if (cdef->contype != CONSTR_CHECK)
2585 continue;
2586
2587 if (cdef->raw_expr != NULL)
2588 {
2589 Assert(cdef->cooked_expr == NULL);
2590
2591 /*
2592 * Transform raw parsetree to executable expression, and verify
2593 * it's valid as a CHECK constraint.
2594 */
2595 expr = cookConstraint(pstate, cdef->raw_expr,
2596 RelationGetRelationName(rel));
2597 }
2598 else
2599 {
2600 Assert(cdef->cooked_expr != NULL);
2601
2602 /*
2603 * Here, we assume the parser will only pass us valid CHECK
2604 * expressions, so we do no particular checking.
2605 */
2606 expr = stringToNode(cdef->cooked_expr);
2607 }
2608
2609 /*
2610 * Check name uniqueness, or generate a name if none was given.
2611 */
2612 if (cdef->conname != NULL)
2613 {
2614 ListCell *cell2;
2615
2616 ccname = cdef->conname;
2617 /* Check against other new constraints */
2618 /* Needed because we don't do CommandCounterIncrement in loop */
2619 foreach(cell2, checknames)
2620 {
2621 if (strcmp((char *) lfirst(cell2), ccname) == 0)
2622 ereport(ERROR,
2623 (errcode(ERRCODE_DUPLICATE_OBJECT),
2624 errmsg("check constraint \"%s\" already exists",
2625 ccname)));
2626 }
2627
2628 /* save name for future checks */
2629 checknames = lappend(checknames, ccname);
2630
2631 /*
2632 * Check against pre-existing constraints. If we are allowed to
2633 * merge with an existing constraint, there's no more to do here.
2634 * (We omit the duplicate constraint from the result, which is
2635 * what ATAddCheckConstraint wants.)
2636 */
2637 if (MergeWithExistingConstraint(rel, ccname, expr,
2638 allow_merge, is_local,
2639 cdef->initially_valid,
2640 cdef->is_no_inherit))
2641 continue;
2642 }
2643 else
2644 {
2645 /*
2646 * When generating a name, we want to create "tab_col_check" for a
2647 * column constraint and "tab_check" for a table constraint. We
2648 * no longer have any info about the syntactic positioning of the
2649 * constraint phrase, so we approximate this by seeing whether the
2650 * expression references more than one column. (If the user
2651 * played by the rules, the result is the same...)
2652 *
2653 * Note: pull_var_clause() doesn't descend into sublinks, but we
2654 * eliminated those above; and anyway this only needs to be an
2655 * approximate answer.
2656 */
2657 List *vars;
2658 char *colname;
2659
2660 vars = pull_var_clause(expr, 0);
2661
2662 /* eliminate duplicates */
2663 vars = list_union(NIL, vars);
2664
2665 if (list_length(vars) == 1)
2666 colname = get_attname(RelationGetRelid(rel),
2667 ((Var *) linitial(vars))->varattno,
2668 true);
2669 else
2670 colname = NULL;
2671
2672 ccname = ChooseConstraintName(RelationGetRelationName(rel),
2673 colname,
2674 "check",
2675 RelationGetNamespace(rel),
2676 checknames);
2677
2678 /* save name for future checks */
2679 checknames = lappend(checknames, ccname);
2680 }
2681
2682 /*
2683 * OK, store it.
2684 */
2685 constrOid =
2686 StoreRelCheck(rel, ccname, expr, cdef->initially_valid, is_local,
2687 is_local ? 0 : 1, cdef->is_no_inherit, is_internal);
2688
2689 numchecks++;
2690
2691 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
2692 cooked->contype = CONSTR_CHECK;
2693 cooked->conoid = constrOid;
2694 cooked->name = ccname;
2695 cooked->attnum = 0;
2696 cooked->expr = expr;
2697 cooked->skip_validation = cdef->skip_validation;
2698 cooked->is_local = is_local;
2699 cooked->inhcount = is_local ? 0 : 1;
2700 cooked->is_no_inherit = cdef->is_no_inherit;
2701 cookedConstraints = lappend(cookedConstraints, cooked);
2702 }
2703
2704 /*
2705 * Update the count of constraints in the relation's pg_class tuple. We do
2706 * this even if there was no change, in order to ensure that an SI update
2707 * message is sent out for the pg_class tuple, which will force other
2708 * backends to rebuild their relcache entries for the rel. (This is
2709 * critical if we added defaults but not constraints.)
2710 */
2711 SetRelationNumChecks(rel, numchecks);
2712
2713 return cookedConstraints;
2714 }
2715
2716 /*
2717 * Check for a pre-existing check constraint that conflicts with a proposed
2718 * new one, and either adjust its conislocal/coninhcount settings or throw
2719 * error as needed.
2720 *
2721 * Returns true if merged (constraint is a duplicate), or false if it's
2722 * got a so-far-unique name, or throws error if conflict.
2723 *
2724 * XXX See MergeConstraintsIntoExisting too if you change this code.
2725 */
2726 static bool
MergeWithExistingConstraint(Relation rel,const char * ccname,Node * expr,bool allow_merge,bool is_local,bool is_initially_valid,bool is_no_inherit)2727 MergeWithExistingConstraint(Relation rel, const char *ccname, Node *expr,
2728 bool allow_merge, bool is_local,
2729 bool is_initially_valid,
2730 bool is_no_inherit)
2731 {
2732 bool found;
2733 Relation conDesc;
2734 SysScanDesc conscan;
2735 ScanKeyData skey[3];
2736 HeapTuple tup;
2737
2738 /* Search for a pg_constraint entry with same name and relation */
2739 conDesc = heap_open(ConstraintRelationId, RowExclusiveLock);
2740
2741 found = false;
2742
2743 ScanKeyInit(&skey[0],
2744 Anum_pg_constraint_conrelid,
2745 BTEqualStrategyNumber, F_OIDEQ,
2746 ObjectIdGetDatum(RelationGetRelid(rel)));
2747 ScanKeyInit(&skey[1],
2748 Anum_pg_constraint_contypid,
2749 BTEqualStrategyNumber, F_OIDEQ,
2750 ObjectIdGetDatum(InvalidOid));
2751 ScanKeyInit(&skey[2],
2752 Anum_pg_constraint_conname,
2753 BTEqualStrategyNumber, F_NAMEEQ,
2754 CStringGetDatum(ccname));
2755
2756 conscan = systable_beginscan(conDesc, ConstraintRelidTypidNameIndexId, true,
2757 NULL, 3, skey);
2758
2759 /* There can be at most one matching row */
2760 if (HeapTupleIsValid(tup = systable_getnext(conscan)))
2761 {
2762 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tup);
2763
2764 /* Found it. Conflicts if not identical check constraint */
2765 if (con->contype == CONSTRAINT_CHECK)
2766 {
2767 Datum val;
2768 bool isnull;
2769
2770 val = fastgetattr(tup,
2771 Anum_pg_constraint_conbin,
2772 conDesc->rd_att, &isnull);
2773 if (isnull)
2774 elog(ERROR, "null conbin for rel %s",
2775 RelationGetRelationName(rel));
2776 if (equal(expr, stringToNode(TextDatumGetCString(val))))
2777 found = true;
2778 }
2779
2780 /*
2781 * If the existing constraint is purely inherited (no local
2782 * definition) then interpret addition of a local constraint as a
2783 * legal merge. This allows ALTER ADD CONSTRAINT on parent and child
2784 * tables to be given in either order with same end state. However if
2785 * the relation is a partition, all inherited constraints are always
2786 * non-local, including those that were merged.
2787 */
2788 if (is_local && !con->conislocal && !rel->rd_rel->relispartition)
2789 allow_merge = true;
2790
2791 if (!found || !allow_merge)
2792 ereport(ERROR,
2793 (errcode(ERRCODE_DUPLICATE_OBJECT),
2794 errmsg("constraint \"%s\" for relation \"%s\" already exists",
2795 ccname, RelationGetRelationName(rel))));
2796
2797 /* If the child constraint is "no inherit" then cannot merge */
2798 if (con->connoinherit)
2799 ereport(ERROR,
2800 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
2801 errmsg("constraint \"%s\" conflicts with non-inherited constraint on relation \"%s\"",
2802 ccname, RelationGetRelationName(rel))));
2803
2804 /*
2805 * Must not change an existing inherited constraint to "no inherit"
2806 * status. That's because inherited constraints should be able to
2807 * propagate to lower-level children.
2808 */
2809 if (con->coninhcount > 0 && is_no_inherit)
2810 ereport(ERROR,
2811 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
2812 errmsg("constraint \"%s\" conflicts with inherited constraint on relation \"%s\"",
2813 ccname, RelationGetRelationName(rel))));
2814
2815 /*
2816 * If the child constraint is "not valid" then cannot merge with a
2817 * valid parent constraint.
2818 */
2819 if (is_initially_valid && !con->convalidated)
2820 ereport(ERROR,
2821 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
2822 errmsg("constraint \"%s\" conflicts with NOT VALID constraint on relation \"%s\"",
2823 ccname, RelationGetRelationName(rel))));
2824
2825 /* OK to update the tuple */
2826 ereport(NOTICE,
2827 (errmsg("merging constraint \"%s\" with inherited definition",
2828 ccname)));
2829
2830 tup = heap_copytuple(tup);
2831 con = (Form_pg_constraint) GETSTRUCT(tup);
2832
2833 /*
2834 * In case of partitions, an inherited constraint must be inherited
2835 * only once since it cannot have multiple parents and it is never
2836 * considered local.
2837 */
2838 if (rel->rd_rel->relispartition)
2839 {
2840 con->coninhcount = 1;
2841 con->conislocal = false;
2842 }
2843 else
2844 {
2845 if (is_local)
2846 con->conislocal = true;
2847 else
2848 con->coninhcount++;
2849 }
2850
2851 if (is_no_inherit)
2852 {
2853 Assert(is_local);
2854 con->connoinherit = true;
2855 }
2856
2857 CatalogTupleUpdate(conDesc, &tup->t_self, tup);
2858 }
2859
2860 systable_endscan(conscan);
2861 heap_close(conDesc, RowExclusiveLock);
2862
2863 return found;
2864 }
2865
2866 /*
2867 * Update the count of constraints in the relation's pg_class tuple.
2868 *
2869 * Caller had better hold exclusive lock on the relation.
2870 *
2871 * An important side effect is that a SI update message will be sent out for
2872 * the pg_class tuple, which will force other backends to rebuild their
2873 * relcache entries for the rel. Also, this backend will rebuild its
2874 * own relcache entry at the next CommandCounterIncrement.
2875 */
2876 static void
SetRelationNumChecks(Relation rel,int numchecks)2877 SetRelationNumChecks(Relation rel, int numchecks)
2878 {
2879 Relation relrel;
2880 HeapTuple reltup;
2881 Form_pg_class relStruct;
2882
2883 relrel = heap_open(RelationRelationId, RowExclusiveLock);
2884 reltup = SearchSysCacheCopy1(RELOID,
2885 ObjectIdGetDatum(RelationGetRelid(rel)));
2886 if (!HeapTupleIsValid(reltup))
2887 elog(ERROR, "cache lookup failed for relation %u",
2888 RelationGetRelid(rel));
2889 relStruct = (Form_pg_class) GETSTRUCT(reltup);
2890
2891 if (relStruct->relchecks != numchecks)
2892 {
2893 relStruct->relchecks = numchecks;
2894
2895 CatalogTupleUpdate(relrel, &reltup->t_self, reltup);
2896 }
2897 else
2898 {
2899 /* Skip the disk update, but force relcache inval anyway */
2900 CacheInvalidateRelcache(rel);
2901 }
2902
2903 heap_freetuple(reltup);
2904 heap_close(relrel, RowExclusiveLock);
2905 }
2906
2907 /*
2908 * Take a raw default and convert it to a cooked format ready for
2909 * storage.
2910 *
2911 * Parse state should be set up to recognize any vars that might appear
2912 * in the expression. (Even though we plan to reject vars, it's more
2913 * user-friendly to give the correct error message than "unknown var".)
2914 *
2915 * If atttypid is not InvalidOid, coerce the expression to the specified
2916 * type (and typmod atttypmod). attname is only needed in this case:
2917 * it is used in the error message, if any.
2918 */
2919 Node *
cookDefault(ParseState * pstate,Node * raw_default,Oid atttypid,int32 atttypmod,const char * attname)2920 cookDefault(ParseState *pstate,
2921 Node *raw_default,
2922 Oid atttypid,
2923 int32 atttypmod,
2924 const char *attname)
2925 {
2926 Node *expr;
2927
2928 Assert(raw_default != NULL);
2929
2930 /*
2931 * Transform raw parsetree to executable expression.
2932 */
2933 expr = transformExpr(pstate, raw_default, EXPR_KIND_COLUMN_DEFAULT);
2934
2935 /*
2936 * Make sure default expr does not refer to any vars (we need this check
2937 * since the pstate includes the target table).
2938 */
2939 if (contain_var_clause(expr))
2940 ereport(ERROR,
2941 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
2942 errmsg("cannot use column references in default expression")));
2943
2944 /*
2945 * transformExpr() should have already rejected subqueries, aggregates,
2946 * window functions, and SRFs, based on the EXPR_KIND_ for a default
2947 * expression.
2948 */
2949
2950 /*
2951 * Coerce the expression to the correct type and typmod, if given. This
2952 * should match the parser's processing of non-defaulted expressions ---
2953 * see transformAssignedExpr().
2954 */
2955 if (OidIsValid(atttypid))
2956 {
2957 Oid type_id = exprType(expr);
2958
2959 expr = coerce_to_target_type(pstate, expr, type_id,
2960 atttypid, atttypmod,
2961 COERCION_ASSIGNMENT,
2962 COERCE_IMPLICIT_CAST,
2963 -1);
2964 if (expr == NULL)
2965 ereport(ERROR,
2966 (errcode(ERRCODE_DATATYPE_MISMATCH),
2967 errmsg("column \"%s\" is of type %s"
2968 " but default expression is of type %s",
2969 attname,
2970 format_type_be(atttypid),
2971 format_type_be(type_id)),
2972 errhint("You will need to rewrite or cast the expression.")));
2973 }
2974
2975 /*
2976 * Finally, take care of collations in the finished expression.
2977 */
2978 assign_expr_collations(pstate, expr);
2979
2980 return expr;
2981 }
2982
2983 /*
2984 * Take a raw CHECK constraint expression and convert it to a cooked format
2985 * ready for storage.
2986 *
2987 * Parse state must be set up to recognize any vars that might appear
2988 * in the expression.
2989 */
2990 static Node *
cookConstraint(ParseState * pstate,Node * raw_constraint,char * relname)2991 cookConstraint(ParseState *pstate,
2992 Node *raw_constraint,
2993 char *relname)
2994 {
2995 Node *expr;
2996
2997 /*
2998 * Transform raw parsetree to executable expression.
2999 */
3000 expr = transformExpr(pstate, raw_constraint, EXPR_KIND_CHECK_CONSTRAINT);
3001
3002 /*
3003 * Make sure it yields a boolean result.
3004 */
3005 expr = coerce_to_boolean(pstate, expr, "CHECK");
3006
3007 /*
3008 * Take care of collations.
3009 */
3010 assign_expr_collations(pstate, expr);
3011
3012 /*
3013 * Make sure no outside relations are referred to (this is probably dead
3014 * code now that add_missing_from is history).
3015 */
3016 if (list_length(pstate->p_rtable) != 1)
3017 ereport(ERROR,
3018 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
3019 errmsg("only table \"%s\" can be referenced in check constraint",
3020 relname)));
3021
3022 return expr;
3023 }
3024
3025
3026 /*
3027 * RemoveStatistics --- remove entries in pg_statistic for a rel or column
3028 *
3029 * If attnum is zero, remove all entries for rel; else remove only the one(s)
3030 * for that column.
3031 */
3032 void
RemoveStatistics(Oid relid,AttrNumber attnum)3033 RemoveStatistics(Oid relid, AttrNumber attnum)
3034 {
3035 Relation pgstatistic;
3036 SysScanDesc scan;
3037 ScanKeyData key[2];
3038 int nkeys;
3039 HeapTuple tuple;
3040
3041 pgstatistic = heap_open(StatisticRelationId, RowExclusiveLock);
3042
3043 ScanKeyInit(&key[0],
3044 Anum_pg_statistic_starelid,
3045 BTEqualStrategyNumber, F_OIDEQ,
3046 ObjectIdGetDatum(relid));
3047
3048 if (attnum == 0)
3049 nkeys = 1;
3050 else
3051 {
3052 ScanKeyInit(&key[1],
3053 Anum_pg_statistic_staattnum,
3054 BTEqualStrategyNumber, F_INT2EQ,
3055 Int16GetDatum(attnum));
3056 nkeys = 2;
3057 }
3058
3059 scan = systable_beginscan(pgstatistic, StatisticRelidAttnumInhIndexId, true,
3060 NULL, nkeys, key);
3061
3062 /* we must loop even when attnum != 0, in case of inherited stats */
3063 while (HeapTupleIsValid(tuple = systable_getnext(scan)))
3064 CatalogTupleDelete(pgstatistic, &tuple->t_self);
3065
3066 systable_endscan(scan);
3067
3068 heap_close(pgstatistic, RowExclusiveLock);
3069 }
3070
3071
3072 /*
3073 * RelationTruncateIndexes - truncate all indexes associated
3074 * with the heap relation to zero tuples.
3075 *
3076 * The routine will truncate and then reconstruct the indexes on
3077 * the specified relation. Caller must hold exclusive lock on rel.
3078 */
3079 static void
RelationTruncateIndexes(Relation heapRelation)3080 RelationTruncateIndexes(Relation heapRelation)
3081 {
3082 ListCell *indlist;
3083
3084 /* Ask the relcache to produce a list of the indexes of the rel */
3085 foreach(indlist, RelationGetIndexList(heapRelation))
3086 {
3087 Oid indexId = lfirst_oid(indlist);
3088 Relation currentIndex;
3089 IndexInfo *indexInfo;
3090
3091 /* Open the index relation; use exclusive lock, just to be sure */
3092 currentIndex = index_open(indexId, AccessExclusiveLock);
3093
3094 /*
3095 * Fetch info needed for index_build. Since we know there are no
3096 * tuples that actually need indexing, we can use a dummy IndexInfo.
3097 * This is slightly cheaper to build, but the real point is to avoid
3098 * possibly running user-defined code in index expressions or
3099 * predicates. We might be getting invoked during ON COMMIT
3100 * processing, and we don't want to run any such code then.
3101 */
3102 indexInfo = BuildDummyIndexInfo(currentIndex);
3103
3104 /*
3105 * Now truncate the actual file (and discard buffers).
3106 */
3107 RelationTruncate(currentIndex, 0);
3108
3109 /* Initialize the index and rebuild */
3110 /* Note: we do not need to re-establish pkey setting */
3111 index_build(heapRelation, currentIndex, indexInfo, false, true, false);
3112
3113 /* We're done with this index */
3114 index_close(currentIndex, NoLock);
3115 }
3116 }
3117
3118 /*
3119 * heap_truncate
3120 *
3121 * This routine deletes all data within all the specified relations.
3122 *
3123 * This is not transaction-safe! There is another, transaction-safe
3124 * implementation in commands/tablecmds.c. We now use this only for
3125 * ON COMMIT truncation of temporary tables, where it doesn't matter.
3126 */
3127 void
heap_truncate(List * relids)3128 heap_truncate(List *relids)
3129 {
3130 List *relations = NIL;
3131 ListCell *cell;
3132
3133 /* Open relations for processing, and grab exclusive access on each */
3134 foreach(cell, relids)
3135 {
3136 Oid rid = lfirst_oid(cell);
3137 Relation rel;
3138
3139 rel = heap_open(rid, AccessExclusiveLock);
3140 relations = lappend(relations, rel);
3141 }
3142
3143 /* Don't allow truncate on tables that are referenced by foreign keys */
3144 heap_truncate_check_FKs(relations, true);
3145
3146 /* OK to do it */
3147 foreach(cell, relations)
3148 {
3149 Relation rel = lfirst(cell);
3150
3151 /* Truncate the relation */
3152 heap_truncate_one_rel(rel);
3153
3154 /* Close the relation, but keep exclusive lock on it until commit */
3155 heap_close(rel, NoLock);
3156 }
3157 }
3158
3159 /*
3160 * heap_truncate_one_rel
3161 *
3162 * This routine deletes all data within the specified relation.
3163 *
3164 * This is not transaction-safe, because the truncation is done immediately
3165 * and cannot be rolled back later. Caller is responsible for having
3166 * checked permissions etc, and must have obtained AccessExclusiveLock.
3167 */
3168 void
heap_truncate_one_rel(Relation rel)3169 heap_truncate_one_rel(Relation rel)
3170 {
3171 Oid toastrelid;
3172
3173 /*
3174 * Truncate the relation. Partitioned tables have no storage, so there is
3175 * nothing to do for them here.
3176 */
3177 if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
3178 return;
3179
3180 /* Truncate the actual file (and discard buffers) */
3181 RelationTruncate(rel, 0);
3182
3183 /* If the relation has indexes, truncate the indexes too */
3184 RelationTruncateIndexes(rel);
3185
3186 /* If there is a toast table, truncate that too */
3187 toastrelid = rel->rd_rel->reltoastrelid;
3188 if (OidIsValid(toastrelid))
3189 {
3190 Relation toastrel = heap_open(toastrelid, AccessExclusiveLock);
3191
3192 RelationTruncate(toastrel, 0);
3193 RelationTruncateIndexes(toastrel);
3194 /* keep the lock... */
3195 heap_close(toastrel, NoLock);
3196 }
3197 }
3198
3199 /*
3200 * heap_truncate_check_FKs
3201 * Check for foreign keys referencing a list of relations that
3202 * are to be truncated, and raise error if there are any
3203 *
3204 * We disallow such FKs (except self-referential ones) since the whole point
3205 * of TRUNCATE is to not scan the individual rows to be thrown away.
3206 *
3207 * This is split out so it can be shared by both implementations of truncate.
3208 * Caller should already hold a suitable lock on the relations.
3209 *
3210 * tempTables is only used to select an appropriate error message.
3211 */
3212 void
heap_truncate_check_FKs(List * relations,bool tempTables)3213 heap_truncate_check_FKs(List *relations, bool tempTables)
3214 {
3215 List *oids = NIL;
3216 List *dependents;
3217 ListCell *cell;
3218
3219 /*
3220 * Build a list of OIDs of the interesting relations.
3221 *
3222 * If a relation has no triggers, then it can neither have FKs nor be
3223 * referenced by a FK from another table, so we can ignore it. For
3224 * partitioned tables, FKs have no triggers, so we must include them
3225 * anyway.
3226 */
3227 foreach(cell, relations)
3228 {
3229 Relation rel = lfirst(cell);
3230
3231 if (rel->rd_rel->relhastriggers ||
3232 rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
3233 oids = lappend_oid(oids, RelationGetRelid(rel));
3234 }
3235
3236 /*
3237 * Fast path: if no relation has triggers, none has FKs either.
3238 */
3239 if (oids == NIL)
3240 return;
3241
3242 /*
3243 * Otherwise, must scan pg_constraint. We make one pass with all the
3244 * relations considered; if this finds nothing, then all is well.
3245 */
3246 dependents = heap_truncate_find_FKs(oids);
3247 if (dependents == NIL)
3248 return;
3249
3250 /*
3251 * Otherwise we repeat the scan once per relation to identify a particular
3252 * pair of relations to complain about. This is pretty slow, but
3253 * performance shouldn't matter much in a failure path. The reason for
3254 * doing things this way is to ensure that the message produced is not
3255 * dependent on chance row locations within pg_constraint.
3256 */
3257 foreach(cell, oids)
3258 {
3259 Oid relid = lfirst_oid(cell);
3260 ListCell *cell2;
3261
3262 dependents = heap_truncate_find_FKs(list_make1_oid(relid));
3263
3264 foreach(cell2, dependents)
3265 {
3266 Oid relid2 = lfirst_oid(cell2);
3267
3268 if (!list_member_oid(oids, relid2))
3269 {
3270 char *relname = get_rel_name(relid);
3271 char *relname2 = get_rel_name(relid2);
3272
3273 if (tempTables)
3274 ereport(ERROR,
3275 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
3276 errmsg("unsupported ON COMMIT and foreign key combination"),
3277 errdetail("Table \"%s\" references \"%s\", but they do not have the same ON COMMIT setting.",
3278 relname2, relname)));
3279 else
3280 ereport(ERROR,
3281 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
3282 errmsg("cannot truncate a table referenced in a foreign key constraint"),
3283 errdetail("Table \"%s\" references \"%s\".",
3284 relname2, relname),
3285 errhint("Truncate table \"%s\" at the same time, "
3286 "or use TRUNCATE ... CASCADE.",
3287 relname2)));
3288 }
3289 }
3290 }
3291 }
3292
3293 /*
3294 * heap_truncate_find_FKs
3295 * Find relations having foreign keys referencing any of the given rels
3296 *
3297 * Input and result are both lists of relation OIDs. The result contains
3298 * no duplicates, does *not* include any rels that were already in the input
3299 * list, and is sorted in OID order. (The last property is enforced mainly
3300 * to guarantee consistent behavior in the regression tests; we don't want
3301 * behavior to change depending on chance locations of rows in pg_constraint.)
3302 *
3303 * Note: caller should already have appropriate lock on all rels mentioned
3304 * in relationIds. Since adding or dropping an FK requires exclusive lock
3305 * on both rels, this ensures that the answer will be stable.
3306 */
3307 List *
heap_truncate_find_FKs(List * relationIds)3308 heap_truncate_find_FKs(List *relationIds)
3309 {
3310 List *result = NIL;
3311 Relation fkeyRel;
3312 SysScanDesc fkeyScan;
3313 HeapTuple tuple;
3314
3315 /*
3316 * Must scan pg_constraint. Right now, it is a seqscan because there is
3317 * no available index on confrelid.
3318 */
3319 fkeyRel = heap_open(ConstraintRelationId, AccessShareLock);
3320
3321 fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
3322 NULL, 0, NULL);
3323
3324 while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
3325 {
3326 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
3327
3328 /* Not a foreign key */
3329 if (con->contype != CONSTRAINT_FOREIGN)
3330 continue;
3331
3332 /* Not referencing one of our list of tables */
3333 if (!list_member_oid(relationIds, con->confrelid))
3334 continue;
3335
3336 /* Add referencer unless already in input or result list */
3337 if (!list_member_oid(relationIds, con->conrelid))
3338 result = insert_ordered_unique_oid(result, con->conrelid);
3339 }
3340
3341 systable_endscan(fkeyScan);
3342 heap_close(fkeyRel, AccessShareLock);
3343
3344 return result;
3345 }
3346
3347 /*
3348 * insert_ordered_unique_oid
3349 * Insert a new Oid into a sorted list of Oids, preserving ordering,
3350 * and eliminating duplicates
3351 *
3352 * Building the ordered list this way is O(N^2), but with a pretty small
3353 * constant, so for the number of entries we expect it will probably be
3354 * faster than trying to apply qsort(). It seems unlikely someone would be
3355 * trying to truncate a table with thousands of dependent tables ...
3356 */
3357 static List *
insert_ordered_unique_oid(List * list,Oid datum)3358 insert_ordered_unique_oid(List *list, Oid datum)
3359 {
3360 ListCell *prev;
3361
3362 /* Does the datum belong at the front? */
3363 if (list == NIL || datum < linitial_oid(list))
3364 return lcons_oid(datum, list);
3365 /* Does it match the first entry? */
3366 if (datum == linitial_oid(list))
3367 return list; /* duplicate, so don't insert */
3368 /* No, so find the entry it belongs after */
3369 prev = list_head(list);
3370 for (;;)
3371 {
3372 ListCell *curr = lnext(prev);
3373
3374 if (curr == NULL || datum < lfirst_oid(curr))
3375 break; /* it belongs after 'prev', before 'curr' */
3376
3377 if (datum == lfirst_oid(curr))
3378 return list; /* duplicate, so don't insert */
3379
3380 prev = curr;
3381 }
3382 /* Insert datum into list after 'prev' */
3383 lappend_cell_oid(list, prev, datum);
3384 return list;
3385 }
3386
3387 /*
3388 * StorePartitionKey
3389 * Store information about the partition key rel into the catalog
3390 */
3391 void
StorePartitionKey(Relation rel,char strategy,int16 partnatts,AttrNumber * partattrs,List * partexprs,Oid * partopclass,Oid * partcollation)3392 StorePartitionKey(Relation rel,
3393 char strategy,
3394 int16 partnatts,
3395 AttrNumber *partattrs,
3396 List *partexprs,
3397 Oid *partopclass,
3398 Oid *partcollation)
3399 {
3400 int i;
3401 int2vector *partattrs_vec;
3402 oidvector *partopclass_vec;
3403 oidvector *partcollation_vec;
3404 Datum partexprDatum;
3405 Relation pg_partitioned_table;
3406 HeapTuple tuple;
3407 Datum values[Natts_pg_partitioned_table];
3408 bool nulls[Natts_pg_partitioned_table];
3409 ObjectAddress myself;
3410 ObjectAddress referenced;
3411
3412 Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
3413
3414 /* Copy the partition attribute numbers, opclass OIDs into arrays */
3415 partattrs_vec = buildint2vector(partattrs, partnatts);
3416 partopclass_vec = buildoidvector(partopclass, partnatts);
3417 partcollation_vec = buildoidvector(partcollation, partnatts);
3418
3419 /* Convert the expressions (if any) to a text datum */
3420 if (partexprs)
3421 {
3422 char *exprString;
3423
3424 exprString = nodeToString(partexprs);
3425 partexprDatum = CStringGetTextDatum(exprString);
3426 pfree(exprString);
3427 }
3428 else
3429 partexprDatum = (Datum) 0;
3430
3431 pg_partitioned_table = heap_open(PartitionedRelationId, RowExclusiveLock);
3432
3433 MemSet(nulls, false, sizeof(nulls));
3434
3435 /* Only this can ever be NULL */
3436 if (!partexprDatum)
3437 nulls[Anum_pg_partitioned_table_partexprs - 1] = true;
3438
3439 values[Anum_pg_partitioned_table_partrelid - 1] = ObjectIdGetDatum(RelationGetRelid(rel));
3440 values[Anum_pg_partitioned_table_partstrat - 1] = CharGetDatum(strategy);
3441 values[Anum_pg_partitioned_table_partnatts - 1] = Int16GetDatum(partnatts);
3442 values[Anum_pg_partitioned_table_partdefid - 1] = ObjectIdGetDatum(InvalidOid);
3443 values[Anum_pg_partitioned_table_partattrs - 1] = PointerGetDatum(partattrs_vec);
3444 values[Anum_pg_partitioned_table_partclass - 1] = PointerGetDatum(partopclass_vec);
3445 values[Anum_pg_partitioned_table_partcollation - 1] = PointerGetDatum(partcollation_vec);
3446 values[Anum_pg_partitioned_table_partexprs - 1] = partexprDatum;
3447
3448 tuple = heap_form_tuple(RelationGetDescr(pg_partitioned_table), values, nulls);
3449
3450 CatalogTupleInsert(pg_partitioned_table, tuple);
3451 heap_close(pg_partitioned_table, RowExclusiveLock);
3452
3453 /* Mark this relation as dependent on a few things as follows */
3454 myself.classId = RelationRelationId;
3455 myself.objectId = RelationGetRelid(rel);
3456 myself.objectSubId = 0;
3457
3458 /* Operator class and collation per key column */
3459 for (i = 0; i < partnatts; i++)
3460 {
3461 referenced.classId = OperatorClassRelationId;
3462 referenced.objectId = partopclass[i];
3463 referenced.objectSubId = 0;
3464
3465 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
3466
3467 /* The default collation is pinned, so don't bother recording it */
3468 if (OidIsValid(partcollation[i]) &&
3469 partcollation[i] != DEFAULT_COLLATION_OID)
3470 {
3471 referenced.classId = CollationRelationId;
3472 referenced.objectId = partcollation[i];
3473 referenced.objectSubId = 0;
3474
3475 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
3476 }
3477 }
3478
3479 /*
3480 * The partitioning columns are made internally dependent on the table,
3481 * because we cannot drop any of them without dropping the whole table.
3482 * (ATExecDropColumn independently enforces that, but it's not bulletproof
3483 * so we need the dependencies too.)
3484 */
3485 for (i = 0; i < partnatts; i++)
3486 {
3487 if (partattrs[i] == 0)
3488 continue; /* ignore expressions here */
3489
3490 referenced.classId = RelationRelationId;
3491 referenced.objectId = RelationGetRelid(rel);
3492 referenced.objectSubId = partattrs[i];
3493
3494 recordDependencyOn(&referenced, &myself, DEPENDENCY_INTERNAL);
3495 }
3496
3497 /*
3498 * Also consider anything mentioned in partition expressions. External
3499 * references (e.g. functions) get NORMAL dependencies. Table columns
3500 * mentioned in the expressions are handled the same as plain partitioning
3501 * columns, i.e. they become internally dependent on the whole table.
3502 */
3503 if (partexprs)
3504 recordDependencyOnSingleRelExpr(&myself,
3505 (Node *) partexprs,
3506 RelationGetRelid(rel),
3507 DEPENDENCY_NORMAL,
3508 DEPENDENCY_INTERNAL,
3509 true /* reverse the self-deps */ );
3510
3511 /*
3512 * We must invalidate the relcache so that the next
3513 * CommandCounterIncrement() will cause the same to be rebuilt using the
3514 * information in just created catalog entry.
3515 */
3516 CacheInvalidateRelcache(rel);
3517 }
3518
3519 /*
3520 * RemovePartitionKeyByRelId
3521 * Remove pg_partitioned_table entry for a relation
3522 */
3523 void
RemovePartitionKeyByRelId(Oid relid)3524 RemovePartitionKeyByRelId(Oid relid)
3525 {
3526 Relation rel;
3527 HeapTuple tuple;
3528
3529 rel = heap_open(PartitionedRelationId, RowExclusiveLock);
3530
3531 tuple = SearchSysCache1(PARTRELID, ObjectIdGetDatum(relid));
3532 if (!HeapTupleIsValid(tuple))
3533 elog(ERROR, "cache lookup failed for partition key of relation %u",
3534 relid);
3535
3536 CatalogTupleDelete(rel, &tuple->t_self);
3537
3538 ReleaseSysCache(tuple);
3539 heap_close(rel, RowExclusiveLock);
3540 }
3541
3542 /*
3543 * StorePartitionBound
3544 * Update pg_class tuple of rel to store the partition bound and set
3545 * relispartition to true
3546 *
3547 * If this is the default partition, also update the default partition OID in
3548 * pg_partitioned_table.
3549 *
3550 * Also, invalidate the parent's relcache, so that the next rebuild will load
3551 * the new partition's info into its partition descriptor. If there is a
3552 * default partition, we must invalidate its relcache entry as well.
3553 */
3554 void
StorePartitionBound(Relation rel,Relation parent,PartitionBoundSpec * bound)3555 StorePartitionBound(Relation rel, Relation parent, PartitionBoundSpec *bound)
3556 {
3557 Relation classRel;
3558 HeapTuple tuple,
3559 newtuple;
3560 Datum new_val[Natts_pg_class];
3561 bool new_null[Natts_pg_class],
3562 new_repl[Natts_pg_class];
3563 Oid defaultPartOid;
3564
3565 /* Update pg_class tuple */
3566 classRel = heap_open(RelationRelationId, RowExclusiveLock);
3567 tuple = SearchSysCacheCopy1(RELOID,
3568 ObjectIdGetDatum(RelationGetRelid(rel)));
3569 if (!HeapTupleIsValid(tuple))
3570 elog(ERROR, "cache lookup failed for relation %u",
3571 RelationGetRelid(rel));
3572
3573 #ifdef USE_ASSERT_CHECKING
3574 {
3575 Form_pg_class classForm;
3576 bool isnull;
3577
3578 classForm = (Form_pg_class) GETSTRUCT(tuple);
3579 Assert(!classForm->relispartition);
3580 (void) SysCacheGetAttr(RELOID, tuple, Anum_pg_class_relpartbound,
3581 &isnull);
3582 Assert(isnull);
3583 }
3584 #endif
3585
3586 /* Fill in relpartbound value */
3587 memset(new_val, 0, sizeof(new_val));
3588 memset(new_null, false, sizeof(new_null));
3589 memset(new_repl, false, sizeof(new_repl));
3590 new_val[Anum_pg_class_relpartbound - 1] = CStringGetTextDatum(nodeToString(bound));
3591 new_null[Anum_pg_class_relpartbound - 1] = false;
3592 new_repl[Anum_pg_class_relpartbound - 1] = true;
3593 newtuple = heap_modify_tuple(tuple, RelationGetDescr(classRel),
3594 new_val, new_null, new_repl);
3595 /* Also set the flag */
3596 ((Form_pg_class) GETSTRUCT(newtuple))->relispartition = true;
3597 CatalogTupleUpdate(classRel, &newtuple->t_self, newtuple);
3598 heap_freetuple(newtuple);
3599 heap_close(classRel, RowExclusiveLock);
3600
3601 /*
3602 * If we're storing bounds for the default partition, update
3603 * pg_partitioned_table too.
3604 */
3605 if (bound->is_default)
3606 update_default_partition_oid(RelationGetRelid(parent),
3607 RelationGetRelid(rel));
3608
3609 /* Make these updates visible */
3610 CommandCounterIncrement();
3611
3612 /*
3613 * The partition constraint for the default partition depends on the
3614 * partition bounds of every other partition, so we must invalidate the
3615 * relcache entry for that partition every time a partition is added or
3616 * removed.
3617 */
3618 defaultPartOid = get_default_oid_from_partdesc(RelationGetPartitionDesc(parent));
3619 if (OidIsValid(defaultPartOid))
3620 CacheInvalidateRelcacheByRelid(defaultPartOid);
3621
3622 CacheInvalidateRelcache(parent);
3623 }
3624