granicus.if.org Git - postgresql/blob - src/backend/catalog/heap.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * heap.c
   4  *        code to create and destroy POSTGRES heap relations
   5  *
   6  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.369 2010/02/03 01:14:16 tgl Exp $
  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/heapam.h"
  34 #include "access/sysattr.h"
  35 #include "access/transam.h"
  36 #include "access/xact.h"
  37 #include "catalog/catalog.h"
  38 #include "catalog/dependency.h"
  39 #include "catalog/heap.h"
  40 #include "catalog/index.h"
  41 #include "catalog/indexing.h"
  42 #include "catalog/pg_attrdef.h"
  43 #include "catalog/pg_constraint.h"
  44 #include "catalog/pg_inherits.h"
  45 #include "catalog/pg_namespace.h"
  46 #include "catalog/pg_statistic.h"
  47 #include "catalog/pg_tablespace.h"
  48 #include "catalog/pg_type.h"
  49 #include "catalog/pg_type_fn.h"
  50 #include "catalog/storage.h"
  51 #include "commands/tablecmds.h"
  52 #include "commands/typecmds.h"
  53 #include "miscadmin.h"
  54 #include "nodes/nodeFuncs.h"
  55 #include "optimizer/var.h"
  56 #include "parser/parse_coerce.h"
  57 #include "parser/parse_expr.h"
  58 #include "parser/parse_relation.h"
  59 #include "storage/bufmgr.h"
  60 #include "storage/freespace.h"
  61 #include "storage/smgr.h"
  62 #include "utils/acl.h"
  63 #include "utils/builtins.h"
  64 #include "utils/fmgroids.h"
  65 #include "utils/inval.h"
  66 #include "utils/lsyscache.h"
  67 #include "utils/relcache.h"
  68 #include "utils/snapmgr.h"
  69 #include "utils/syscache.h"
  70 #include "utils/tqual.h"
  71
  72
  73 /* Kluge for upgrade-in-place support */
  74 Oid binary_upgrade_next_heap_relfilenode = InvalidOid;
  75 Oid binary_upgrade_next_toast_relfilenode = InvalidOid;
  76
  77 static void AddNewRelationTuple(Relation pg_class_desc,
  78                                         Relation new_rel_desc,
  79                                         Oid new_rel_oid,
  80                                         Oid new_type_oid,
  81                                         Oid reloftype,
  82                                         Oid relowner,
  83                                         char relkind,
  84                                         Datum relacl,
  85                                         Datum reloptions);
  86 static Oid AddNewRelationType(const char *typeName,
  87                                    Oid typeNamespace,
  88                                    Oid new_rel_oid,
  89                                    char new_rel_kind,
  90                                    Oid ownerid,
  91                                    Oid new_row_type,
  92                                    Oid new_array_type);
  93 static void RelationRemoveInheritance(Oid relid);
  94 static void StoreRelCheck(Relation rel, char *ccname, Node *expr,
  95                           bool is_local, int inhcount);
  96 static void StoreConstraints(Relation rel, List *cooked_constraints);
  97 static bool MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
  98                                                         bool allow_merge, bool is_local);
  99 static void SetRelationNumChecks(Relation rel, int numchecks);
 100 static Node *cookConstraint(ParseState *pstate,
 101                            Node *raw_constraint,
 102                            char *relname);
 103 static List *insert_ordered_unique_oid(List *list, Oid datum);
 104
 105
 106 /* ----------------------------------------------------------------
 107  *                              XXX UGLY HARD CODED BADNESS FOLLOWS XXX
 108  *
 109  *              these should all be moved to someplace in the lib/catalog
 110  *              module, if not obliterated first.
 111  * ----------------------------------------------------------------
 112  */
 113
 114
 115 /*
 116  * Note:
 117  *              Should the system special case these attributes in the future?
 118  *              Advantage:      consume much less space in the ATTRIBUTE relation.
 119  *              Disadvantage:  special cases will be all over the place.
 120  */
 121
 122 /*
 123  * The initializers below do not include the attoptions or attacl fields,
 124  * but that's OK - we're never going to reference anything beyond the
 125  * fixed-size portion of the structure anyway.
 126  */
 127
 128 static FormData_pg_attribute a1 = {
 129         0, {"ctid"}, TIDOID, 0, sizeof(ItemPointerData),
 130         SelfItemPointerAttributeNumber, 0, -1, -1,
 131         false, 'p', 's', true, false, false, true, 0
 132 };
 133
 134 static FormData_pg_attribute a2 = {
 135         0, {"oid"}, OIDOID, 0, sizeof(Oid),
 136         ObjectIdAttributeNumber, 0, -1, -1,
 137         true, 'p', 'i', true, false, false, true, 0
 138 };
 139
 140 static FormData_pg_attribute a3 = {
 141         0, {"xmin"}, XIDOID, 0, sizeof(TransactionId),
 142         MinTransactionIdAttributeNumber, 0, -1, -1,
 143         true, 'p', 'i', true, false, false, true, 0
 144 };
 145
 146 static FormData_pg_attribute a4 = {
 147         0, {"cmin"}, CIDOID, 0, sizeof(CommandId),
 148         MinCommandIdAttributeNumber, 0, -1, -1,
 149         true, 'p', 'i', true, false, false, true, 0
 150 };
 151
 152 static FormData_pg_attribute a5 = {
 153         0, {"xmax"}, XIDOID, 0, sizeof(TransactionId),
 154         MaxTransactionIdAttributeNumber, 0, -1, -1,
 155         true, 'p', 'i', true, false, false, true, 0
 156 };
 157
 158 static FormData_pg_attribute a6 = {
 159         0, {"cmax"}, CIDOID, 0, sizeof(CommandId),
 160         MaxCommandIdAttributeNumber, 0, -1, -1,
 161         true, 'p', 'i', true, false, false, true, 0
 162 };
 163
 164 /*
 165  * We decided to call this attribute "tableoid" rather than say
 166  * "classoid" on the basis that in the future there may be more than one
 167  * table of a particular class/type. In any case table is still the word
 168  * used in SQL.
 169  */
 170 static FormData_pg_attribute a7 = {
 171         0, {"tableoid"}, OIDOID, 0, sizeof(Oid),
 172         TableOidAttributeNumber, 0, -1, -1,
 173         true, 'p', 'i', true, false, false, true, 0
 174 };
 175
 176 static const Form_pg_attribute SysAtt[] = {&a1, &a2, &a3, &a4, &a5, &a6, &a7};
 177
 178 /*
 179  * This function returns a Form_pg_attribute pointer for a system attribute.
 180  * Note that we elog if the presented attno is invalid, which would only
 181  * happen if there's a problem upstream.
 182  */
 183 Form_pg_attribute
 184 SystemAttributeDefinition(AttrNumber attno, bool relhasoids)
 185 {
 186         if (attno >= 0 || attno < -(int) lengthof(SysAtt))
 187                 elog(ERROR, "invalid system attribute number %d", attno);
 188         if (attno == ObjectIdAttributeNumber && !relhasoids)
 189                 elog(ERROR, "invalid system attribute number %d", attno);
 190         return SysAtt[-attno - 1];
 191 }
 192
 193 /*
 194  * If the given name is a system attribute name, return a Form_pg_attribute
 195  * pointer for a prototype definition.  If not, return NULL.
 196  */
 197 Form_pg_attribute
 198 SystemAttributeByName(const char *attname, bool relhasoids)
 199 {
 200         int                     j;
 201
 202         for (j = 0; j < (int) lengthof(SysAtt); j++)
 203         {
 204                 Form_pg_attribute att = SysAtt[j];
 205
 206                 if (relhasoids || att->attnum != ObjectIdAttributeNumber)
 207                 {
 208                         if (strcmp(NameStr(att->attname), attname) == 0)
 209                                 return att;
 210                 }
 211         }
 212
 213         return NULL;
 214 }
 215
 216
 217 /* ----------------------------------------------------------------
 218  *                              XXX END OF UGLY HARD CODED BADNESS XXX
 219  * ---------------------------------------------------------------- */
 220
 221
 222 /* ----------------------------------------------------------------
 223  *              heap_create             - Create an uncataloged heap relation
 224  *
 225  *              Note API change: the caller must now always provide the OID
 226  *              to use for the relation.
 227  *
 228  *              rel->rd_rel is initialized by RelationBuildLocalRelation,
 229  *              and is mostly zeroes at return.
 230  * ----------------------------------------------------------------
 231  */
 232 Relation
 233 heap_create(const char *relname,
 234                         Oid relnamespace,
 235                         Oid reltablespace,
 236                         Oid relid,
 237                         TupleDesc tupDesc,
 238                         char relkind,
 239                         bool shared_relation,
 240                         bool allow_system_table_mods)
 241 {
 242         bool            create_storage;
 243         Relation        rel;
 244
 245         /* The caller must have provided an OID for the relation. */
 246         Assert(OidIsValid(relid));
 247
 248         /*
 249          * sanity checks
 250          */
 251         if (!allow_system_table_mods &&
 252                 (IsSystemNamespace(relnamespace) || IsToastNamespace(relnamespace)) &&
 253                 IsNormalProcessingMode())
 254                 ereport(ERROR,
 255                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 256                                  errmsg("permission denied to create \"%s.%s\"",
 257                                                 get_namespace_name(relnamespace), relname),
 258                 errdetail("System catalog modifications are currently disallowed.")));
 259
 260         /*
 261          * Decide if we need storage or not, and handle a couple other special
 262          * cases for particular relkinds.
 263          */
 264         switch (relkind)
 265         {
 266                 case RELKIND_VIEW:
 267                 case RELKIND_COMPOSITE_TYPE:
 268                         create_storage = false;
 269
 270                         /*
 271                          * Force reltablespace to zero if the relation has no physical
 272                          * storage.  This is mainly just for cleanliness' sake.
 273                          */
 274                         reltablespace = InvalidOid;
 275                         break;
 276                 case RELKIND_SEQUENCE:
 277                         create_storage = true;
 278
 279                         /*
 280                          * Force reltablespace to zero for sequences, since we don't
 281                          * support moving them around into different tablespaces.
 282                          */
 283                         reltablespace = InvalidOid;
 284                         break;
 285                 default:
 286                         create_storage = true;
 287                         break;
 288         }
 289
 290         /*
 291          * Never allow a pg_class entry to explicitly specify the database's
 292          * default tablespace in reltablespace; force it to zero instead. This
 293          * ensures that if the database is cloned with a different default
 294          * tablespace, the pg_class entry will still match where CREATE DATABASE
 295          * will put the physically copied relation.
 296          *
 297          * Yes, this is a bit of a hack.
 298          */
 299         if (reltablespace == MyDatabaseTableSpace)
 300                 reltablespace = InvalidOid;
 301
 302         /*
 303          * build the relcache entry.
 304          */
 305         rel = RelationBuildLocalRelation(relname,
 306                                                                          relnamespace,
 307                                                                          tupDesc,
 308                                                                          relid,
 309                                                                          reltablespace,
 310                                                                          shared_relation);
 311
 312         /*
 313          * Have the storage manager create the relation's disk file, if needed.
 314          *
 315          * We only create the main fork here, other forks will be created on
 316          * demand.
 317          */
 318         if (create_storage)
 319         {
 320                 RelationOpenSmgr(rel);
 321                 RelationCreateStorage(rel->rd_node, rel->rd_istemp);
 322         }
 323
 324         return rel;
 325 }
 326
 327 /* ----------------------------------------------------------------
 328  *              heap_create_with_catalog                - Create a cataloged relation
 329  *
 330  *              this is done in multiple steps:
 331  *
 332  *              1) CheckAttributeNamesTypes() is used to make certain the tuple
 333  *                 descriptor contains a valid set of attribute names and types
 334  *
 335  *              2) pg_class is opened and get_relname_relid()
 336  *                 performs a scan to ensure that no relation with the
 337  *                 same name already exists.
 338  *
 339  *              3) heap_create() is called to create the new relation on disk.
 340  *
 341  *              4) TypeCreate() is called to define a new type corresponding
 342  *                 to the new relation.
 343  *
 344  *              5) AddNewRelationTuple() is called to register the
 345  *                 relation in pg_class.
 346  *
 347  *              6) AddNewAttributeTuples() is called to register the
 348  *                 new relation's schema in pg_attribute.
 349  *
 350  *              7) StoreConstraints is called ()                - vadim 08/22/97
 351  *
 352  *              8) the relations are closed and the new relation's oid
 353  *                 is returned.
 354  *
 355  * ----------------------------------------------------------------
 356  */
 357
 358 /* --------------------------------
 359  *              CheckAttributeNamesTypes
 360  *
 361  *              this is used to make certain the tuple descriptor contains a
 362  *              valid set of attribute names and datatypes.  a problem simply
 363  *              generates ereport(ERROR) which aborts the current transaction.
 364  * --------------------------------
 365  */
 366 void
 367 CheckAttributeNamesTypes(TupleDesc tupdesc, char relkind)
 368 {
 369         int                     i;
 370         int                     j;
 371         int                     natts = tupdesc->natts;
 372
 373         /* Sanity check on column count */
 374         if (natts < 0 || natts > MaxHeapAttributeNumber)
 375                 ereport(ERROR,
 376                                 (errcode(ERRCODE_TOO_MANY_COLUMNS),
 377                                  errmsg("tables can have at most %d columns",
 378                                                 MaxHeapAttributeNumber)));
 379
 380         /*
 381          * first check for collision with system attribute names
 382          *
 383          * Skip this for a view or type relation, since those don't have system
 384          * attributes.
 385          */
 386         if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
 387         {
 388                 for (i = 0; i < natts; i++)
 389                 {
 390                         if (SystemAttributeByName(NameStr(tupdesc->attrs[i]->attname),
 391                                                                           tupdesc->tdhasoid) != NULL)
 392                                 ereport(ERROR,
 393                                                 (errcode(ERRCODE_DUPLICATE_COLUMN),
 394                                                  errmsg("column name \"%s\" conflicts with a system column name",
 395                                                                 NameStr(tupdesc->attrs[i]->attname))));
 396                 }
 397         }
 398
 399         /*
 400          * next check for repeated attribute names
 401          */
 402         for (i = 1; i < natts; i++)
 403         {
 404                 for (j = 0; j < i; j++)
 405                 {
 406                         if (strcmp(NameStr(tupdesc->attrs[j]->attname),
 407                                            NameStr(tupdesc->attrs[i]->attname)) == 0)
 408                                 ereport(ERROR,
 409                                                 (errcode(ERRCODE_DUPLICATE_COLUMN),
 410                                                  errmsg("column name \"%s\" specified more than once",
 411                                                                 NameStr(tupdesc->attrs[j]->attname))));
 412                 }
 413         }
 414
 415         /*
 416          * next check the attribute types
 417          */
 418         for (i = 0; i < natts; i++)
 419         {
 420                 CheckAttributeType(NameStr(tupdesc->attrs[i]->attname),
 421                                                    tupdesc->attrs[i]->atttypid);
 422         }
 423 }
 424
 425 /* --------------------------------
 426  *              CheckAttributeType
 427  *
 428  *              Verify that the proposed datatype of an attribute is legal.
 429  *              This is needed because there are types (and pseudo-types)
 430  *              in the catalogs that we do not support as elements of real tuples.
 431  * --------------------------------
 432  */
 433 void
 434 CheckAttributeType(const char *attname, Oid atttypid)
 435 {
 436         char            att_typtype = get_typtype(atttypid);
 437
 438         if (atttypid == UNKNOWNOID)
 439         {
 440                 /*
 441                  * Warn user, but don't fail, if column to be created has UNKNOWN type
 442                  * (usually as a result of a 'retrieve into' - jolly)
 443                  */
 444                 ereport(WARNING,
 445                                 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 446                                  errmsg("column \"%s\" has type \"unknown\"", attname),
 447                                  errdetail("Proceeding with relation creation anyway.")));
 448         }
 449         else if (att_typtype == TYPTYPE_PSEUDO)
 450         {
 451                 /*
 452                  * Refuse any attempt to create a pseudo-type column, except for a
 453                  * special hack for pg_statistic: allow ANYARRAY during initdb
 454                  */
 455                 if (atttypid != ANYARRAYOID || IsUnderPostmaster)
 456                         ereport(ERROR,
 457                                         (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 458                                          errmsg("column \"%s\" has pseudo-type %s",
 459                                                         attname, format_type_be(atttypid))));
 460         }
 461         else if (att_typtype == TYPTYPE_COMPOSITE)
 462         {
 463                 /*
 464                  * For a composite type, recurse into its attributes.  You might think
 465                  * this isn't necessary, but since we allow system catalogs to break
 466                  * the rule, we have to guard against the case.
 467                  */
 468                 Relation        relation;
 469                 TupleDesc       tupdesc;
 470                 int                     i;
 471
 472                 relation = relation_open(get_typ_typrelid(atttypid), AccessShareLock);
 473
 474                 tupdesc = RelationGetDescr(relation);
 475
 476                 for (i = 0; i < tupdesc->natts; i++)
 477                 {
 478                         Form_pg_attribute attr = tupdesc->attrs[i];
 479
 480                         if (attr->attisdropped)
 481                                 continue;
 482                         CheckAttributeType(NameStr(attr->attname), attr->atttypid);
 483                 }
 484
 485                 relation_close(relation, AccessShareLock);
 486         }
 487 }
 488
 489 /*
 490  * InsertPgAttributeTuple
 491  *              Construct and insert a new tuple in pg_attribute.
 492  *
 493  * Caller has already opened and locked pg_attribute.  new_attribute is the
 494  * attribute to insert (but we ignore attacl and attoptions, which are always
 495  * initialized to NULL).
 496  *
 497  * indstate is the index state for CatalogIndexInsert.  It can be passed as
 498  * NULL, in which case we'll fetch the necessary info.  (Don't do this when
 499  * inserting multiple attributes, because it's a tad more expensive.)
 500  */
 501 void
 502 InsertPgAttributeTuple(Relation pg_attribute_rel,
 503                                            Form_pg_attribute new_attribute,
 504                                            CatalogIndexState indstate)
 505 {
 506         Datum           values[Natts_pg_attribute];
 507         bool            nulls[Natts_pg_attribute];
 508         HeapTuple       tup;
 509
 510         /* This is a tad tedious, but way cleaner than what we used to do... */
 511         memset(values, 0, sizeof(values));
 512         memset(nulls, false, sizeof(nulls));
 513
 514         values[Anum_pg_attribute_attrelid - 1] = ObjectIdGetDatum(new_attribute->attrelid);
 515         values[Anum_pg_attribute_attname - 1] = NameGetDatum(&new_attribute->attname);
 516         values[Anum_pg_attribute_atttypid - 1] = ObjectIdGetDatum(new_attribute->atttypid);
 517         values[Anum_pg_attribute_attstattarget - 1] = Int32GetDatum(new_attribute->attstattarget);
 518         values[Anum_pg_attribute_attlen - 1] = Int16GetDatum(new_attribute->attlen);
 519         values[Anum_pg_attribute_attnum - 1] = Int16GetDatum(new_attribute->attnum);
 520         values[Anum_pg_attribute_attndims - 1] = Int32GetDatum(new_attribute->attndims);
 521         values[Anum_pg_attribute_attcacheoff - 1] = Int32GetDatum(new_attribute->attcacheoff);
 522         values[Anum_pg_attribute_atttypmod - 1] = Int32GetDatum(new_attribute->atttypmod);
 523         values[Anum_pg_attribute_attbyval - 1] = BoolGetDatum(new_attribute->attbyval);
 524         values[Anum_pg_attribute_attstorage - 1] = CharGetDatum(new_attribute->attstorage);
 525         values[Anum_pg_attribute_attalign - 1] = CharGetDatum(new_attribute->attalign);
 526         values[Anum_pg_attribute_attnotnull - 1] = BoolGetDatum(new_attribute->attnotnull);
 527         values[Anum_pg_attribute_atthasdef - 1] = BoolGetDatum(new_attribute->atthasdef);
 528         values[Anum_pg_attribute_attisdropped - 1] = BoolGetDatum(new_attribute->attisdropped);
 529         values[Anum_pg_attribute_attislocal - 1] = BoolGetDatum(new_attribute->attislocal);
 530         values[Anum_pg_attribute_attinhcount - 1] = Int32GetDatum(new_attribute->attinhcount);
 531
 532         /* start out with empty permissions and empty options */
 533         nulls[Anum_pg_attribute_attacl - 1] = true;
 534         nulls[Anum_pg_attribute_attoptions - 1] = true;
 535
 536         tup = heap_form_tuple(RelationGetDescr(pg_attribute_rel), values, nulls);
 537
 538         /* finally insert the new tuple, update the indexes, and clean up */
 539         simple_heap_insert(pg_attribute_rel, tup);
 540
 541         if (indstate != NULL)
 542                 CatalogIndexInsert(indstate, tup);
 543         else
 544                 CatalogUpdateIndexes(pg_attribute_rel, tup);
 545
 546         heap_freetuple(tup);
 547 }
 548
 549 /* --------------------------------
 550  *              AddNewAttributeTuples
 551  *
 552  *              this registers the new relation's schema by adding
 553  *              tuples to pg_attribute.
 554  * --------------------------------
 555  */
 556 static void
 557 AddNewAttributeTuples(Oid new_rel_oid,
 558                                           TupleDesc tupdesc,
 559                                           char relkind,
 560                                           bool oidislocal,
 561                                           int oidinhcount)
 562 {
 563         Form_pg_attribute attr;
 564         int                     i;
 565         Relation        rel;
 566         CatalogIndexState indstate;
 567         int                     natts = tupdesc->natts;
 568         ObjectAddress myself,
 569                                 referenced;
 570
 571         /*
 572          * open pg_attribute and its indexes.
 573          */
 574         rel = heap_open(AttributeRelationId, RowExclusiveLock);
 575
 576         indstate = CatalogOpenIndexes(rel);
 577
 578         /*
 579          * First we add the user attributes.  This is also a convenient place to
 580          * add dependencies on their datatypes.
 581          */
 582         for (i = 0; i < natts; i++)
 583         {
 584                 attr = tupdesc->attrs[i];
 585                 /* Fill in the correct relation OID */
 586                 attr->attrelid = new_rel_oid;
 587                 /* Make sure these are OK, too */
 588                 attr->attstattarget = -1;
 589                 attr->attcacheoff = -1;
 590
 591                 InsertPgAttributeTuple(rel, attr, indstate);
 592
 593                 /* Add dependency info */
 594                 myself.classId = RelationRelationId;
 595                 myself.objectId = new_rel_oid;
 596                 myself.objectSubId = i + 1;
 597                 referenced.classId = TypeRelationId;
 598                 referenced.objectId = attr->atttypid;
 599                 referenced.objectSubId = 0;
 600                 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 601         }
 602
 603         /*
 604          * Next we add the system attributes.  Skip OID if rel has no OIDs. Skip
 605          * all for a view or type relation.  We don't bother with making datatype
 606          * dependencies here, since presumably all these types are pinned.
 607          */
 608         if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
 609         {
 610                 for (i = 0; i < (int) lengthof(SysAtt); i++)
 611                 {
 612                         FormData_pg_attribute attStruct;
 613
 614                         /* skip OID where appropriate */
 615                         if (!tupdesc->tdhasoid &&
 616                                 SysAtt[i]->attnum == ObjectIdAttributeNumber)
 617                                 continue;
 618
 619                         memcpy(&attStruct, (char *) SysAtt[i], sizeof(FormData_pg_attribute));
 620
 621                         /* Fill in the correct relation OID in the copied tuple */
 622                         attStruct.attrelid = new_rel_oid;
 623
 624                         /* Fill in correct inheritance info for the OID column */
 625                         if (attStruct.attnum == ObjectIdAttributeNumber)
 626                         {
 627                                 attStruct.attislocal = oidislocal;
 628                                 attStruct.attinhcount = oidinhcount;
 629                         }
 630
 631                         InsertPgAttributeTuple(rel, &attStruct, indstate);
 632                 }
 633         }
 634
 635         /*
 636          * clean up
 637          */
 638         CatalogCloseIndexes(indstate);
 639
 640         heap_close(rel, RowExclusiveLock);
 641 }
 642
 643 /* --------------------------------
 644  *              InsertPgClassTuple
 645  *
 646  *              Construct and insert a new tuple in pg_class.
 647  *
 648  * Caller has already opened and locked pg_class.
 649  * Tuple data is taken from new_rel_desc->rd_rel, except for the
 650  * variable-width fields which are not present in a cached reldesc.
 651  * relacl and reloptions are passed in Datum form (to avoid having
 652  * to reference the data types in heap.h).  Pass (Datum) 0 to set them
 653  * to NULL.
 654  * --------------------------------
 655  */
 656 void
 657 InsertPgClassTuple(Relation pg_class_desc,
 658                                    Relation new_rel_desc,
 659                                    Oid new_rel_oid,
 660                                    Datum relacl,
 661                                    Datum reloptions)
 662 {
 663         Form_pg_class rd_rel = new_rel_desc->rd_rel;
 664         Datum           values[Natts_pg_class];
 665         bool            nulls[Natts_pg_class];
 666         HeapTuple       tup;
 667
 668         /* This is a tad tedious, but way cleaner than what we used to do... */
 669         memset(values, 0, sizeof(values));
 670         memset(nulls, false, sizeof(nulls));
 671
 672         values[Anum_pg_class_relname - 1] = NameGetDatum(&rd_rel->relname);
 673         values[Anum_pg_class_relnamespace - 1] = ObjectIdGetDatum(rd_rel->relnamespace);
 674         values[Anum_pg_class_reltype - 1] = ObjectIdGetDatum(rd_rel->reltype);
 675         values[Anum_pg_class_reloftype - 1] = ObjectIdGetDatum(rd_rel->reloftype);
 676         values[Anum_pg_class_relowner - 1] = ObjectIdGetDatum(rd_rel->relowner);
 677         values[Anum_pg_class_relam - 1] = ObjectIdGetDatum(rd_rel->relam);
 678         values[Anum_pg_class_relfilenode - 1] = ObjectIdGetDatum(rd_rel->relfilenode);
 679         values[Anum_pg_class_reltablespace - 1] = ObjectIdGetDatum(rd_rel->reltablespace);
 680         values[Anum_pg_class_relpages - 1] = Int32GetDatum(rd_rel->relpages);
 681         values[Anum_pg_class_reltuples - 1] = Float4GetDatum(rd_rel->reltuples);
 682         values[Anum_pg_class_reltoastrelid - 1] = ObjectIdGetDatum(rd_rel->reltoastrelid);
 683         values[Anum_pg_class_reltoastidxid - 1] = ObjectIdGetDatum(rd_rel->reltoastidxid);
 684         values[Anum_pg_class_relhasindex - 1] = BoolGetDatum(rd_rel->relhasindex);
 685         values[Anum_pg_class_relisshared - 1] = BoolGetDatum(rd_rel->relisshared);
 686         values[Anum_pg_class_relistemp - 1] = BoolGetDatum(rd_rel->relistemp);
 687         values[Anum_pg_class_relkind - 1] = CharGetDatum(rd_rel->relkind);
 688         values[Anum_pg_class_relnatts - 1] = Int16GetDatum(rd_rel->relnatts);
 689         values[Anum_pg_class_relchecks - 1] = Int16GetDatum(rd_rel->relchecks);
 690         values[Anum_pg_class_relhasoids - 1] = BoolGetDatum(rd_rel->relhasoids);
 691         values[Anum_pg_class_relhaspkey - 1] = BoolGetDatum(rd_rel->relhaspkey);
 692         values[Anum_pg_class_relhasexclusion - 1] = BoolGetDatum(rd_rel->relhasexclusion);
 693         values[Anum_pg_class_relhasrules - 1] = BoolGetDatum(rd_rel->relhasrules);
 694         values[Anum_pg_class_relhastriggers - 1] = BoolGetDatum(rd_rel->relhastriggers);
 695         values[Anum_pg_class_relhassubclass - 1] = BoolGetDatum(rd_rel->relhassubclass);
 696         values[Anum_pg_class_relfrozenxid - 1] = TransactionIdGetDatum(rd_rel->relfrozenxid);
 697         if (relacl != (Datum) 0)
 698                 values[Anum_pg_class_relacl - 1] = relacl;
 699         else
 700                 nulls[Anum_pg_class_relacl - 1] = true;
 701         if (reloptions != (Datum) 0)
 702                 values[Anum_pg_class_reloptions - 1] = reloptions;
 703         else
 704                 nulls[Anum_pg_class_reloptions - 1] = true;
 705
 706         tup = heap_form_tuple(RelationGetDescr(pg_class_desc), values, nulls);
 707
 708         /*
 709          * The new tuple must have the oid already chosen for the rel.  Sure would
 710          * be embarrassing to do this sort of thing in polite company.
 711          */
 712         HeapTupleSetOid(tup, new_rel_oid);
 713
 714         /* finally insert the new tuple, update the indexes, and clean up */
 715         simple_heap_insert(pg_class_desc, tup);
 716
 717         CatalogUpdateIndexes(pg_class_desc, tup);
 718
 719         heap_freetuple(tup);
 720 }
 721
 722 /* --------------------------------
 723  *              AddNewRelationTuple
 724  *
 725  *              this registers the new relation in the catalogs by
 726  *              adding a tuple to pg_class.
 727  * --------------------------------
 728  */
 729 static void
 730 AddNewRelationTuple(Relation pg_class_desc,
 731                                         Relation new_rel_desc,
 732                                         Oid new_rel_oid,
 733                                         Oid new_type_oid,
 734                                         Oid reloftype,
 735                                         Oid relowner,
 736                                         char relkind,
 737                                         Datum relacl,
 738                                         Datum reloptions)
 739 {
 740         Form_pg_class new_rel_reltup;
 741
 742         /*
 743          * first we update some of the information in our uncataloged relation's
 744          * relation descriptor.
 745          */
 746         new_rel_reltup = new_rel_desc->rd_rel;
 747
 748         switch (relkind)
 749         {
 750                 case RELKIND_RELATION:
 751                 case RELKIND_INDEX:
 752                 case RELKIND_TOASTVALUE:
 753                         /* The relation is real, but as yet empty */
 754                         new_rel_reltup->relpages = 0;
 755                         new_rel_reltup->reltuples = 0;
 756                         break;
 757                 case RELKIND_SEQUENCE:
 758                         /* Sequences always have a known size */
 759                         new_rel_reltup->relpages = 1;
 760                         new_rel_reltup->reltuples = 1;
 761                         break;
 762                 default:
 763                         /* Views, etc, have no disk storage */
 764                         new_rel_reltup->relpages = 0;
 765                         new_rel_reltup->reltuples = 0;
 766                         break;
 767         }
 768
 769         /* Initialize relfrozenxid */
 770         if (relkind == RELKIND_RELATION ||
 771                 relkind == RELKIND_TOASTVALUE)
 772         {
 773                 /*
 774                  * Initialize to the minimum XID that could put tuples in the table.
 775                  * We know that no xacts older than RecentXmin are still running, so
 776                  * that will do.
 777                  */
 778                 new_rel_reltup->relfrozenxid = RecentXmin;
 779         }
 780         else
 781         {
 782                 /*
 783                  * Other relation types will not contain XIDs, so set relfrozenxid to
 784                  * InvalidTransactionId.  (Note: a sequence does contain a tuple, but
 785                  * we force its xmin to be FrozenTransactionId always; see
 786                  * commands/sequence.c.)
 787                  */
 788                 new_rel_reltup->relfrozenxid = InvalidTransactionId;
 789         }
 790
 791         new_rel_reltup->relowner = relowner;
 792         new_rel_reltup->reltype = new_type_oid;
 793         new_rel_reltup->reloftype = reloftype;
 794         new_rel_reltup->relkind = relkind;
 795
 796         new_rel_desc->rd_att->tdtypeid = new_type_oid;
 797
 798         /* Now build and insert the tuple */
 799         InsertPgClassTuple(pg_class_desc, new_rel_desc, new_rel_oid,
 800                                            relacl, reloptions);
 801 }
 802
 803
 804 /* --------------------------------
 805  *              AddNewRelationType -
 806  *
 807  *              define a composite type corresponding to the new relation
 808  * --------------------------------
 809  */
 810 static Oid
 811 AddNewRelationType(const char *typeName,
 812                                    Oid typeNamespace,
 813                                    Oid new_rel_oid,
 814                                    char new_rel_kind,
 815                                    Oid ownerid,
 816                                    Oid new_row_type,
 817                                    Oid new_array_type)
 818 {
 819         return
 820                 TypeCreate(new_row_type,                /* optional predetermined OID */
 821                                    typeName,    /* type name */
 822                                    typeNamespace,               /* type namespace */
 823                                    new_rel_oid, /* relation oid */
 824                                    new_rel_kind,        /* relation kind */
 825                                    ownerid,             /* owner's ID */
 826                                    -1,                  /* internal size (varlena) */
 827                                    TYPTYPE_COMPOSITE,   /* type-type (composite) */
 828                                    TYPCATEGORY_COMPOSITE,               /* type-category (ditto) */
 829                                    false,               /* composite types are never preferred */
 830                                    DEFAULT_TYPDELIM,    /* default array delimiter */
 831                                    F_RECORD_IN, /* input procedure */
 832                                    F_RECORD_OUT,        /* output procedure */
 833                                    F_RECORD_RECV,               /* receive procedure */
 834                                    F_RECORD_SEND,               /* send procedure */
 835                                    InvalidOid,  /* typmodin procedure - none */
 836                                    InvalidOid,  /* typmodout procedure - none */
 837                                    InvalidOid,  /* analyze procedure - default */
 838                                    InvalidOid,  /* array element type - irrelevant */
 839                                    false,               /* this is not an array type */
 840                                    new_array_type,              /* array type if any */
 841                                    InvalidOid,  /* domain base type - irrelevant */
 842                                    NULL,                /* default value - none */
 843                                    NULL,                /* default binary representation */
 844                                    false,               /* passed by reference */
 845                                    'd',                 /* alignment - must be the largest! */
 846                                    'x',                 /* fully TOASTable */
 847                                    -1,                  /* typmod */
 848                                    0,                   /* array dimensions for typBaseType */
 849                                    false);              /* Type NOT NULL */
 850 }
 851
 852 /* --------------------------------
 853  *              heap_create_with_catalog
 854  *
 855  *              creates a new cataloged relation.  see comments above.
 856  *
 857  * Arguments:
 858  *      relname: name to give to new rel
 859  *      relnamespace: OID of namespace it goes in
 860  *      reltablespace: OID of tablespace it goes in
 861  *      relid: OID to assign to new rel, or InvalidOid to select a new OID
 862  *      reltypeid: OID to assign to rel's rowtype, or InvalidOid to select one
 863  *      ownerid: OID of new rel's owner
 864  *      tupdesc: tuple descriptor (source of column definitions)
 865  *      cooked_constraints: list of precooked check constraints and defaults
 866  *      relkind: relkind for new rel
 867  *      shared_relation: TRUE if it's to be a shared relation
 868  *      oidislocal: TRUE if oid column (if any) should be marked attislocal
 869  *      oidinhcount: attinhcount to assign to oid column (if any)
 870  *      oncommit: ON COMMIT marking (only relevant if it's a temp table)
 871  *      reloptions: reloptions in Datum form, or (Datum) 0 if none
 872  *      use_user_acl: TRUE if should look for user-defined default permissions;
 873  *              if FALSE, relacl is always set NULL
 874  *      allow_system_table_mods: TRUE to allow creation in system namespaces
 875  *
 876  * Returns the OID of the new relation
 877  * --------------------------------
 878  */
 879 Oid
 880 heap_create_with_catalog(const char *relname,
 881                                                  Oid relnamespace,
 882                                                  Oid reltablespace,
 883                                                  Oid relid,
 884                                                  Oid reltypeid,
 885                                                  Oid reloftypeid,
 886                                                  Oid ownerid,
 887                                                  TupleDesc tupdesc,
 888                                                  List *cooked_constraints,
 889                                                  char relkind,
 890                                                  bool shared_relation,
 891                                                  bool oidislocal,
 892                                                  int oidinhcount,
 893                                                  OnCommitAction oncommit,
 894                                                  Datum reloptions,
 895                                                  bool use_user_acl,
 896                                                  bool allow_system_table_mods)
 897 {
 898         Relation        pg_class_desc;
 899         Relation        new_rel_desc;
 900         Acl                *relacl;
 901         Oid                     old_type_oid;
 902         Oid                     new_type_oid;
 903         Oid                     new_array_oid = InvalidOid;
 904
 905         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
 906
 907         /*
 908          * sanity checks
 909          */
 910         Assert(IsNormalProcessingMode() || IsBootstrapProcessingMode());
 911
 912         CheckAttributeNamesTypes(tupdesc, relkind);
 913
 914         if (get_relname_relid(relname, relnamespace))
 915                 ereport(ERROR,
 916                                 (errcode(ERRCODE_DUPLICATE_TABLE),
 917                                  errmsg("relation \"%s\" already exists", relname)));
 918
 919         /*
 920          * Since we are going to create a rowtype as well, also check for
 921          * collision with an existing type name.  If there is one and it's an
 922          * autogenerated array, we can rename it out of the way; otherwise we can
 923          * at least give a good error message.
 924          */
 925         old_type_oid = GetSysCacheOid(TYPENAMENSP,
 926                                                                   CStringGetDatum(relname),
 927                                                                   ObjectIdGetDatum(relnamespace),
 928                                                                   0, 0);
 929         if (OidIsValid(old_type_oid))
 930         {
 931                 if (!moveArrayTypeName(old_type_oid, relname, relnamespace))
 932                         ereport(ERROR,
 933                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
 934                                          errmsg("type \"%s\" already exists", relname),
 935                            errhint("A relation has an associated type of the same name, "
 936                                            "so you must use a name that doesn't conflict "
 937                                            "with any existing type.")));
 938         }
 939
 940         /*
 941          * Validate shared/non-shared tablespace (must check this before doing
 942          * GetNewRelFileNode, to prevent Assert therein)
 943          */
 944         if (shared_relation)
 945         {
 946                 if (reltablespace != GLOBALTABLESPACE_OID)
 947                         /* elog since this is not a user-facing error */
 948                         elog(ERROR,
 949                                  "shared relations must be placed in pg_global tablespace");
 950         }
 951         else
 952         {
 953                 if (reltablespace == GLOBALTABLESPACE_OID)
 954                         ereport(ERROR,
 955                                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 956                                          errmsg("only shared relations can be placed in pg_global tablespace")));
 957         }
 958
 959         /*
 960          * Allocate an OID for the relation, unless we were told what to use.
 961          *
 962          * The OID will be the relfilenode as well, so make sure it doesn't
 963          * collide with either pg_class OIDs or existing physical files.
 964          */
 965         if (!OidIsValid(relid))
 966         {
 967                 /* Use binary-upgrade overrides if applicable */
 968                 if (OidIsValid(binary_upgrade_next_heap_relfilenode) &&
 969                         (relkind == RELKIND_RELATION || relkind == RELKIND_SEQUENCE ||
 970                          relkind == RELKIND_VIEW || relkind == RELKIND_COMPOSITE_TYPE))
 971                 {
 972                         relid = binary_upgrade_next_heap_relfilenode;
 973                         binary_upgrade_next_heap_relfilenode = InvalidOid;
 974                 }
 975                 else if (OidIsValid(binary_upgrade_next_toast_relfilenode) &&
 976                                  relkind == RELKIND_TOASTVALUE)
 977                 {
 978                         relid = binary_upgrade_next_toast_relfilenode;
 979                         binary_upgrade_next_toast_relfilenode = InvalidOid;
 980                 }
 981                 else
 982                         relid = GetNewRelFileNode(reltablespace, shared_relation,
 983                                                                           pg_class_desc);
 984         }
 985
 986         /*
 987          * Determine the relation's initial permissions.
 988          */
 989         if (use_user_acl)
 990         {
 991                 switch (relkind)
 992                 {
 993                         case RELKIND_RELATION:
 994                         case RELKIND_VIEW:
 995                                 relacl = get_user_default_acl(ACL_OBJECT_RELATION, ownerid,
 996                                                                                           relnamespace);
 997                                 break;
 998                         case RELKIND_SEQUENCE:
 999                                 relacl = get_user_default_acl(ACL_OBJECT_SEQUENCE, ownerid,
1000                                                                                           relnamespace);
1001                                 break;
1002                         default:
1003                                 relacl = NULL;
1004                                 break;
1005                 }
1006         }
1007         else
1008                 relacl = NULL;
1009
1010         /*
1011          * Create the relcache entry (mostly dummy at this point) and the physical
1012          * disk file.  (If we fail further down, it's the smgr's responsibility to
1013          * remove the disk file again.)
1014          */
1015         new_rel_desc = heap_create(relname,
1016                                                            relnamespace,
1017                                                            reltablespace,
1018                                                            relid,
1019                                                            tupdesc,
1020                                                            relkind,
1021                                                            shared_relation,
1022                                                            allow_system_table_mods);
1023
1024         Assert(relid == RelationGetRelid(new_rel_desc));
1025
1026         /*
1027          * Decide whether to create an array type over the relation's rowtype. We
1028          * do not create any array types for system catalogs (ie, those made
1029          * during initdb).      We create array types for regular relations, views,
1030          * and composite types ... but not, eg, for toast tables or sequences.
1031          */
1032         if (IsUnderPostmaster && (relkind == RELKIND_RELATION ||
1033                                                           relkind == RELKIND_VIEW ||
1034                                                           relkind == RELKIND_COMPOSITE_TYPE))
1035                 new_array_oid = AssignTypeArrayOid();
1036
1037         /*
1038          * Since defining a relation also defines a complex type, we add a new
1039          * system type corresponding to the new relation.  The OID of the type
1040          * can be preselected by the caller, but if reltypeid is InvalidOid,
1041          * we'll generate a new OID for it.
1042          *
1043          * NOTE: we could get a unique-index failure here, in case someone else is
1044          * creating the same type name in parallel but hadn't committed yet when
1045          * we checked for a duplicate name above.
1046          */
1047         new_type_oid = AddNewRelationType(relname,
1048                                                                           relnamespace,
1049                                                                           relid,
1050                                                                           relkind,
1051                                                                           ownerid,
1052                                                                           reltypeid,
1053                                                                           new_array_oid);
1054
1055         /*
1056          * Now make the array type if wanted.
1057          */
1058         if (OidIsValid(new_array_oid))
1059         {
1060                 char       *relarrayname;
1061
1062                 relarrayname = makeArrayTypeName(relname, relnamespace);
1063
1064                 TypeCreate(new_array_oid,               /* force the type's OID to this */
1065                                    relarrayname,        /* Array type name */
1066                                    relnamespace,        /* Same namespace as parent */
1067                                    InvalidOid,  /* Not composite, no relationOid */
1068                                    0,                   /* relkind, also N/A here */
1069                                    ownerid,             /* owner's ID */
1070                                    -1,                  /* Internal size (varlena) */
1071                                    TYPTYPE_BASE,        /* Not composite - typelem is */
1072                                    TYPCATEGORY_ARRAY,   /* type-category (array) */
1073                                    false,               /* array types are never preferred */
1074                                    DEFAULT_TYPDELIM,    /* default array delimiter */
1075                                    F_ARRAY_IN,  /* array input proc */
1076                                    F_ARRAY_OUT, /* array output proc */
1077                                    F_ARRAY_RECV,        /* array recv (bin) proc */
1078                                    F_ARRAY_SEND,        /* array send (bin) proc */
1079                                    InvalidOid,  /* typmodin procedure - none */
1080                                    InvalidOid,  /* typmodout procedure - none */
1081                                    InvalidOid,  /* analyze procedure - default */
1082                                    new_type_oid,        /* array element type - the rowtype */
1083                                    true,                /* yes, this is an array type */
1084                                    InvalidOid,  /* this has no array type */
1085                                    InvalidOid,  /* domain base type - irrelevant */
1086                                    NULL,                /* default value - none */
1087                                    NULL,                /* default binary representation */
1088                                    false,               /* passed by reference */
1089                                    'd',                 /* alignment - must be the largest! */
1090                                    'x',                 /* fully TOASTable */
1091                                    -1,                  /* typmod */
1092                                    0,                   /* array dimensions for typBaseType */
1093                                    false);              /* Type NOT NULL */
1094
1095                 pfree(relarrayname);
1096         }
1097
1098         /*
1099          * now create an entry in pg_class for the relation.
1100          *
1101          * NOTE: we could get a unique-index failure here, in case someone else is
1102          * creating the same relation name in parallel but hadn't committed yet
1103          * when we checked for a duplicate name above.
1104          */
1105         AddNewRelationTuple(pg_class_desc,
1106                                                 new_rel_desc,
1107                                                 relid,
1108                                                 new_type_oid,
1109                                                 reloftypeid,
1110                                                 ownerid,
1111                                                 relkind,
1112                                                 PointerGetDatum(relacl),
1113                                                 reloptions);
1114
1115         /*
1116          * now add tuples to pg_attribute for the attributes in our new relation.
1117          */
1118         AddNewAttributeTuples(relid, new_rel_desc->rd_att, relkind,
1119                                                   oidislocal, oidinhcount);
1120
1121         /*
1122          * Make a dependency link to force the relation to be deleted if its
1123          * namespace is.  Also make a dependency link to its owner, as well
1124          * as dependencies for any roles mentioned in the default ACL.
1125          *
1126          * For composite types, these dependencies are tracked for the pg_type
1127          * entry, so we needn't record them here.  Likewise, TOAST tables don't
1128          * need a namespace dependency (they live in a pinned namespace) nor an
1129          * owner dependency (they depend indirectly through the parent table),
1130          * nor should they have any ACL entries.
1131          *
1132          * Also, skip this in bootstrap mode, since we don't make dependencies
1133          * while bootstrapping.
1134          */
1135         if (relkind != RELKIND_COMPOSITE_TYPE &&
1136                 relkind != RELKIND_TOASTVALUE &&
1137                 !IsBootstrapProcessingMode())
1138         {
1139                 ObjectAddress myself,
1140                                         referenced;
1141
1142                 myself.classId = RelationRelationId;
1143                 myself.objectId = relid;
1144                 myself.objectSubId = 0;
1145                 referenced.classId = NamespaceRelationId;
1146                 referenced.objectId = relnamespace;
1147                 referenced.objectSubId = 0;
1148                 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
1149
1150                 recordDependencyOnOwner(RelationRelationId, relid, ownerid);
1151
1152                 if (reloftypeid)
1153                 {
1154                         referenced.classId = TypeRelationId;
1155                         referenced.objectId = reloftypeid;
1156                         referenced.objectSubId = 0;
1157                         recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
1158                 }
1159
1160                 if (relacl != NULL)
1161                 {
1162                         int                     nnewmembers;
1163                         Oid                *newmembers;
1164
1165                         nnewmembers = aclmembers(relacl, &newmembers);
1166                         updateAclDependencies(RelationRelationId, relid, 0,
1167                                                                   ownerid, true,
1168                                                                   0, NULL,
1169                                                                   nnewmembers, newmembers);
1170                 }
1171         }
1172
1173         /*
1174          * Store any supplied constraints and defaults.
1175          *
1176          * NB: this may do a CommandCounterIncrement and rebuild the relcache
1177          * entry, so the relation must be valid and self-consistent at this point.
1178          * In particular, there are not yet constraints and defaults anywhere.
1179          */
1180         StoreConstraints(new_rel_desc, cooked_constraints);
1181
1182         /*
1183          * If there's a special on-commit action, remember it
1184          */
1185         if (oncommit != ONCOMMIT_NOOP)
1186                 register_on_commit_action(relid, oncommit);
1187
1188         /*
1189          * ok, the relation has been cataloged, so close our relations and return
1190          * the OID of the newly created relation.
1191          */
1192         heap_close(new_rel_desc, NoLock);       /* do not unlock till end of xact */
1193         heap_close(pg_class_desc, RowExclusiveLock);
1194
1195         return relid;
1196 }
1197
1198
1199 /*
1200  *              RelationRemoveInheritance
1201  *
1202  * Formerly, this routine checked for child relations and aborted the
1203  * deletion if any were found.  Now we rely on the dependency mechanism
1204  * to check for or delete child relations.      By the time we get here,
1205  * there are no children and we need only remove any pg_inherits rows
1206  * linking this relation to its parent(s).
1207  */
1208 static void
1209 RelationRemoveInheritance(Oid relid)
1210 {
1211         Relation        catalogRelation;
1212         SysScanDesc scan;
1213         ScanKeyData key;
1214         HeapTuple       tuple;
1215
1216         catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
1217
1218         ScanKeyInit(&key,
1219                                 Anum_pg_inherits_inhrelid,
1220                                 BTEqualStrategyNumber, F_OIDEQ,
1221                                 ObjectIdGetDatum(relid));
1222
1223         scan = systable_beginscan(catalogRelation, InheritsRelidSeqnoIndexId, true,
1224                                                           SnapshotNow, 1, &key);
1225
1226         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1227                 simple_heap_delete(catalogRelation, &tuple->t_self);
1228
1229         systable_endscan(scan);
1230         heap_close(catalogRelation, RowExclusiveLock);
1231 }
1232
1233 /*
1234  *              DeleteRelationTuple
1235  *
1236  * Remove pg_class row for the given relid.
1237  *
1238  * Note: this is shared by relation deletion and index deletion.  It's
1239  * not intended for use anyplace else.
1240  */
1241 void
1242 DeleteRelationTuple(Oid relid)
1243 {
1244         Relation        pg_class_desc;
1245         HeapTuple       tup;
1246
1247         /* Grab an appropriate lock on the pg_class relation */
1248         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
1249
1250         tup = SearchSysCache(RELOID,
1251                                                  ObjectIdGetDatum(relid),
1252                                                  0, 0, 0);
1253         if (!HeapTupleIsValid(tup))
1254                 elog(ERROR, "cache lookup failed for relation %u", relid);
1255
1256         /* delete the relation tuple from pg_class, and finish up */
1257         simple_heap_delete(pg_class_desc, &tup->t_self);
1258
1259         ReleaseSysCache(tup);
1260
1261         heap_close(pg_class_desc, RowExclusiveLock);
1262 }
1263
1264 /*
1265  *              DeleteAttributeTuples
1266  *
1267  * Remove pg_attribute rows for the given relid.
1268  *
1269  * Note: this is shared by relation deletion and index deletion.  It's
1270  * not intended for use anyplace else.
1271  */
1272 void
1273 DeleteAttributeTuples(Oid relid)
1274 {
1275         Relation        attrel;
1276         SysScanDesc scan;
1277         ScanKeyData key[1];
1278         HeapTuple       atttup;
1279
1280         /* Grab an appropriate lock on the pg_attribute relation */
1281         attrel = heap_open(AttributeRelationId, RowExclusiveLock);
1282
1283         /* Use the index to scan only attributes of the target relation */
1284         ScanKeyInit(&key[0],
1285                                 Anum_pg_attribute_attrelid,
1286                                 BTEqualStrategyNumber, F_OIDEQ,
1287                                 ObjectIdGetDatum(relid));
1288
1289         scan = systable_beginscan(attrel, AttributeRelidNumIndexId, true,
1290                                                           SnapshotNow, 1, key);
1291
1292         /* Delete all the matching tuples */
1293         while ((atttup = systable_getnext(scan)) != NULL)
1294                 simple_heap_delete(attrel, &atttup->t_self);
1295
1296         /* Clean up after the scan */
1297         systable_endscan(scan);
1298         heap_close(attrel, RowExclusiveLock);
1299 }
1300
1301 /*
1302  *              RemoveAttributeById
1303  *
1304  * This is the guts of ALTER TABLE DROP COLUMN: actually mark the attribute
1305  * deleted in pg_attribute.  We also remove pg_statistic entries for it.
1306  * (Everything else needed, such as getting rid of any pg_attrdef entry,
1307  * is handled by dependency.c.)
1308  */
1309 void
1310 RemoveAttributeById(Oid relid, AttrNumber attnum)
1311 {
1312         Relation        rel;
1313         Relation        attr_rel;
1314         HeapTuple       tuple;
1315         Form_pg_attribute attStruct;
1316         char            newattname[NAMEDATALEN];
1317
1318         /*
1319          * Grab an exclusive lock on the target table, which we will NOT release
1320          * until end of transaction.  (In the simple case where we are directly
1321          * dropping this column, AlterTableDropColumn already did this ... but
1322          * when cascading from a drop of some other object, we may not have any
1323          * lock.)
1324          */
1325         rel = relation_open(relid, AccessExclusiveLock);
1326
1327         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1328
1329         tuple = SearchSysCacheCopy(ATTNUM,
1330                                                            ObjectIdGetDatum(relid),
1331                                                            Int16GetDatum(attnum),
1332                                                            0, 0);
1333         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
1334                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1335                          attnum, relid);
1336         attStruct = (Form_pg_attribute) GETSTRUCT(tuple);
1337
1338         if (attnum < 0)
1339         {
1340                 /* System attribute (probably OID) ... just delete the row */
1341
1342                 simple_heap_delete(attr_rel, &tuple->t_self);
1343         }
1344         else
1345         {
1346                 /* Dropping user attributes is lots harder */
1347
1348                 /* Mark the attribute as dropped */
1349                 attStruct->attisdropped = true;
1350
1351                 /*
1352                  * Set the type OID to invalid.  A dropped attribute's type link
1353                  * cannot be relied on (once the attribute is dropped, the type might
1354                  * be too). Fortunately we do not need the type row --- the only
1355                  * really essential information is the type's typlen and typalign,
1356                  * which are preserved in the attribute's attlen and attalign.  We set
1357                  * atttypid to zero here as a means of catching code that incorrectly
1358                  * expects it to be valid.
1359                  */
1360                 attStruct->atttypid = InvalidOid;
1361
1362                 /* Remove any NOT NULL constraint the column may have */
1363                 attStruct->attnotnull = false;
1364
1365                 /* We don't want to keep stats for it anymore */
1366                 attStruct->attstattarget = 0;
1367
1368                 /*
1369                  * Change the column name to something that isn't likely to conflict
1370                  */
1371                 snprintf(newattname, sizeof(newattname),
1372                                  "........pg.dropped.%d........", attnum);
1373                 namestrcpy(&(attStruct->attname), newattname);
1374
1375                 simple_heap_update(attr_rel, &tuple->t_self, tuple);
1376
1377                 /* keep the system catalog indexes current */
1378                 CatalogUpdateIndexes(attr_rel, tuple);
1379         }
1380
1381         /*
1382          * Because updating the pg_attribute row will trigger a relcache flush for
1383          * the target relation, we need not do anything else to notify other
1384          * backends of the change.
1385          */
1386
1387         heap_close(attr_rel, RowExclusiveLock);
1388
1389         if (attnum > 0)
1390                 RemoveStatistics(relid, attnum);
1391
1392         relation_close(rel, NoLock);
1393 }
1394
1395 /*
1396  *              RemoveAttrDefault
1397  *
1398  * If the specified relation/attribute has a default, remove it.
1399  * (If no default, raise error if complain is true, else return quietly.)
1400  */
1401 void
1402 RemoveAttrDefault(Oid relid, AttrNumber attnum,
1403                                   DropBehavior behavior, bool complain)
1404 {
1405         Relation        attrdef_rel;
1406         ScanKeyData scankeys[2];
1407         SysScanDesc scan;
1408         HeapTuple       tuple;
1409         bool            found = false;
1410
1411         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1412
1413         ScanKeyInit(&scankeys[0],
1414                                 Anum_pg_attrdef_adrelid,
1415                                 BTEqualStrategyNumber, F_OIDEQ,
1416                                 ObjectIdGetDatum(relid));
1417         ScanKeyInit(&scankeys[1],
1418                                 Anum_pg_attrdef_adnum,
1419                                 BTEqualStrategyNumber, F_INT2EQ,
1420                                 Int16GetDatum(attnum));
1421
1422         scan = systable_beginscan(attrdef_rel, AttrDefaultIndexId, true,
1423                                                           SnapshotNow, 2, scankeys);
1424
1425         /* There should be at most one matching tuple, but we loop anyway */
1426         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1427         {
1428                 ObjectAddress object;
1429
1430                 object.classId = AttrDefaultRelationId;
1431                 object.objectId = HeapTupleGetOid(tuple);
1432                 object.objectSubId = 0;
1433
1434                 performDeletion(&object, behavior);
1435
1436                 found = true;
1437         }
1438
1439         systable_endscan(scan);
1440         heap_close(attrdef_rel, RowExclusiveLock);
1441
1442         if (complain && !found)
1443                 elog(ERROR, "could not find attrdef tuple for relation %u attnum %d",
1444                          relid, attnum);
1445 }
1446
1447 /*
1448  *              RemoveAttrDefaultById
1449  *
1450  * Remove a pg_attrdef entry specified by OID.  This is the guts of
1451  * attribute-default removal.  Note it should be called via performDeletion,
1452  * not directly.
1453  */
1454 void
1455 RemoveAttrDefaultById(Oid attrdefId)
1456 {
1457         Relation        attrdef_rel;
1458         Relation        attr_rel;
1459         Relation        myrel;
1460         ScanKeyData scankeys[1];
1461         SysScanDesc scan;
1462         HeapTuple       tuple;
1463         Oid                     myrelid;
1464         AttrNumber      myattnum;
1465
1466         /* Grab an appropriate lock on the pg_attrdef relation */
1467         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1468
1469         /* Find the pg_attrdef tuple */
1470         ScanKeyInit(&scankeys[0],
1471                                 ObjectIdAttributeNumber,
1472                                 BTEqualStrategyNumber, F_OIDEQ,
1473                                 ObjectIdGetDatum(attrdefId));
1474
1475         scan = systable_beginscan(attrdef_rel, AttrDefaultOidIndexId, true,
1476                                                           SnapshotNow, 1, scankeys);
1477
1478         tuple = systable_getnext(scan);
1479         if (!HeapTupleIsValid(tuple))
1480                 elog(ERROR, "could not find tuple for attrdef %u", attrdefId);
1481
1482         myrelid = ((Form_pg_attrdef) GETSTRUCT(tuple))->adrelid;
1483         myattnum = ((Form_pg_attrdef) GETSTRUCT(tuple))->adnum;
1484
1485         /* Get an exclusive lock on the relation owning the attribute */
1486         myrel = relation_open(myrelid, AccessExclusiveLock);
1487
1488         /* Now we can delete the pg_attrdef row */
1489         simple_heap_delete(attrdef_rel, &tuple->t_self);
1490
1491         systable_endscan(scan);
1492         heap_close(attrdef_rel, RowExclusiveLock);
1493
1494         /* Fix the pg_attribute row */
1495         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1496
1497         tuple = SearchSysCacheCopy(ATTNUM,
1498                                                            ObjectIdGetDatum(myrelid),
1499                                                            Int16GetDatum(myattnum),
1500                                                            0, 0);
1501         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
1502                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1503                          myattnum, myrelid);
1504
1505         ((Form_pg_attribute) GETSTRUCT(tuple))->atthasdef = false;
1506
1507         simple_heap_update(attr_rel, &tuple->t_self, tuple);
1508
1509         /* keep the system catalog indexes current */
1510         CatalogUpdateIndexes(attr_rel, tuple);
1511
1512         /*
1513          * Our update of the pg_attribute row will force a relcache rebuild, so
1514          * there's nothing else to do here.
1515          */
1516         heap_close(attr_rel, RowExclusiveLock);
1517
1518         /* Keep lock on attribute's rel until end of xact */
1519         relation_close(myrel, NoLock);
1520 }
1521
1522 /*
1523  * heap_drop_with_catalog       - removes specified relation from catalogs
1524  *
1525  * Note that this routine is not responsible for dropping objects that are
1526  * linked to the pg_class entry via dependencies (for example, indexes and
1527  * constraints).  Those are deleted by the dependency-tracing logic in
1528  * dependency.c before control gets here.  In general, therefore, this routine
1529  * should never be called directly; go through performDeletion() instead.
1530  */
1531 void
1532 heap_drop_with_catalog(Oid relid)
1533 {
1534         Relation        rel;
1535
1536         /*
1537          * Open and lock the relation.
1538          */
1539         rel = relation_open(relid, AccessExclusiveLock);
1540
1541         /*
1542          * There can no longer be anyone *else* touching the relation, but we
1543          * might still have open queries or cursors in our own session.
1544          */
1545         if (rel->rd_refcnt != 1)
1546                 ereport(ERROR,
1547                                 (errcode(ERRCODE_OBJECT_IN_USE),
1548                                  errmsg("cannot drop \"%s\" because "
1549                                                 "it is being used by active queries in this session",
1550                                                 RelationGetRelationName(rel))));
1551
1552         /*
1553          * Schedule unlinking of the relation's physical files at commit.
1554          */
1555         if (rel->rd_rel->relkind != RELKIND_VIEW &&
1556                 rel->rd_rel->relkind != RELKIND_COMPOSITE_TYPE)
1557         {
1558                 RelationDropStorage(rel);
1559         }
1560
1561         /*
1562          * Close relcache entry, but *keep* AccessExclusiveLock on the relation
1563          * until transaction commit.  This ensures no one else will try to do
1564          * something with the doomed relation.
1565          */
1566         relation_close(rel, NoLock);
1567
1568         /*
1569          * Forget any ON COMMIT action for the rel
1570          */
1571         remove_on_commit_action(relid);
1572
1573         /*
1574          * Flush the relation from the relcache.  We want to do this before
1575          * starting to remove catalog entries, just to be certain that no relcache
1576          * entry rebuild will happen partway through.  (That should not really
1577          * matter, since we don't do CommandCounterIncrement here, but let's be
1578          * safe.)
1579          */
1580         RelationForgetRelation(relid);
1581
1582         /*
1583          * remove inheritance information
1584          */
1585         RelationRemoveInheritance(relid);
1586
1587         /*
1588          * delete statistics
1589          */
1590         RemoveStatistics(relid, 0);
1591
1592         /*
1593          * delete attribute tuples
1594          */
1595         DeleteAttributeTuples(relid);
1596
1597         /*
1598          * delete relation tuple
1599          */
1600         DeleteRelationTuple(relid);
1601 }
1602
1603
1604 /*
1605  * Store a default expression for column attnum of relation rel.
1606  */
1607 void
1608 StoreAttrDefault(Relation rel, AttrNumber attnum, Node *expr)
1609 {
1610         char       *adbin;
1611         char       *adsrc;
1612         Relation        adrel;
1613         HeapTuple       tuple;
1614         Datum           values[4];
1615         static bool nulls[4] = {false, false, false, false};
1616         Relation        attrrel;
1617         HeapTuple       atttup;
1618         Form_pg_attribute attStruct;
1619         Oid                     attrdefOid;
1620         ObjectAddress colobject,
1621                                 defobject;
1622
1623         /*
1624          * Flatten expression to string form for storage.
1625          */
1626         adbin = nodeToString(expr);
1627
1628         /*
1629          * Also deparse it to form the mostly-obsolete adsrc field.
1630          */
1631         adsrc = deparse_expression(expr,
1632                                                         deparse_context_for(RelationGetRelationName(rel),
1633                                                                                                 RelationGetRelid(rel)),
1634                                                            false, false);
1635
1636         /*
1637          * Make the pg_attrdef entry.
1638          */
1639         values[Anum_pg_attrdef_adrelid - 1] = RelationGetRelid(rel);
1640         values[Anum_pg_attrdef_adnum - 1] = attnum;
1641         values[Anum_pg_attrdef_adbin - 1] = CStringGetTextDatum(adbin);
1642         values[Anum_pg_attrdef_adsrc - 1] = CStringGetTextDatum(adsrc);
1643
1644         adrel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1645
1646         tuple = heap_form_tuple(adrel->rd_att, values, nulls);
1647         attrdefOid = simple_heap_insert(adrel, tuple);
1648
1649         CatalogUpdateIndexes(adrel, tuple);
1650
1651         defobject.classId = AttrDefaultRelationId;
1652         defobject.objectId = attrdefOid;
1653         defobject.objectSubId = 0;
1654
1655         heap_close(adrel, RowExclusiveLock);
1656
1657         /* now can free some of the stuff allocated above */
1658         pfree(DatumGetPointer(values[Anum_pg_attrdef_adbin - 1]));
1659         pfree(DatumGetPointer(values[Anum_pg_attrdef_adsrc - 1]));
1660         heap_freetuple(tuple);
1661         pfree(adbin);
1662         pfree(adsrc);
1663
1664         /*
1665          * Update the pg_attribute entry for the column to show that a default
1666          * exists.
1667          */
1668         attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
1669         atttup = SearchSysCacheCopy(ATTNUM,
1670                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
1671                                                                 Int16GetDatum(attnum),
1672                                                                 0, 0);
1673         if (!HeapTupleIsValid(atttup))
1674                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1675                          attnum, RelationGetRelid(rel));
1676         attStruct = (Form_pg_attribute) GETSTRUCT(atttup);
1677         if (!attStruct->atthasdef)
1678         {
1679                 attStruct->atthasdef = true;
1680                 simple_heap_update(attrrel, &atttup->t_self, atttup);
1681                 /* keep catalog indexes current */
1682                 CatalogUpdateIndexes(attrrel, atttup);
1683         }
1684         heap_close(attrrel, RowExclusiveLock);
1685         heap_freetuple(atttup);
1686
1687         /*
1688          * Make a dependency so that the pg_attrdef entry goes away if the column
1689          * (or whole table) is deleted.
1690          */
1691         colobject.classId = RelationRelationId;
1692         colobject.objectId = RelationGetRelid(rel);
1693         colobject.objectSubId = attnum;
1694
1695         recordDependencyOn(&defobject, &colobject, DEPENDENCY_AUTO);
1696
1697         /*
1698          * Record dependencies on objects used in the expression, too.
1699          */
1700         recordDependencyOnExpr(&defobject, expr, NIL, DEPENDENCY_NORMAL);
1701 }
1702
1703 /*
1704  * Store a check-constraint expression for the given relation.
1705  *
1706  * Caller is responsible for updating the count of constraints
1707  * in the pg_class entry for the relation.
1708  */
1709 static void
1710 StoreRelCheck(Relation rel, char *ccname, Node *expr,
1711                           bool is_local, int inhcount)
1712 {
1713         char       *ccbin;
1714         char       *ccsrc;
1715         List       *varList;
1716         int                     keycount;
1717         int16      *attNos;
1718
1719         /*
1720          * Flatten expression to string form for storage.
1721          */
1722         ccbin = nodeToString(expr);
1723
1724         /*
1725          * Also deparse it to form the mostly-obsolete consrc field.
1726          */
1727         ccsrc = deparse_expression(expr,
1728                                                         deparse_context_for(RelationGetRelationName(rel),
1729                                                                                                 RelationGetRelid(rel)),
1730                                                            false, false);
1731
1732         /*
1733          * Find columns of rel that are used in expr
1734          *
1735          * NB: pull_var_clause is okay here only because we don't allow subselects
1736          * in check constraints; it would fail to examine the contents of
1737          * subselects.
1738          */
1739         varList = pull_var_clause(expr, PVC_REJECT_PLACEHOLDERS);
1740         keycount = list_length(varList);
1741
1742         if (keycount > 0)
1743         {
1744                 ListCell   *vl;
1745                 int                     i = 0;
1746
1747                 attNos = (int16 *) palloc(keycount * sizeof(int16));
1748                 foreach(vl, varList)
1749                 {
1750                         Var                *var = (Var *) lfirst(vl);
1751                         int                     j;
1752
1753                         for (j = 0; j < i; j++)
1754                                 if (attNos[j] == var->varattno)
1755                                         break;
1756                         if (j == i)
1757                                 attNos[i++] = var->varattno;
1758                 }
1759                 keycount = i;
1760         }
1761         else
1762                 attNos = NULL;
1763
1764         /*
1765          * Create the Check Constraint
1766          */
1767         CreateConstraintEntry(ccname,           /* Constraint Name */
1768                                                   RelationGetNamespace(rel),    /* namespace */
1769                                                   CONSTRAINT_CHECK,             /* Constraint Type */
1770                                                   false,        /* Is Deferrable */
1771                                                   false,        /* Is Deferred */
1772                                                   RelationGetRelid(rel),                /* relation */
1773                                                   attNos,               /* attrs in the constraint */
1774                                                   keycount,             /* # attrs in the constraint */
1775                                                   InvalidOid,   /* not a domain constraint */
1776                                                   InvalidOid,   /* no associated index */
1777                                                   InvalidOid,   /* Foreign key fields */
1778                                                   NULL,
1779                                                   NULL,
1780                                                   NULL,
1781                                                   NULL,
1782                                                   0,
1783                                                   ' ',
1784                                                   ' ',
1785                                                   ' ',
1786                                                   NULL, /* not an exclusion constraint */
1787                                                   expr, /* Tree form of check constraint */
1788                                                   ccbin,        /* Binary form of check constraint */
1789                                                   ccsrc,        /* Source form of check constraint */
1790                                                   is_local,             /* conislocal */
1791                                                   inhcount);    /* coninhcount */
1792
1793         pfree(ccbin);
1794         pfree(ccsrc);
1795 }
1796
1797 /*
1798  * Store defaults and constraints (passed as a list of CookedConstraint).
1799  *
1800  * NOTE: only pre-cooked expressions will be passed this way, which is to
1801  * say expressions inherited from an existing relation.  Newly parsed
1802  * expressions can be added later, by direct calls to StoreAttrDefault
1803  * and StoreRelCheck (see AddRelationNewConstraints()).
1804  */
1805 static void
1806 StoreConstraints(Relation rel, List *cooked_constraints)
1807 {
1808         int                     numchecks = 0;
1809         ListCell   *lc;
1810
1811         if (!cooked_constraints)
1812                 return;                                 /* nothing to do */
1813
1814         /*
1815          * Deparsing of constraint expressions will fail unless the just-created
1816          * pg_attribute tuples for this relation are made visible.      So, bump the
1817          * command counter.  CAUTION: this will cause a relcache entry rebuild.
1818          */
1819         CommandCounterIncrement();
1820
1821         foreach(lc, cooked_constraints)
1822         {
1823                 CookedConstraint *con = (CookedConstraint *) lfirst(lc);
1824
1825                 switch (con->contype)
1826                 {
1827                         case CONSTR_DEFAULT:
1828                                 StoreAttrDefault(rel, con->attnum, con->expr);
1829                                 break;
1830                         case CONSTR_CHECK:
1831                                 StoreRelCheck(rel, con->name, con->expr,
1832                                                           con->is_local, con->inhcount);
1833                                 numchecks++;
1834                                 break;
1835                         default:
1836                                 elog(ERROR, "unrecognized constraint type: %d",
1837                                          (int) con->contype);
1838                 }
1839         }
1840
1841         if (numchecks > 0)
1842                 SetRelationNumChecks(rel, numchecks);
1843 }
1844
1845 /*
1846  * AddRelationNewConstraints
1847  *
1848  * Add new column default expressions and/or constraint check expressions
1849  * to an existing relation.  This is defined to do both for efficiency in
1850  * DefineRelation, but of course you can do just one or the other by passing
1851  * empty lists.
1852  *
1853  * rel: relation to be modified
1854  * newColDefaults: list of RawColumnDefault structures
1855  * newConstraints: list of Constraint nodes
1856  * allow_merge: TRUE if check constraints may be merged with existing ones
1857  * is_local: TRUE if definition is local, FALSE if it's inherited
1858  *
1859  * All entries in newColDefaults will be processed.  Entries in newConstraints
1860  * will be processed only if they are CONSTR_CHECK type.
1861  *
1862  * Returns a list of CookedConstraint nodes that shows the cooked form of
1863  * the default and constraint expressions added to the relation.
1864  *
1865  * NB: caller should have opened rel with AccessExclusiveLock, and should
1866  * hold that lock till end of transaction.      Also, we assume the caller has
1867  * done a CommandCounterIncrement if necessary to make the relation's catalog
1868  * tuples visible.
1869  */
1870 List *
1871 AddRelationNewConstraints(Relation rel,
1872                                                   List *newColDefaults,
1873                                                   List *newConstraints,
1874                                                   bool allow_merge,
1875                                                   bool is_local)
1876 {
1877         List       *cookedConstraints = NIL;
1878         TupleDesc       tupleDesc;
1879         TupleConstr *oldconstr;
1880         int                     numoldchecks;
1881         ParseState *pstate;
1882         RangeTblEntry *rte;
1883         int                     numchecks;
1884         List       *checknames;
1885         ListCell   *cell;
1886         Node       *expr;
1887         CookedConstraint *cooked;
1888
1889         /*
1890          * Get info about existing constraints.
1891          */
1892         tupleDesc = RelationGetDescr(rel);
1893         oldconstr = tupleDesc->constr;
1894         if (oldconstr)
1895                 numoldchecks = oldconstr->num_check;
1896         else
1897                 numoldchecks = 0;
1898
1899         /*
1900          * Create a dummy ParseState and insert the target relation as its sole
1901          * rangetable entry.  We need a ParseState for transformExpr.
1902          */
1903         pstate = make_parsestate(NULL);
1904         rte = addRangeTableEntryForRelation(pstate,
1905                                                                                 rel,
1906                                                                                 NULL,
1907                                                                                 false,
1908                                                                                 true);
1909         addRTEtoQuery(pstate, rte, true, true, true);
1910
1911         /*
1912          * Process column default expressions.
1913          */
1914         foreach(cell, newColDefaults)
1915         {
1916                 RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell);
1917                 Form_pg_attribute atp = rel->rd_att->attrs[colDef->attnum - 1];
1918
1919                 expr = cookDefault(pstate, colDef->raw_default,
1920                                                    atp->atttypid, atp->atttypmod,
1921                                                    NameStr(atp->attname));
1922
1923                 /*
1924                  * If the expression is just a NULL constant, we do not bother to make
1925                  * an explicit pg_attrdef entry, since the default behavior is
1926                  * equivalent.
1927                  *
1928                  * Note a nonobvious property of this test: if the column is of a
1929                  * domain type, what we'll get is not a bare null Const but a
1930                  * CoerceToDomain expr, so we will not discard the default.  This is
1931                  * critical because the column default needs to be retained to
1932                  * override any default that the domain might have.
1933                  */
1934                 if (expr == NULL ||
1935                         (IsA(expr, Const) &&((Const *) expr)->constisnull))
1936                         continue;
1937
1938                 StoreAttrDefault(rel, colDef->attnum, expr);
1939
1940                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
1941                 cooked->contype = CONSTR_DEFAULT;
1942                 cooked->name = NULL;
1943                 cooked->attnum = colDef->attnum;
1944                 cooked->expr = expr;
1945                 cooked->is_local = is_local;
1946                 cooked->inhcount = is_local ? 0 : 1;
1947                 cookedConstraints = lappend(cookedConstraints, cooked);
1948         }
1949
1950         /*
1951          * Process constraint expressions.
1952          */
1953         numchecks = numoldchecks;
1954         checknames = NIL;
1955         foreach(cell, newConstraints)
1956         {
1957                 Constraint *cdef = (Constraint *) lfirst(cell);
1958                 char       *ccname;
1959
1960                 if (cdef->contype != CONSTR_CHECK)
1961                         continue;
1962
1963                 if (cdef->raw_expr != NULL)
1964                 {
1965                         Assert(cdef->cooked_expr == NULL);
1966
1967                         /*
1968                          * Transform raw parsetree to executable expression, and verify
1969                          * it's valid as a CHECK constraint.
1970                          */
1971                         expr = cookConstraint(pstate, cdef->raw_expr,
1972                                                                   RelationGetRelationName(rel));
1973                 }
1974                 else
1975                 {
1976                         Assert(cdef->cooked_expr != NULL);
1977
1978                         /*
1979                          * Here, we assume the parser will only pass us valid CHECK
1980                          * expressions, so we do no particular checking.
1981                          */
1982                         expr = stringToNode(cdef->cooked_expr);
1983                 }
1984
1985                 /*
1986                  * Check name uniqueness, or generate a name if none was given.
1987                  */
1988                 if (cdef->conname != NULL)
1989                 {
1990                         ListCell   *cell2;
1991
1992                         ccname = cdef->conname;
1993                         /* Check against other new constraints */
1994                         /* Needed because we don't do CommandCounterIncrement in loop */
1995                         foreach(cell2, checknames)
1996                         {
1997                                 if (strcmp((char *) lfirst(cell2), ccname) == 0)
1998                                         ereport(ERROR,
1999                                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
2000                                                          errmsg("check constraint \"%s\" already exists",
2001                                                                         ccname)));
2002                         }
2003
2004                         /* save name for future checks */
2005                         checknames = lappend(checknames, ccname);
2006
2007                         /*
2008                          * Check against pre-existing constraints.      If we are allowed to
2009                          * merge with an existing constraint, there's no more to do here.
2010                          * (We omit the duplicate constraint from the result, which is
2011                          * what ATAddCheckConstraint wants.)
2012                          */
2013                         if (MergeWithExistingConstraint(rel, ccname, expr,
2014                                                                                         allow_merge, is_local))
2015                                 continue;
2016                 }
2017                 else
2018                 {
2019                         /*
2020                          * When generating a name, we want to create "tab_col_check" for a
2021                          * column constraint and "tab_check" for a table constraint.  We
2022                          * no longer have any info about the syntactic positioning of the
2023                          * constraint phrase, so we approximate this by seeing whether the
2024                          * expression references more than one column.  (If the user
2025                          * played by the rules, the result is the same...)
2026                          *
2027                          * Note: pull_var_clause() doesn't descend into sublinks, but we
2028                          * eliminated those above; and anyway this only needs to be an
2029                          * approximate answer.
2030                          */
2031                         List       *vars;
2032                         char       *colname;
2033
2034                         vars = pull_var_clause(expr, PVC_REJECT_PLACEHOLDERS);
2035
2036                         /* eliminate duplicates */
2037                         vars = list_union(NIL, vars);
2038
2039                         if (list_length(vars) == 1)
2040                                 colname = get_attname(RelationGetRelid(rel),
2041                                                                           ((Var *) linitial(vars))->varattno);
2042                         else
2043                                 colname = NULL;
2044
2045                         ccname = ChooseConstraintName(RelationGetRelationName(rel),
2046                                                                                   colname,
2047                                                                                   "check",
2048                                                                                   RelationGetNamespace(rel),
2049                                                                                   checknames);
2050
2051                         /* save name for future checks */
2052                         checknames = lappend(checknames, ccname);
2053                 }
2054
2055                 /*
2056                  * OK, store it.
2057                  */
2058                 StoreRelCheck(rel, ccname, expr, is_local, is_local ? 0 : 1);
2059
2060                 numchecks++;
2061
2062                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
2063                 cooked->contype = CONSTR_CHECK;
2064                 cooked->name = ccname;
2065                 cooked->attnum = 0;
2066                 cooked->expr = expr;
2067                 cooked->is_local = is_local;
2068                 cooked->inhcount = is_local ? 0 : 1;
2069                 cookedConstraints = lappend(cookedConstraints, cooked);
2070         }
2071
2072         /*
2073          * Update the count of constraints in the relation's pg_class tuple. We do
2074          * this even if there was no change, in order to ensure that an SI update
2075          * message is sent out for the pg_class tuple, which will force other
2076          * backends to rebuild their relcache entries for the rel. (This is
2077          * critical if we added defaults but not constraints.)
2078          */
2079         SetRelationNumChecks(rel, numchecks);
2080
2081         return cookedConstraints;
2082 }
2083
2084 /*
2085  * Check for a pre-existing check constraint that conflicts with a proposed
2086  * new one, and either adjust its conislocal/coninhcount settings or throw
2087  * error as needed.
2088  *
2089  * Returns TRUE if merged (constraint is a duplicate), or FALSE if it's
2090  * got a so-far-unique name, or throws error if conflict.
2091  */
2092 static bool
2093 MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
2094                                                         bool allow_merge, bool is_local)
2095 {
2096         bool            found;
2097         Relation        conDesc;
2098         SysScanDesc conscan;
2099         ScanKeyData skey[2];
2100         HeapTuple       tup;
2101
2102         /* Search for a pg_constraint entry with same name and relation */
2103         conDesc = heap_open(ConstraintRelationId, RowExclusiveLock);
2104
2105         found = false;
2106
2107         ScanKeyInit(&skey[0],
2108                                 Anum_pg_constraint_conname,
2109                                 BTEqualStrategyNumber, F_NAMEEQ,
2110                                 CStringGetDatum(ccname));
2111
2112         ScanKeyInit(&skey[1],
2113                                 Anum_pg_constraint_connamespace,
2114                                 BTEqualStrategyNumber, F_OIDEQ,
2115                                 ObjectIdGetDatum(RelationGetNamespace(rel)));
2116
2117         conscan = systable_beginscan(conDesc, ConstraintNameNspIndexId, true,
2118                                                                  SnapshotNow, 2, skey);
2119
2120         while (HeapTupleIsValid(tup = systable_getnext(conscan)))
2121         {
2122                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tup);
2123
2124                 if (con->conrelid == RelationGetRelid(rel))
2125                 {
2126                         /* Found it.  Conflicts if not identical check constraint */
2127                         if (con->contype == CONSTRAINT_CHECK)
2128                         {
2129                                 Datum           val;
2130                                 bool            isnull;
2131
2132                                 val = fastgetattr(tup,
2133                                                                   Anum_pg_constraint_conbin,
2134                                                                   conDesc->rd_att, &isnull);
2135                                 if (isnull)
2136                                         elog(ERROR, "null conbin for rel %s",
2137                                                  RelationGetRelationName(rel));
2138                                 if (equal(expr, stringToNode(TextDatumGetCString(val))))
2139                                         found = true;
2140                         }
2141                         if (!found || !allow_merge)
2142                                 ereport(ERROR,
2143                                                 (errcode(ERRCODE_DUPLICATE_OBJECT),
2144                                 errmsg("constraint \"%s\" for relation \"%s\" already exists",
2145                                            ccname, RelationGetRelationName(rel))));
2146                         /* OK to update the tuple */
2147                         ereport(NOTICE,
2148                            (errmsg("merging constraint \"%s\" with inherited definition",
2149                                            ccname)));
2150                         tup = heap_copytuple(tup);
2151                         con = (Form_pg_constraint) GETSTRUCT(tup);
2152                         if (is_local)
2153                                 con->conislocal = true;
2154                         else
2155                                 con->coninhcount++;
2156                         simple_heap_update(conDesc, &tup->t_self, tup);
2157                         CatalogUpdateIndexes(conDesc, tup);
2158                         break;
2159                 }
2160         }
2161
2162         systable_endscan(conscan);
2163         heap_close(conDesc, RowExclusiveLock);
2164
2165         return found;
2166 }
2167
2168 /*
2169  * Update the count of constraints in the relation's pg_class tuple.
2170  *
2171  * Caller had better hold exclusive lock on the relation.
2172  *
2173  * An important side effect is that a SI update message will be sent out for
2174  * the pg_class tuple, which will force other backends to rebuild their
2175  * relcache entries for the rel.  Also, this backend will rebuild its
2176  * own relcache entry at the next CommandCounterIncrement.
2177  */
2178 static void
2179 SetRelationNumChecks(Relation rel, int numchecks)
2180 {
2181         Relation        relrel;
2182         HeapTuple       reltup;
2183         Form_pg_class relStruct;
2184
2185         relrel = heap_open(RelationRelationId, RowExclusiveLock);
2186         reltup = SearchSysCacheCopy(RELOID,
2187                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
2188                                                                 0, 0, 0);
2189         if (!HeapTupleIsValid(reltup))
2190                 elog(ERROR, "cache lookup failed for relation %u",
2191                          RelationGetRelid(rel));
2192         relStruct = (Form_pg_class) GETSTRUCT(reltup);
2193
2194         if (relStruct->relchecks != numchecks)
2195         {
2196                 relStruct->relchecks = numchecks;
2197
2198                 simple_heap_update(relrel, &reltup->t_self, reltup);
2199
2200                 /* keep catalog indexes current */
2201                 CatalogUpdateIndexes(relrel, reltup);
2202         }
2203         else
2204         {
2205                 /* Skip the disk update, but force relcache inval anyway */
2206                 CacheInvalidateRelcache(rel);
2207         }
2208
2209         heap_freetuple(reltup);
2210         heap_close(relrel, RowExclusiveLock);
2211 }
2212
2213 /*
2214  * Take a raw default and convert it to a cooked format ready for
2215  * storage.
2216  *
2217  * Parse state should be set up to recognize any vars that might appear
2218  * in the expression.  (Even though we plan to reject vars, it's more
2219  * user-friendly to give the correct error message than "unknown var".)
2220  *
2221  * If atttypid is not InvalidOid, coerce the expression to the specified
2222  * type (and typmod atttypmod).   attname is only needed in this case:
2223  * it is used in the error message, if any.
2224  */
2225 Node *
2226 cookDefault(ParseState *pstate,
2227                         Node *raw_default,
2228                         Oid atttypid,
2229                         int32 atttypmod,
2230                         char *attname)
2231 {
2232         Node       *expr;
2233
2234         Assert(raw_default != NULL);
2235
2236         /*
2237          * Transform raw parsetree to executable expression.
2238          */
2239         expr = transformExpr(pstate, raw_default);
2240
2241         /*
2242          * Make sure default expr does not refer to any vars.
2243          */
2244         if (contain_var_clause(expr))
2245                 ereport(ERROR,
2246                                 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
2247                           errmsg("cannot use column references in default expression")));
2248
2249         /*
2250          * It can't return a set either.
2251          */
2252         if (expression_returns_set(expr))
2253                 ereport(ERROR,
2254                                 (errcode(ERRCODE_DATATYPE_MISMATCH),
2255                                  errmsg("default expression must not return a set")));
2256
2257         /*
2258          * No subplans or aggregates, either...
2259          */
2260         if (pstate->p_hasSubLinks)
2261                 ereport(ERROR,
2262                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2263                                  errmsg("cannot use subquery in default expression")));
2264         if (pstate->p_hasAggs)
2265                 ereport(ERROR,
2266                                 (errcode(ERRCODE_GROUPING_ERROR),
2267                          errmsg("cannot use aggregate function in default expression")));
2268         if (pstate->p_hasWindowFuncs)
2269                 ereport(ERROR,
2270                                 (errcode(ERRCODE_WINDOWING_ERROR),
2271                                  errmsg("cannot use window function in default expression")));
2272
2273         /*
2274          * Coerce the expression to the correct type and typmod, if given. This
2275          * should match the parser's processing of non-defaulted expressions ---
2276          * see transformAssignedExpr().
2277          */
2278         if (OidIsValid(atttypid))
2279         {
2280                 Oid                     type_id = exprType(expr);
2281
2282                 expr = coerce_to_target_type(pstate, expr, type_id,
2283                                                                          atttypid, atttypmod,
2284                                                                          COERCION_ASSIGNMENT,
2285                                                                          COERCE_IMPLICIT_CAST,
2286                                                                          -1);
2287                 if (expr == NULL)
2288                         ereport(ERROR,
2289                                         (errcode(ERRCODE_DATATYPE_MISMATCH),
2290                                          errmsg("column \"%s\" is of type %s"
2291                                                         " but default expression is of type %s",
2292                                                         attname,
2293                                                         format_type_be(atttypid),
2294                                                         format_type_be(type_id)),
2295                            errhint("You will need to rewrite or cast the expression.")));
2296         }
2297
2298         return expr;
2299 }
2300
2301 /*
2302  * Take a raw CHECK constraint expression and convert it to a cooked format
2303  * ready for storage.
2304  *
2305  * Parse state must be set up to recognize any vars that might appear
2306  * in the expression.
2307  */
2308 static Node *
2309 cookConstraint(ParseState *pstate,
2310                            Node *raw_constraint,
2311                            char *relname)
2312 {
2313         Node       *expr;
2314
2315         /*
2316          * Transform raw parsetree to executable expression.
2317          */
2318         expr = transformExpr(pstate, raw_constraint);
2319
2320         /*
2321          * Make sure it yields a boolean result.
2322          */
2323         expr = coerce_to_boolean(pstate, expr, "CHECK");
2324
2325         /*
2326          * Make sure no outside relations are referred to.
2327          */
2328         if (list_length(pstate->p_rtable) != 1)
2329                 ereport(ERROR,
2330                                 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
2331                         errmsg("only table \"%s\" can be referenced in check constraint",
2332                                    relname)));
2333
2334         /*
2335          * No subplans or aggregates, either...
2336          */
2337         if (pstate->p_hasSubLinks)
2338                 ereport(ERROR,
2339                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2340                                  errmsg("cannot use subquery in check constraint")));
2341         if (pstate->p_hasAggs)
2342                 ereport(ERROR,
2343                                 (errcode(ERRCODE_GROUPING_ERROR),
2344                            errmsg("cannot use aggregate function in check constraint")));
2345         if (pstate->p_hasWindowFuncs)
2346                 ereport(ERROR,
2347                                 (errcode(ERRCODE_WINDOWING_ERROR),
2348                                  errmsg("cannot use window function in check constraint")));
2349
2350         return expr;
2351 }
2352
2353
2354 /*
2355  * RemoveStatistics --- remove entries in pg_statistic for a rel or column
2356  *
2357  * If attnum is zero, remove all entries for rel; else remove only the one(s)
2358  * for that column.
2359  */
2360 void
2361 RemoveStatistics(Oid relid, AttrNumber attnum)
2362 {
2363         Relation        pgstatistic;
2364         SysScanDesc scan;
2365         ScanKeyData key[2];
2366         int                     nkeys;
2367         HeapTuple       tuple;
2368
2369         pgstatistic = heap_open(StatisticRelationId, RowExclusiveLock);
2370
2371         ScanKeyInit(&key[0],
2372                                 Anum_pg_statistic_starelid,
2373                                 BTEqualStrategyNumber, F_OIDEQ,
2374                                 ObjectIdGetDatum(relid));
2375
2376         if (attnum == 0)
2377                 nkeys = 1;
2378         else
2379         {
2380                 ScanKeyInit(&key[1],
2381                                         Anum_pg_statistic_staattnum,
2382                                         BTEqualStrategyNumber, F_INT2EQ,
2383                                         Int16GetDatum(attnum));
2384                 nkeys = 2;
2385         }
2386
2387         scan = systable_beginscan(pgstatistic, StatisticRelidAttnumInhIndexId, true,
2388                                                           SnapshotNow, nkeys, key);
2389
2390         /* we must loop even when attnum != 0, in case of inherited stats */
2391         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
2392                 simple_heap_delete(pgstatistic, &tuple->t_self);
2393
2394         systable_endscan(scan);
2395
2396         heap_close(pgstatistic, RowExclusiveLock);
2397 }
2398
2399
2400 /*
2401  * RelationTruncateIndexes - truncate all indexes associated
2402  * with the heap relation to zero tuples.
2403  *
2404  * The routine will truncate and then reconstruct the indexes on
2405  * the specified relation.      Caller must hold exclusive lock on rel.
2406  */
2407 static void
2408 RelationTruncateIndexes(Relation heapRelation)
2409 {
2410         ListCell   *indlist;
2411
2412         /* Ask the relcache to produce a list of the indexes of the rel */
2413         foreach(indlist, RelationGetIndexList(heapRelation))
2414         {
2415                 Oid                     indexId = lfirst_oid(indlist);
2416                 Relation        currentIndex;
2417                 IndexInfo  *indexInfo;
2418
2419                 /* Open the index relation; use exclusive lock, just to be sure */
2420                 currentIndex = index_open(indexId, AccessExclusiveLock);
2421
2422                 /* Fetch info needed for index_build */
2423                 indexInfo = BuildIndexInfo(currentIndex);
2424
2425                 /*
2426                  * Now truncate the actual file (and discard buffers).
2427                  */
2428                 RelationTruncate(currentIndex, 0);
2429
2430                 /* Initialize the index and rebuild */
2431                 /* Note: we do not need to re-establish pkey setting */
2432                 index_build(heapRelation, currentIndex, indexInfo, false);
2433
2434                 /* We're done with this index */
2435                 index_close(currentIndex, NoLock);
2436         }
2437 }
2438
2439 /*
2440  *       heap_truncate
2441  *
2442  *       This routine deletes all data within all the specified relations.
2443  *
2444  * This is not transaction-safe!  There is another, transaction-safe
2445  * implementation in commands/tablecmds.c.      We now use this only for
2446  * ON COMMIT truncation of temporary tables, where it doesn't matter.
2447  */
2448 void
2449 heap_truncate(List *relids)
2450 {
2451         List       *relations = NIL;
2452         ListCell   *cell;
2453
2454         /* Open relations for processing, and grab exclusive access on each */
2455         foreach(cell, relids)
2456         {
2457                 Oid                     rid = lfirst_oid(cell);
2458                 Relation        rel;
2459
2460                 rel = heap_open(rid, AccessExclusiveLock);
2461                 relations = lappend(relations, rel);
2462         }
2463
2464         /* Don't allow truncate on tables that are referenced by foreign keys */
2465         heap_truncate_check_FKs(relations, true);
2466
2467         /* OK to do it */
2468         foreach(cell, relations)
2469         {
2470                 Relation        rel = lfirst(cell);
2471
2472                 /* Truncate the relation */
2473                 heap_truncate_one_rel(rel);
2474
2475                 /* Close the relation, but keep exclusive lock on it until commit */
2476                 heap_close(rel, NoLock);
2477         }
2478 }
2479
2480 /*
2481  *       heap_truncate_one_rel
2482  *
2483  *       This routine deletes all data within the specified relation.
2484  *
2485  * This is not transaction-safe, because the truncation is done immediately
2486  * and cannot be rolled back later.  Caller is responsible for having
2487  * checked permissions etc, and must have obtained AccessExclusiveLock.
2488  */
2489 void
2490 heap_truncate_one_rel(Relation rel)
2491 {
2492         Oid                     toastrelid;
2493
2494         /* Truncate the actual file (and discard buffers) */
2495         RelationTruncate(rel, 0);
2496
2497         /* If the relation has indexes, truncate the indexes too */
2498         RelationTruncateIndexes(rel);
2499
2500         /* If there is a toast table, truncate that too */
2501         toastrelid = rel->rd_rel->reltoastrelid;
2502         if (OidIsValid(toastrelid))
2503         {
2504                 Relation        toastrel = heap_open(toastrelid, AccessExclusiveLock);
2505
2506                 RelationTruncate(toastrel, 0);
2507                 RelationTruncateIndexes(toastrel);
2508                 /* keep the lock... */
2509                 heap_close(toastrel, NoLock);
2510         }
2511 }
2512
2513 /*
2514  * heap_truncate_check_FKs
2515  *              Check for foreign keys referencing a list of relations that
2516  *              are to be truncated, and raise error if there are any
2517  *
2518  * We disallow such FKs (except self-referential ones) since the whole point
2519  * of TRUNCATE is to not scan the individual rows to be thrown away.
2520  *
2521  * This is split out so it can be shared by both implementations of truncate.
2522  * Caller should already hold a suitable lock on the relations.
2523  *
2524  * tempTables is only used to select an appropriate error message.
2525  */
2526 void
2527 heap_truncate_check_FKs(List *relations, bool tempTables)
2528 {
2529         List       *oids = NIL;
2530         List       *dependents;
2531         ListCell   *cell;
2532
2533         /*
2534          * Build a list of OIDs of the interesting relations.
2535          *
2536          * If a relation has no triggers, then it can neither have FKs nor be
2537          * referenced by a FK from another table, so we can ignore it.
2538          */
2539         foreach(cell, relations)
2540         {
2541                 Relation        rel = lfirst(cell);
2542
2543                 if (rel->rd_rel->relhastriggers)
2544                         oids = lappend_oid(oids, RelationGetRelid(rel));
2545         }
2546
2547         /*
2548          * Fast path: if no relation has triggers, none has FKs either.
2549          */
2550         if (oids == NIL)
2551                 return;
2552
2553         /*
2554          * Otherwise, must scan pg_constraint.  We make one pass with all the
2555          * relations considered; if this finds nothing, then all is well.
2556          */
2557         dependents = heap_truncate_find_FKs(oids);
2558         if (dependents == NIL)
2559                 return;
2560
2561         /*
2562          * Otherwise we repeat the scan once per relation to identify a particular
2563          * pair of relations to complain about.  This is pretty slow, but
2564          * performance shouldn't matter much in a failure path.  The reason for
2565          * doing things this way is to ensure that the message produced is not
2566          * dependent on chance row locations within pg_constraint.
2567          */
2568         foreach(cell, oids)
2569         {
2570                 Oid                     relid = lfirst_oid(cell);
2571                 ListCell   *cell2;
2572
2573                 dependents = heap_truncate_find_FKs(list_make1_oid(relid));
2574
2575                 foreach(cell2, dependents)
2576                 {
2577                         Oid                     relid2 = lfirst_oid(cell2);
2578
2579                         if (!list_member_oid(oids, relid2))
2580                         {
2581                                 char       *relname = get_rel_name(relid);
2582                                 char       *relname2 = get_rel_name(relid2);
2583
2584                                 if (tempTables)
2585                                         ereport(ERROR,
2586                                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2587                                                          errmsg("unsupported ON COMMIT and foreign key combination"),
2588                                                          errdetail("Table \"%s\" references \"%s\", but they do not have the same ON COMMIT setting.",
2589                                                                            relname2, relname)));
2590                                 else
2591                                         ereport(ERROR,
2592                                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2593                                                          errmsg("cannot truncate a table referenced in a foreign key constraint"),
2594                                                          errdetail("Table \"%s\" references \"%s\".",
2595                                                                            relname2, relname),
2596                                                    errhint("Truncate table \"%s\" at the same time, "
2597                                                                    "or use TRUNCATE ... CASCADE.",
2598                                                                    relname2)));
2599                         }
2600                 }
2601         }
2602 }
2603
2604 /*
2605  * heap_truncate_find_FKs
2606  *              Find relations having foreign keys referencing any of the given rels
2607  *
2608  * Input and result are both lists of relation OIDs.  The result contains
2609  * no duplicates, does *not* include any rels that were already in the input
2610  * list, and is sorted in OID order.  (The last property is enforced mainly
2611  * to guarantee consistent behavior in the regression tests; we don't want
2612  * behavior to change depending on chance locations of rows in pg_constraint.)
2613  *
2614  * Note: caller should already have appropriate lock on all rels mentioned
2615  * in relationIds.      Since adding or dropping an FK requires exclusive lock
2616  * on both rels, this ensures that the answer will be stable.
2617  */
2618 List *
2619 heap_truncate_find_FKs(List *relationIds)
2620 {
2621         List       *result = NIL;
2622         Relation        fkeyRel;
2623         SysScanDesc fkeyScan;
2624         HeapTuple       tuple;
2625
2626         /*
2627          * Must scan pg_constraint.  Right now, it is a seqscan because there is
2628          * no available index on confrelid.
2629          */
2630         fkeyRel = heap_open(ConstraintRelationId, AccessShareLock);
2631
2632         fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
2633                                                                   SnapshotNow, 0, NULL);
2634
2635         while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
2636         {
2637                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
2638
2639                 /* Not a foreign key */
2640                 if (con->contype != CONSTRAINT_FOREIGN)
2641                         continue;
2642
2643                 /* Not referencing one of our list of tables */
2644                 if (!list_member_oid(relationIds, con->confrelid))
2645                         continue;
2646
2647                 /* Add referencer unless already in input or result list */
2648                 if (!list_member_oid(relationIds, con->conrelid))
2649                         result = insert_ordered_unique_oid(result, con->conrelid);
2650         }
2651
2652         systable_endscan(fkeyScan);
2653         heap_close(fkeyRel, AccessShareLock);
2654
2655         return result;
2656 }
2657
2658 /*
2659  * insert_ordered_unique_oid
2660  *              Insert a new Oid into a sorted list of Oids, preserving ordering,
2661  *              and eliminating duplicates
2662  *
2663  * Building the ordered list this way is O(N^2), but with a pretty small
2664  * constant, so for the number of entries we expect it will probably be
2665  * faster than trying to apply qsort().  It seems unlikely someone would be
2666  * trying to truncate a table with thousands of dependent tables ...
2667  */
2668 static List *
2669 insert_ordered_unique_oid(List *list, Oid datum)
2670 {
2671         ListCell   *prev;
2672
2673         /* Does the datum belong at the front? */
2674         if (list == NIL || datum < linitial_oid(list))
2675                 return lcons_oid(datum, list);
2676         /* Does it match the first entry? */
2677         if (datum == linitial_oid(list))
2678                 return list;                    /* duplicate, so don't insert */
2679         /* No, so find the entry it belongs after */
2680         prev = list_head(list);
2681         for (;;)
2682         {
2683                 ListCell   *curr = lnext(prev);
2684
2685                 if (curr == NULL || datum < lfirst_oid(curr))
2686                         break;                          /* it belongs after 'prev', before 'curr' */
2687
2688                 if (datum == lfirst_oid(curr))
2689                         return list;            /* duplicate, so don't insert */
2690
2691                 prev = curr;
2692         }
2693         /* Insert datum into list after 'prev' */
2694         lappend_cell_oid(list, prev, datum);
2695         return list;
2696 }