LCOV - code coverage report
Current view: top level - src/backend/catalog - heap.c (source / functions) Hit Total Coverage
Test: PostgreSQL 16beta1 Lines: 849 905 93.8 %
Date: 2023-05-30 18:12:27 Functions: 37 38 97.4 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.14