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-2006, 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.301 2006/06/27 18:35:05 momjian 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/heapam.h"
  33 #include "access/genam.h"
  34 #include "catalog/catalog.h"
  35 #include "catalog/dependency.h"
  36 #include "catalog/heap.h"
  37 #include "catalog/index.h"
  38 #include "catalog/indexing.h"
  39 #include "catalog/pg_attrdef.h"
  40 #include "catalog/pg_constraint.h"
  41 #include "catalog/pg_inherits.h"
  42 #include "catalog/pg_namespace.h"
  43 #include "catalog/pg_statistic.h"
  44 #include "catalog/pg_type.h"
  45 #include "commands/tablecmds.h"
  46 #include "commands/trigger.h"
  47 #include "miscadmin.h"
  48 #include "nodes/makefuncs.h"
  49 #include "optimizer/clauses.h"
  50 #include "optimizer/planmain.h"
  51 #include "optimizer/var.h"
  52 #include "parser/parse_coerce.h"
  53 #include "parser/parse_expr.h"
  54 #include "parser/parse_relation.h"
  55 #include "rewrite/rewriteRemove.h"
  56 #include "storage/smgr.h"
  57 #include "utils/builtins.h"
  58 #include "utils/fmgroids.h"
  59 #include "utils/inval.h"
  60 #include "utils/lsyscache.h"
  61 #include "utils/relcache.h"
  62 #include "utils/syscache.h"
  63
  64
  65 static void AddNewRelationTuple(Relation pg_class_desc,
  66                                         Relation new_rel_desc,
  67                                         Oid new_rel_oid, Oid new_type_oid,
  68                                         Oid relowner,
  69                                         char relkind);
  70 static Oid AddNewRelationType(const char *typeName,
  71                                    Oid typeNamespace,
  72                                    Oid new_rel_oid,
  73                                    char new_rel_kind);
  74 static void RelationRemoveInheritance(Oid relid);
  75 static void StoreRelCheck(Relation rel, char *ccname, char *ccbin);
  76 static void StoreConstraints(Relation rel, TupleDesc tupdesc);
  77 static void SetRelationNumChecks(Relation rel, int numchecks);
  78
  79
  80 /* ----------------------------------------------------------------
  81  *                              XXX UGLY HARD CODED BADNESS FOLLOWS XXX
  82  *
  83  *              these should all be moved to someplace in the lib/catalog
  84  *              module, if not obliterated first.
  85  * ----------------------------------------------------------------
  86  */
  87
  88
  89 /*
  90  * Note:
  91  *              Should the system special case these attributes in the future?
  92  *              Advantage:      consume much less space in the ATTRIBUTE relation.
  93  *              Disadvantage:  special cases will be all over the place.
  94  */
  95
  96 static FormData_pg_attribute a1 = {
  97         0, {"ctid"}, TIDOID, 0, sizeof(ItemPointerData),
  98         SelfItemPointerAttributeNumber, 0, -1, -1,
  99         false, 'p', 's', true, false, false, true, 0
 100 };
 101
 102 static FormData_pg_attribute a2 = {
 103         0, {"oid"}, OIDOID, 0, sizeof(Oid),
 104         ObjectIdAttributeNumber, 0, -1, -1,
 105         true, 'p', 'i', true, false, false, true, 0
 106 };
 107
 108 static FormData_pg_attribute a3 = {
 109         0, {"xmin"}, XIDOID, 0, sizeof(TransactionId),
 110         MinTransactionIdAttributeNumber, 0, -1, -1,
 111         true, 'p', 'i', true, false, false, true, 0
 112 };
 113
 114 static FormData_pg_attribute a4 = {
 115         0, {"cmin"}, CIDOID, 0, sizeof(CommandId),
 116         MinCommandIdAttributeNumber, 0, -1, -1,
 117         true, 'p', 'i', true, false, false, true, 0
 118 };
 119
 120 static FormData_pg_attribute a5 = {
 121         0, {"xmax"}, XIDOID, 0, sizeof(TransactionId),
 122         MaxTransactionIdAttributeNumber, 0, -1, -1,
 123         true, 'p', 'i', true, false, false, true, 0
 124 };
 125
 126 static FormData_pg_attribute a6 = {
 127         0, {"cmax"}, CIDOID, 0, sizeof(CommandId),
 128         MaxCommandIdAttributeNumber, 0, -1, -1,
 129         true, 'p', 'i', true, false, false, true, 0
 130 };
 131
 132 /*
 133  * We decided to call this attribute "tableoid" rather than say
 134  * "classoid" on the basis that in the future there may be more than one
 135  * table of a particular class/type. In any case table is still the word
 136  * used in SQL.
 137  */
 138 static FormData_pg_attribute a7 = {
 139         0, {"tableoid"}, OIDOID, 0, sizeof(Oid),
 140         TableOidAttributeNumber, 0, -1, -1,
 141         true, 'p', 'i', true, false, false, true, 0
 142 };
 143
 144 static const Form_pg_attribute SysAtt[] = {&a1, &a2, &a3, &a4, &a5, &a6, &a7};
 145
 146 /*
 147  * This function returns a Form_pg_attribute pointer for a system attribute.
 148  * Note that we elog if the presented attno is invalid, which would only
 149  * happen if there's a problem upstream.
 150  */
 151 Form_pg_attribute
 152 SystemAttributeDefinition(AttrNumber attno, bool relhasoids)
 153 {
 154         if (attno >= 0 || attno < -(int) lengthof(SysAtt))
 155                 elog(ERROR, "invalid system attribute number %d", attno);
 156         if (attno == ObjectIdAttributeNumber && !relhasoids)
 157                 elog(ERROR, "invalid system attribute number %d", attno);
 158         return SysAtt[-attno - 1];
 159 }
 160
 161 /*
 162  * If the given name is a system attribute name, return a Form_pg_attribute
 163  * pointer for a prototype definition.  If not, return NULL.
 164  */
 165 Form_pg_attribute
 166 SystemAttributeByName(const char *attname, bool relhasoids)
 167 {
 168         int                     j;
 169
 170         for (j = 0; j < (int) lengthof(SysAtt); j++)
 171         {
 172                 Form_pg_attribute att = SysAtt[j];
 173
 174                 if (relhasoids || att->attnum != ObjectIdAttributeNumber)
 175                 {
 176                         if (strcmp(NameStr(att->attname), attname) == 0)
 177                                 return att;
 178                 }
 179         }
 180
 181         return NULL;
 182 }
 183
 184
 185 /* ----------------------------------------------------------------
 186  *                              XXX END OF UGLY HARD CODED BADNESS XXX
 187  * ---------------------------------------------------------------- */
 188
 189
 190 /* ----------------------------------------------------------------
 191  *              heap_create             - Create an uncataloged heap relation
 192  *
 193  *              Note API change: the caller must now always provide the OID
 194  *              to use for the relation.
 195  *
 196  *              rel->rd_rel is initialized by RelationBuildLocalRelation,
 197  *              and is mostly zeroes at return.
 198  * ----------------------------------------------------------------
 199  */
 200 Relation
 201 heap_create(const char *relname,
 202                         Oid relnamespace,
 203                         Oid reltablespace,
 204                         Oid relid,
 205                         TupleDesc tupDesc,
 206                         char relkind,
 207                         bool shared_relation,
 208                         bool allow_system_table_mods)
 209 {
 210         bool            create_storage;
 211         Relation        rel;
 212
 213         /* The caller must have provided an OID for the relation. */
 214         Assert(OidIsValid(relid));
 215
 216         /*
 217          * sanity checks
 218          */
 219         if (!allow_system_table_mods &&
 220                 (IsSystemNamespace(relnamespace) || IsToastNamespace(relnamespace)) &&
 221                 IsNormalProcessingMode())
 222                 ereport(ERROR,
 223                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 224                                  errmsg("permission denied to create \"%s.%s\"",
 225                                                 get_namespace_name(relnamespace), relname),
 226                 errdetail("System catalog modifications are currently disallowed.")));
 227
 228         /*
 229          * Decide if we need storage or not, and handle a couple other special
 230          * cases for particular relkinds.
 231          */
 232         switch (relkind)
 233         {
 234                 case RELKIND_VIEW:
 235                 case RELKIND_COMPOSITE_TYPE:
 236                         create_storage = false;
 237
 238                         /*
 239                          * Force reltablespace to zero if the relation has no physical
 240                          * storage.  This is mainly just for cleanliness' sake.
 241                          */
 242                         reltablespace = InvalidOid;
 243                         break;
 244                 case RELKIND_SEQUENCE:
 245                         create_storage = true;
 246
 247                         /*
 248                          * Force reltablespace to zero for sequences, since we don't
 249                          * support moving them around into different tablespaces.
 250                          */
 251                         reltablespace = InvalidOid;
 252                         break;
 253                 default:
 254                         create_storage = true;
 255                         break;
 256         }
 257
 258         /*
 259          * Never allow a pg_class entry to explicitly specify the database's
 260          * default tablespace in reltablespace; force it to zero instead. This
 261          * ensures that if the database is cloned with a different default
 262          * tablespace, the pg_class entry will still match where CREATE DATABASE
 263          * will put the physically copied relation.
 264          *
 265          * Yes, this is a bit of a hack.
 266          */
 267         if (reltablespace == MyDatabaseTableSpace)
 268                 reltablespace = InvalidOid;
 269
 270         /*
 271          * build the relcache entry.
 272          */
 273         rel = RelationBuildLocalRelation(relname,
 274                                                                          relnamespace,
 275                                                                          tupDesc,
 276                                                                          relid,
 277                                                                          reltablespace,
 278                                                                          shared_relation);
 279
 280         /*
 281          * have the storage manager create the relation's disk file, if needed.
 282          */
 283         if (create_storage)
 284         {
 285                 Assert(rel->rd_smgr == NULL);
 286                 RelationOpenSmgr(rel);
 287                 smgrcreate(rel->rd_smgr, rel->rd_istemp, false);
 288         }
 289
 290         return rel;
 291 }
 292
 293 /* ----------------------------------------------------------------
 294  *              heap_create_with_catalog                - Create a cataloged relation
 295  *
 296  *              this is done in multiple steps:
 297  *
 298  *              1) CheckAttributeNamesTypes() is used to make certain the tuple
 299  *                 descriptor contains a valid set of attribute names and types
 300  *
 301  *              2) pg_class is opened and get_relname_relid()
 302  *                 performs a scan to ensure that no relation with the
 303  *                 same name already exists.
 304  *
 305  *              3) heap_create() is called to create the new relation on disk.
 306  *
 307  *              4) TypeCreate() is called to define a new type corresponding
 308  *                 to the new relation.
 309  *
 310  *              5) AddNewRelationTuple() is called to register the
 311  *                 relation in pg_class.
 312  *
 313  *              6) AddNewAttributeTuples() is called to register the
 314  *                 new relation's schema in pg_attribute.
 315  *
 316  *              7) StoreConstraints is called ()                - vadim 08/22/97
 317  *
 318  *              8) the relations are closed and the new relation's oid
 319  *                 is returned.
 320  *
 321  * ----------------------------------------------------------------
 322  */
 323
 324 /* --------------------------------
 325  *              CheckAttributeNamesTypes
 326  *
 327  *              this is used to make certain the tuple descriptor contains a
 328  *              valid set of attribute names and datatypes.  a problem simply
 329  *              generates ereport(ERROR) which aborts the current transaction.
 330  * --------------------------------
 331  */
 332 void
 333 CheckAttributeNamesTypes(TupleDesc tupdesc, char relkind)
 334 {
 335         int                     i;
 336         int                     j;
 337         int                     natts = tupdesc->natts;
 338
 339         /* Sanity check on column count */
 340         if (natts < 0 || natts > MaxHeapAttributeNumber)
 341                 ereport(ERROR,
 342                                 (errcode(ERRCODE_TOO_MANY_COLUMNS),
 343                                  errmsg("tables can have at most %d columns",
 344                                                 MaxHeapAttributeNumber)));
 345
 346         /*
 347          * first check for collision with system attribute names
 348          *
 349          * Skip this for a view or type relation, since those don't have system
 350          * attributes.
 351          */
 352         if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
 353         {
 354                 for (i = 0; i < natts; i++)
 355                 {
 356                         if (SystemAttributeByName(NameStr(tupdesc->attrs[i]->attname),
 357                                                                           tupdesc->tdhasoid) != NULL)
 358                                 ereport(ERROR,
 359                                                 (errcode(ERRCODE_DUPLICATE_COLUMN),
 360                                                  errmsg("column name \"%s\" conflicts with a system column name",
 361                                                                 NameStr(tupdesc->attrs[i]->attname))));
 362                 }
 363         }
 364
 365         /*
 366          * next check for repeated attribute names
 367          */
 368         for (i = 1; i < natts; i++)
 369         {
 370                 for (j = 0; j < i; j++)
 371                 {
 372                         if (strcmp(NameStr(tupdesc->attrs[j]->attname),
 373                                            NameStr(tupdesc->attrs[i]->attname)) == 0)
 374                                 ereport(ERROR,
 375                                                 (errcode(ERRCODE_DUPLICATE_COLUMN),
 376                                                  errmsg("column name \"%s\" is duplicated",
 377                                                                 NameStr(tupdesc->attrs[j]->attname))));
 378                 }
 379         }
 380
 381         /*
 382          * next check the attribute types
 383          */
 384         for (i = 0; i < natts; i++)
 385         {
 386                 CheckAttributeType(NameStr(tupdesc->attrs[i]->attname),
 387                                                    tupdesc->attrs[i]->atttypid);
 388         }
 389 }
 390
 391 /* --------------------------------
 392  *              CheckAttributeType
 393  *
 394  *              Verify that the proposed datatype of an attribute is legal.
 395  *              This is needed because there are types (and pseudo-types)
 396  *              in the catalogs that we do not support as elements of real tuples.
 397  * --------------------------------
 398  */
 399 void
 400 CheckAttributeType(const char *attname, Oid atttypid)
 401 {
 402         char            att_typtype = get_typtype(atttypid);
 403
 404         /*
 405          * Warn user, but don't fail, if column to be created has UNKNOWN type
 406          * (usually as a result of a 'retrieve into' - jolly)
 407          *
 408          * Refuse any attempt to create a pseudo-type column.
 409          */
 410         if (atttypid == UNKNOWNOID)
 411                 ereport(WARNING,
 412                                 (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 413                                  errmsg("column \"%s\" has type \"unknown\"", attname),
 414                                  errdetail("Proceeding with relation creation anyway.")));
 415         else if (att_typtype == 'p')
 416         {
 417                 /* Special hack for pg_statistic: allow ANYARRAY during initdb */
 418                 if (atttypid != ANYARRAYOID || IsUnderPostmaster)
 419                         ereport(ERROR,
 420                                         (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 421                                          errmsg("column \"%s\" has pseudo-type %s",
 422                                                         attname, format_type_be(atttypid))));
 423         }
 424 }
 425
 426 /* --------------------------------
 427  *              AddNewAttributeTuples
 428  *
 429  *              this registers the new relation's schema by adding
 430  *              tuples to pg_attribute.
 431  * --------------------------------
 432  */
 433 static void
 434 AddNewAttributeTuples(Oid new_rel_oid,
 435                                           TupleDesc tupdesc,
 436                                           char relkind,
 437                                           bool oidislocal,
 438                                           int oidinhcount)
 439 {
 440         const Form_pg_attribute *dpp;
 441         int                     i;
 442         HeapTuple       tup;
 443         Relation        rel;
 444         CatalogIndexState indstate;
 445         int                     natts = tupdesc->natts;
 446         ObjectAddress myself,
 447                                 referenced;
 448
 449         /*
 450          * open pg_attribute and its indexes.
 451          */
 452         rel = heap_open(AttributeRelationId, RowExclusiveLock);
 453
 454         indstate = CatalogOpenIndexes(rel);
 455
 456         /*
 457          * First we add the user attributes.  This is also a convenient place to
 458          * add dependencies on their datatypes.
 459          */
 460         dpp = tupdesc->attrs;
 461         for (i = 0; i < natts; i++)
 462         {
 463                 /* Fill in the correct relation OID */
 464                 (*dpp)->attrelid = new_rel_oid;
 465                 /* Make sure these are OK, too */
 466                 (*dpp)->attstattarget = -1;
 467                 (*dpp)->attcacheoff = -1;
 468
 469                 tup = heap_addheader(Natts_pg_attribute,
 470                                                          false,
 471                                                          ATTRIBUTE_TUPLE_SIZE,
 472                                                          (void *) *dpp);
 473
 474                 simple_heap_insert(rel, tup);
 475
 476                 CatalogIndexInsert(indstate, tup);
 477
 478                 heap_freetuple(tup);
 479
 480                 myself.classId = RelationRelationId;
 481                 myself.objectId = new_rel_oid;
 482                 myself.objectSubId = i + 1;
 483                 referenced.classId = TypeRelationId;
 484                 referenced.objectId = (*dpp)->atttypid;
 485                 referenced.objectSubId = 0;
 486                 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 487
 488                 dpp++;
 489         }
 490
 491         /*
 492          * Next we add the system attributes.  Skip OID if rel has no OIDs. Skip
 493          * all for a view or type relation.  We don't bother with making datatype
 494          * dependencies here, since presumably all these types are pinned.
 495          */
 496         if (relkind != RELKIND_VIEW && relkind != RELKIND_COMPOSITE_TYPE)
 497         {
 498                 dpp = SysAtt;
 499                 for (i = 0; i < -1 - FirstLowInvalidHeapAttributeNumber; i++)
 500                 {
 501                         if (tupdesc->tdhasoid ||
 502                                 (*dpp)->attnum != ObjectIdAttributeNumber)
 503                         {
 504                                 Form_pg_attribute attStruct;
 505
 506                                 tup = heap_addheader(Natts_pg_attribute,
 507                                                                          false,
 508                                                                          ATTRIBUTE_TUPLE_SIZE,
 509                                                                          (void *) *dpp);
 510                                 attStruct = (Form_pg_attribute) GETSTRUCT(tup);
 511
 512                                 /* Fill in the correct relation OID in the copied tuple */
 513                                 attStruct->attrelid = new_rel_oid;
 514
 515                                 /* Fill in correct inheritance info for the OID column */
 516                                 if (attStruct->attnum == ObjectIdAttributeNumber)
 517                                 {
 518                                         attStruct->attislocal = oidislocal;
 519                                         attStruct->attinhcount = oidinhcount;
 520                                 }
 521
 522                                 /*
 523                                  * Unneeded since they should be OK in the constant data
 524                                  * anyway
 525                                  */
 526                                 /* attStruct->attstattarget = 0; */
 527                                 /* attStruct->attcacheoff = -1; */
 528
 529                                 simple_heap_insert(rel, tup);
 530
 531                                 CatalogIndexInsert(indstate, tup);
 532
 533                                 heap_freetuple(tup);
 534                         }
 535                         dpp++;
 536                 }
 537         }
 538
 539         /*
 540          * clean up
 541          */
 542         CatalogCloseIndexes(indstate);
 543
 544         heap_close(rel, RowExclusiveLock);
 545 }
 546
 547 /* --------------------------------
 548  *              AddNewRelationTuple
 549  *
 550  *              this registers the new relation in the catalogs by
 551  *              adding a tuple to pg_class.
 552  * --------------------------------
 553  */
 554 static void
 555 AddNewRelationTuple(Relation pg_class_desc,
 556                                         Relation new_rel_desc,
 557                                         Oid new_rel_oid,
 558                                         Oid new_type_oid,
 559                                         Oid relowner,
 560                                         char relkind)
 561 {
 562         Form_pg_class new_rel_reltup;
 563         HeapTuple       tup;
 564
 565         /*
 566          * first we update some of the information in our uncataloged relation's
 567          * relation descriptor.
 568          */
 569         new_rel_reltup = new_rel_desc->rd_rel;
 570
 571         switch (relkind)
 572         {
 573                 case RELKIND_RELATION:
 574                 case RELKIND_INDEX:
 575                 case RELKIND_TOASTVALUE:
 576                         /* The relation is real, but as yet empty */
 577                         new_rel_reltup->relpages = 0;
 578                         new_rel_reltup->reltuples = 0;
 579                         break;
 580                 case RELKIND_SEQUENCE:
 581                         /* Sequences always have a known size */
 582                         new_rel_reltup->relpages = 1;
 583                         new_rel_reltup->reltuples = 1;
 584                         break;
 585                 default:
 586                         /* Views, etc, have no disk storage */
 587                         new_rel_reltup->relpages = 0;
 588                         new_rel_reltup->reltuples = 0;
 589                         break;
 590         }
 591
 592         new_rel_reltup->relowner = relowner;
 593         new_rel_reltup->reltype = new_type_oid;
 594         new_rel_reltup->relkind = relkind;
 595
 596         new_rel_desc->rd_att->tdtypeid = new_type_oid;
 597
 598         /* ----------------
 599          *      now form a tuple to add to pg_class
 600          *      XXX Natts_pg_class_fixed is a hack - see pg_class.h
 601          * ----------------
 602          */
 603         tup = heap_addheader(Natts_pg_class_fixed,
 604                                                  true,
 605                                                  CLASS_TUPLE_SIZE,
 606                                                  (void *) new_rel_reltup);
 607
 608         /* force tuple to have the desired OID */
 609         HeapTupleSetOid(tup, new_rel_oid);
 610
 611         /*
 612          * finally insert the new tuple, update the indexes, and clean up.
 613          */
 614         simple_heap_insert(pg_class_desc, tup);
 615
 616         CatalogUpdateIndexes(pg_class_desc, tup);
 617
 618         heap_freetuple(tup);
 619 }
 620
 621
 622 /* --------------------------------
 623  *              AddNewRelationType -
 624  *
 625  *              define a composite type corresponding to the new relation
 626  * --------------------------------
 627  */
 628 static Oid
 629 AddNewRelationType(const char *typeName,
 630                                    Oid typeNamespace,
 631                                    Oid new_rel_oid,
 632                                    char new_rel_kind)
 633 {
 634         return
 635                 TypeCreate(typeName,    /* type name */
 636                                    typeNamespace,               /* type namespace */
 637                                    new_rel_oid, /* relation oid */
 638                                    new_rel_kind,        /* relation kind */
 639                                    -1,                  /* internal size (varlena) */
 640                                    'c',                 /* type-type (complex) */
 641                                    ',',                 /* default array delimiter */
 642                                    F_RECORD_IN, /* input procedure */
 643                                    F_RECORD_OUT,        /* output procedure */
 644                                    F_RECORD_RECV,               /* receive procedure */
 645                                    F_RECORD_SEND,               /* send procedure */
 646                                    InvalidOid,  /* analyze procedure - default */
 647                                    InvalidOid,  /* array element type - irrelevant */
 648                                    InvalidOid,  /* domain base type - irrelevant */
 649                                    NULL,                /* default value - none */
 650                                    NULL,                /* default binary representation */
 651                                    false,               /* passed by reference */
 652                                    'd',                 /* alignment - must be the largest! */
 653                                    'x',                 /* fully TOASTable */
 654                                    -1,                  /* typmod */
 655                                    0,                   /* array dimensions for typBaseType */
 656                                    false);              /* Type NOT NULL */
 657 }
 658
 659 /* --------------------------------
 660  *              heap_create_with_catalog
 661  *
 662  *              creates a new cataloged relation.  see comments above.
 663  * --------------------------------
 664  */
 665 Oid
 666 heap_create_with_catalog(const char *relname,
 667                                                  Oid relnamespace,
 668                                                  Oid reltablespace,
 669                                                  Oid relid,
 670                                                  Oid ownerid,
 671                                                  TupleDesc tupdesc,
 672                                                  char relkind,
 673                                                  bool shared_relation,
 674                                                  bool oidislocal,
 675                                                  int oidinhcount,
 676                                                  OnCommitAction oncommit,
 677                                                  bool allow_system_table_mods)
 678 {
 679         Relation        pg_class_desc;
 680         Relation        new_rel_desc;
 681         Oid                     new_type_oid;
 682
 683         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
 684
 685         /*
 686          * sanity checks
 687          */
 688         Assert(IsNormalProcessingMode() || IsBootstrapProcessingMode());
 689
 690         CheckAttributeNamesTypes(tupdesc, relkind);
 691
 692         if (get_relname_relid(relname, relnamespace))
 693                 ereport(ERROR,
 694                                 (errcode(ERRCODE_DUPLICATE_TABLE),
 695                                  errmsg("relation \"%s\" already exists", relname)));
 696
 697         /*
 698          * Allocate an OID for the relation, unless we were told what to use.
 699          *
 700          * The OID will be the relfilenode as well, so make sure it doesn't
 701          * collide with either pg_class OIDs or existing physical files.
 702          */
 703         if (!OidIsValid(relid))
 704                 relid = GetNewRelFileNode(reltablespace, shared_relation,
 705                                                                   pg_class_desc);
 706
 707         /*
 708          * Create the relcache entry (mostly dummy at this point) and the physical
 709          * disk file.  (If we fail further down, it's the smgr's responsibility to
 710          * remove the disk file again.)
 711          */
 712         new_rel_desc = heap_create(relname,
 713                                                            relnamespace,
 714                                                            reltablespace,
 715                                                            relid,
 716                                                            tupdesc,
 717                                                            relkind,
 718                                                            shared_relation,
 719                                                            allow_system_table_mods);
 720
 721         Assert(relid == RelationGetRelid(new_rel_desc));
 722
 723         /*
 724          * since defining a relation also defines a complex type, we add a new
 725          * system type corresponding to the new relation.
 726          *
 727          * NOTE: we could get a unique-index failure here, in case the same name
 728          * has already been used for a type.
 729          */
 730         new_type_oid = AddNewRelationType(relname,
 731                                                                           relnamespace,
 732                                                                           relid,
 733                                                                           relkind);
 734
 735         /*
 736          * now create an entry in pg_class for the relation.
 737          *
 738          * NOTE: we could get a unique-index failure here, in case someone else is
 739          * creating the same relation name in parallel but hadn't committed yet
 740          * when we checked for a duplicate name above.
 741          */
 742         AddNewRelationTuple(pg_class_desc,
 743                                                 new_rel_desc,
 744                                                 relid,
 745                                                 new_type_oid,
 746                                                 ownerid,
 747                                                 relkind);
 748
 749         /*
 750          * now add tuples to pg_attribute for the attributes in our new relation.
 751          */
 752         AddNewAttributeTuples(relid, new_rel_desc->rd_att, relkind,
 753                                                   oidislocal, oidinhcount);
 754
 755         /*
 756          * make a dependency link to force the relation to be deleted if its
 757          * namespace is.  Skip this in bootstrap mode, since we don't make
 758          * dependencies while bootstrapping.
 759          *
 760          * Also make a dependency link to its owner.
 761          */
 762         if (!IsBootstrapProcessingMode())
 763         {
 764                 ObjectAddress myself,
 765                                         referenced;
 766
 767                 myself.classId = RelationRelationId;
 768                 myself.objectId = relid;
 769                 myself.objectSubId = 0;
 770                 referenced.classId = NamespaceRelationId;
 771                 referenced.objectId = relnamespace;
 772                 referenced.objectSubId = 0;
 773                 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 774
 775                 /*
 776                  * For composite types, the dependency on owner is tracked for the
 777                  * pg_type entry, so don't record it here.  All other relkinds need
 778                  * their ownership tracked.
 779                  */
 780                 if (relkind != RELKIND_COMPOSITE_TYPE)
 781                         recordDependencyOnOwner(RelationRelationId, relid, ownerid);
 782         }
 783
 784         /*
 785          * store constraints and defaults passed in the tupdesc, if any.
 786          *
 787          * NB: this may do a CommandCounterIncrement and rebuild the relcache
 788          * entry, so the relation must be valid and self-consistent at this point.
 789          * In particular, there are not yet constraints and defaults anywhere.
 790          */
 791         StoreConstraints(new_rel_desc, tupdesc);
 792
 793         /*
 794          * If there's a special on-commit action, remember it
 795          */
 796         if (oncommit != ONCOMMIT_NOOP)
 797                 register_on_commit_action(relid, oncommit);
 798
 799         /*
 800          * ok, the relation has been cataloged, so close our relations and return
 801          * the OID of the newly created relation.
 802          */
 803         heap_close(new_rel_desc, NoLock);       /* do not unlock till end of xact */
 804         heap_close(pg_class_desc, RowExclusiveLock);
 805
 806         return relid;
 807 }
 808
 809
 810 /*
 811  *              RelationRemoveInheritance
 812  *
 813  * Formerly, this routine checked for child relations and aborted the
 814  * deletion if any were found.  Now we rely on the dependency mechanism
 815  * to check for or delete child relations.      By the time we get here,
 816  * there are no children and we need only remove any pg_inherits rows
 817  * linking this relation to its parent(s).
 818  */
 819 static void
 820 RelationRemoveInheritance(Oid relid)
 821 {
 822         Relation        catalogRelation;
 823         SysScanDesc scan;
 824         ScanKeyData key;
 825         HeapTuple       tuple;
 826
 827         catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
 828
 829         ScanKeyInit(&key,
 830                                 Anum_pg_inherits_inhrelid,
 831                                 BTEqualStrategyNumber, F_OIDEQ,
 832                                 ObjectIdGetDatum(relid));
 833
 834         scan = systable_beginscan(catalogRelation, InheritsRelidSeqnoIndexId, true,
 835                                                           SnapshotNow, 1, &key);
 836
 837         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
 838                 simple_heap_delete(catalogRelation, &tuple->t_self);
 839
 840         systable_endscan(scan);
 841         heap_close(catalogRelation, RowExclusiveLock);
 842 }
 843
 844 /*
 845  *              DeleteRelationTuple
 846  *
 847  * Remove pg_class row for the given relid.
 848  *
 849  * Note: this is shared by relation deletion and index deletion.  It's
 850  * not intended for use anyplace else.
 851  */
 852 void
 853 DeleteRelationTuple(Oid relid)
 854 {
 855         Relation        pg_class_desc;
 856         HeapTuple       tup;
 857
 858         /* Grab an appropriate lock on the pg_class relation */
 859         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
 860
 861         tup = SearchSysCache(RELOID,
 862                                                  ObjectIdGetDatum(relid),
 863                                                  0, 0, 0);
 864         if (!HeapTupleIsValid(tup))
 865                 elog(ERROR, "cache lookup failed for relation %u", relid);
 866
 867         /* delete the relation tuple from pg_class, and finish up */
 868         simple_heap_delete(pg_class_desc, &tup->t_self);
 869
 870         ReleaseSysCache(tup);
 871
 872         heap_close(pg_class_desc, RowExclusiveLock);
 873 }
 874
 875 /*
 876  *              DeleteAttributeTuples
 877  *
 878  * Remove pg_attribute rows for the given relid.
 879  *
 880  * Note: this is shared by relation deletion and index deletion.  It's
 881  * not intended for use anyplace else.
 882  */
 883 void
 884 DeleteAttributeTuples(Oid relid)
 885 {
 886         Relation        attrel;
 887         SysScanDesc scan;
 888         ScanKeyData key[1];
 889         HeapTuple       atttup;
 890
 891         /* Grab an appropriate lock on the pg_attribute relation */
 892         attrel = heap_open(AttributeRelationId, RowExclusiveLock);
 893
 894         /* Use the index to scan only attributes of the target relation */
 895         ScanKeyInit(&key[0],
 896                                 Anum_pg_attribute_attrelid,
 897                                 BTEqualStrategyNumber, F_OIDEQ,
 898                                 ObjectIdGetDatum(relid));
 899
 900         scan = systable_beginscan(attrel, AttributeRelidNumIndexId, true,
 901                                                           SnapshotNow, 1, key);
 902
 903         /* Delete all the matching tuples */
 904         while ((atttup = systable_getnext(scan)) != NULL)
 905                 simple_heap_delete(attrel, &atttup->t_self);
 906
 907         /* Clean up after the scan */
 908         systable_endscan(scan);
 909         heap_close(attrel, RowExclusiveLock);
 910 }
 911
 912 /*
 913  *              RemoveAttributeById
 914  *
 915  * This is the guts of ALTER TABLE DROP COLUMN: actually mark the attribute
 916  * deleted in pg_attribute.  We also remove pg_statistic entries for it.
 917  * (Everything else needed, such as getting rid of any pg_attrdef entry,
 918  * is handled by dependency.c.)
 919  */
 920 void
 921 RemoveAttributeById(Oid relid, AttrNumber attnum)
 922 {
 923         Relation        rel;
 924         Relation        attr_rel;
 925         HeapTuple       tuple;
 926         Form_pg_attribute attStruct;
 927         char            newattname[NAMEDATALEN];
 928
 929         /*
 930          * Grab an exclusive lock on the target table, which we will NOT release
 931          * until end of transaction.  (In the simple case where we are directly
 932          * dropping this column, AlterTableDropColumn already did this ... but
 933          * when cascading from a drop of some other object, we may not have any
 934          * lock.)
 935          */
 936         rel = relation_open(relid, AccessExclusiveLock);
 937
 938         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
 939
 940         tuple = SearchSysCacheCopy(ATTNUM,
 941                                                            ObjectIdGetDatum(relid),
 942                                                            Int16GetDatum(attnum),
 943                                                            0, 0);
 944         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
 945                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
 946                          attnum, relid);
 947         attStruct = (Form_pg_attribute) GETSTRUCT(tuple);
 948
 949         if (attnum < 0)
 950         {
 951                 /* System attribute (probably OID) ... just delete the row */
 952
 953                 simple_heap_delete(attr_rel, &tuple->t_self);
 954         }
 955         else
 956         {
 957                 /* Dropping user attributes is lots harder */
 958
 959                 /* Mark the attribute as dropped */
 960                 attStruct->attisdropped = true;
 961
 962                 /*
 963                  * Set the type OID to invalid.  A dropped attribute's type link
 964                  * cannot be relied on (once the attribute is dropped, the type might
 965                  * be too). Fortunately we do not need the type row --- the only
 966                  * really essential information is the type's typlen and typalign,
 967                  * which are preserved in the attribute's attlen and attalign.  We set
 968                  * atttypid to zero here as a means of catching code that incorrectly
 969                  * expects it to be valid.
 970                  */
 971                 attStruct->atttypid = InvalidOid;
 972
 973                 /* Remove any NOT NULL constraint the column may have */
 974                 attStruct->attnotnull = false;
 975
 976                 /* We don't want to keep stats for it anymore */
 977                 attStruct->attstattarget = 0;
 978
 979                 /*
 980                  * Change the column name to something that isn't likely to conflict
 981                  */
 982                 snprintf(newattname, sizeof(newattname),
 983                                  "........pg.dropped.%d........", attnum);
 984                 namestrcpy(&(attStruct->attname), newattname);
 985
 986                 simple_heap_update(attr_rel, &tuple->t_self, tuple);
 987
 988                 /* keep the system catalog indexes current */
 989                 CatalogUpdateIndexes(attr_rel, tuple);
 990         }
 991
 992         /*
 993          * Because updating the pg_attribute row will trigger a relcache flush for
 994          * the target relation, we need not do anything else to notify other
 995          * backends of the change.
 996          */
 997
 998         heap_close(attr_rel, RowExclusiveLock);
 999
1000         if (attnum > 0)
1001                 RemoveStatistics(relid, attnum);
1002
1003         relation_close(rel, NoLock);
1004 }
1005
1006 /*
1007  *              RemoveAttrDefault
1008  *
1009  * If the specified relation/attribute has a default, remove it.
1010  * (If no default, raise error if complain is true, else return quietly.)
1011  */
1012 void
1013 RemoveAttrDefault(Oid relid, AttrNumber attnum,
1014                                   DropBehavior behavior, bool complain)
1015 {
1016         Relation        attrdef_rel;
1017         ScanKeyData scankeys[2];
1018         SysScanDesc scan;
1019         HeapTuple       tuple;
1020         bool            found = false;
1021
1022         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1023
1024         ScanKeyInit(&scankeys[0],
1025                                 Anum_pg_attrdef_adrelid,
1026                                 BTEqualStrategyNumber, F_OIDEQ,
1027                                 ObjectIdGetDatum(relid));
1028         ScanKeyInit(&scankeys[1],
1029                                 Anum_pg_attrdef_adnum,
1030                                 BTEqualStrategyNumber, F_INT2EQ,
1031                                 Int16GetDatum(attnum));
1032
1033         scan = systable_beginscan(attrdef_rel, AttrDefaultIndexId, true,
1034                                                           SnapshotNow, 2, scankeys);
1035
1036         /* There should be at most one matching tuple, but we loop anyway */
1037         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1038         {
1039                 ObjectAddress object;
1040
1041                 object.classId = AttrDefaultRelationId;
1042                 object.objectId = HeapTupleGetOid(tuple);
1043                 object.objectSubId = 0;
1044
1045                 performDeletion(&object, behavior);
1046
1047                 found = true;
1048         }
1049
1050         systable_endscan(scan);
1051         heap_close(attrdef_rel, RowExclusiveLock);
1052
1053         if (complain && !found)
1054                 elog(ERROR, "could not find attrdef tuple for relation %u attnum %d",
1055                          relid, attnum);
1056 }
1057
1058 /*
1059  *              RemoveAttrDefaultById
1060  *
1061  * Remove a pg_attrdef entry specified by OID.  This is the guts of
1062  * attribute-default removal.  Note it should be called via performDeletion,
1063  * not directly.
1064  */
1065 void
1066 RemoveAttrDefaultById(Oid attrdefId)
1067 {
1068         Relation        attrdef_rel;
1069         Relation        attr_rel;
1070         Relation        myrel;
1071         ScanKeyData scankeys[1];
1072         SysScanDesc scan;
1073         HeapTuple       tuple;
1074         Oid                     myrelid;
1075         AttrNumber      myattnum;
1076
1077         /* Grab an appropriate lock on the pg_attrdef relation */
1078         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1079
1080         /* Find the pg_attrdef tuple */
1081         ScanKeyInit(&scankeys[0],
1082                                 ObjectIdAttributeNumber,
1083                                 BTEqualStrategyNumber, F_OIDEQ,
1084                                 ObjectIdGetDatum(attrdefId));
1085
1086         scan = systable_beginscan(attrdef_rel, AttrDefaultOidIndexId, true,
1087                                                           SnapshotNow, 1, scankeys);
1088
1089         tuple = systable_getnext(scan);
1090         if (!HeapTupleIsValid(tuple))
1091                 elog(ERROR, "could not find tuple for attrdef %u", attrdefId);
1092
1093         myrelid = ((Form_pg_attrdef) GETSTRUCT(tuple))->adrelid;
1094         myattnum = ((Form_pg_attrdef) GETSTRUCT(tuple))->adnum;
1095
1096         /* Get an exclusive lock on the relation owning the attribute */
1097         myrel = relation_open(myrelid, AccessExclusiveLock);
1098
1099         /* Now we can delete the pg_attrdef row */
1100         simple_heap_delete(attrdef_rel, &tuple->t_self);
1101
1102         systable_endscan(scan);
1103         heap_close(attrdef_rel, RowExclusiveLock);
1104
1105         /* Fix the pg_attribute row */
1106         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1107
1108         tuple = SearchSysCacheCopy(ATTNUM,
1109                                                            ObjectIdGetDatum(myrelid),
1110                                                            Int16GetDatum(myattnum),
1111                                                            0, 0);
1112         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
1113                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1114                          myattnum, myrelid);
1115
1116         ((Form_pg_attribute) GETSTRUCT(tuple))->atthasdef = false;
1117
1118         simple_heap_update(attr_rel, &tuple->t_self, tuple);
1119
1120         /* keep the system catalog indexes current */
1121         CatalogUpdateIndexes(attr_rel, tuple);
1122
1123         /*
1124          * Our update of the pg_attribute row will force a relcache rebuild, so
1125          * there's nothing else to do here.
1126          */
1127         heap_close(attr_rel, RowExclusiveLock);
1128
1129         /* Keep lock on attribute's rel until end of xact */
1130         relation_close(myrel, NoLock);
1131 }
1132
1133 /*
1134  * heap_drop_with_catalog       - removes specified relation from catalogs
1135  *
1136  * Note that this routine is not responsible for dropping objects that are
1137  * linked to the pg_class entry via dependencies (for example, indexes and
1138  * constraints).  Those are deleted by the dependency-tracing logic in
1139  * dependency.c before control gets here.  In general, therefore, this routine
1140  * should never be called directly; go through performDeletion() instead.
1141  */
1142 void
1143 heap_drop_with_catalog(Oid relid)
1144 {
1145         Relation        rel;
1146
1147         /*
1148          * Open and lock the relation.
1149          */
1150         rel = relation_open(relid, AccessExclusiveLock);
1151
1152         /*
1153          * Schedule unlinking of the relation's physical file at commit.
1154          */
1155         if (rel->rd_rel->relkind != RELKIND_VIEW &&
1156                 rel->rd_rel->relkind != RELKIND_COMPOSITE_TYPE)
1157         {
1158                 RelationOpenSmgr(rel);
1159                 smgrscheduleunlink(rel->rd_smgr, rel->rd_istemp);
1160         }
1161
1162         /*
1163          * Close relcache entry, but *keep* AccessExclusiveLock on the relation
1164          * until transaction commit.  This ensures no one else will try to do
1165          * something with the doomed relation.
1166          */
1167         relation_close(rel, NoLock);
1168
1169         /*
1170          * Forget any ON COMMIT action for the rel
1171          */
1172         remove_on_commit_action(relid);
1173
1174         /*
1175          * Flush the relation from the relcache.  We want to do this before
1176          * starting to remove catalog entries, just to be certain that no relcache
1177          * entry rebuild will happen partway through.  (That should not really
1178          * matter, since we don't do CommandCounterIncrement here, but let's be
1179          * safe.)
1180          */
1181         RelationForgetRelation(relid);
1182
1183         /*
1184          * remove inheritance information
1185          */
1186         RelationRemoveInheritance(relid);
1187
1188         /*
1189          * delete statistics
1190          */
1191         RemoveStatistics(relid, 0);
1192
1193         /*
1194          * delete attribute tuples
1195          */
1196         DeleteAttributeTuples(relid);
1197
1198         /*
1199          * delete relation tuple
1200          */
1201         DeleteRelationTuple(relid);
1202 }
1203
1204
1205 /*
1206  * Store a default expression for column attnum of relation rel.
1207  * The expression must be presented as a nodeToString() string.
1208  */
1209 void
1210 StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
1211 {
1212         Node       *expr;
1213         char       *adsrc;
1214         Relation        adrel;
1215         HeapTuple       tuple;
1216         Datum           values[4];
1217         static char nulls[4] = {' ', ' ', ' ', ' '};
1218         Relation        attrrel;
1219         HeapTuple       atttup;
1220         Form_pg_attribute attStruct;
1221         Oid                     attrdefOid;
1222         ObjectAddress colobject,
1223                                 defobject;
1224
1225         /*
1226          * Need to construct source equivalent of given node-string.
1227          */
1228         expr = stringToNode(adbin);
1229
1230         /*
1231          * deparse it
1232          */
1233         adsrc = deparse_expression(expr,
1234                                                         deparse_context_for(RelationGetRelationName(rel),
1235                                                                                                 RelationGetRelid(rel)),
1236                                                            false, false);
1237
1238         /*
1239          * Make the pg_attrdef entry.
1240          */
1241         values[Anum_pg_attrdef_adrelid - 1] = RelationGetRelid(rel);
1242         values[Anum_pg_attrdef_adnum - 1] = attnum;
1243         values[Anum_pg_attrdef_adbin - 1] = DirectFunctionCall1(textin,
1244                                                                                                          CStringGetDatum(adbin));
1245         values[Anum_pg_attrdef_adsrc - 1] = DirectFunctionCall1(textin,
1246                                                                                                          CStringGetDatum(adsrc));
1247
1248         adrel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1249
1250         tuple = heap_formtuple(adrel->rd_att, values, nulls);
1251         attrdefOid = simple_heap_insert(adrel, tuple);
1252
1253         CatalogUpdateIndexes(adrel, tuple);
1254
1255         defobject.classId = AttrDefaultRelationId;
1256         defobject.objectId = attrdefOid;
1257         defobject.objectSubId = 0;
1258
1259         heap_close(adrel, RowExclusiveLock);
1260
1261         /* now can free some of the stuff allocated above */
1262         pfree(DatumGetPointer(values[Anum_pg_attrdef_adbin - 1]));
1263         pfree(DatumGetPointer(values[Anum_pg_attrdef_adsrc - 1]));
1264         heap_freetuple(tuple);
1265         pfree(adsrc);
1266
1267         /*
1268          * Update the pg_attribute entry for the column to show that a default
1269          * exists.
1270          */
1271         attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
1272         atttup = SearchSysCacheCopy(ATTNUM,
1273                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
1274                                                                 Int16GetDatum(attnum),
1275                                                                 0, 0);
1276         if (!HeapTupleIsValid(atttup))
1277                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1278                          attnum, RelationGetRelid(rel));
1279         attStruct = (Form_pg_attribute) GETSTRUCT(atttup);
1280         if (!attStruct->atthasdef)
1281         {
1282                 attStruct->atthasdef = true;
1283                 simple_heap_update(attrrel, &atttup->t_self, atttup);
1284                 /* keep catalog indexes current */
1285                 CatalogUpdateIndexes(attrrel, atttup);
1286         }
1287         heap_close(attrrel, RowExclusiveLock);
1288         heap_freetuple(atttup);
1289
1290         /*
1291          * Make a dependency so that the pg_attrdef entry goes away if the column
1292          * (or whole table) is deleted.
1293          */
1294         colobject.classId = RelationRelationId;
1295         colobject.objectId = RelationGetRelid(rel);
1296         colobject.objectSubId = attnum;
1297
1298         recordDependencyOn(&defobject, &colobject, DEPENDENCY_AUTO);
1299
1300         /*
1301          * Record dependencies on objects used in the expression, too.
1302          */
1303         recordDependencyOnExpr(&defobject, expr, NIL, DEPENDENCY_NORMAL);
1304 }
1305
1306 /*
1307  * Store a check-constraint expression for the given relation.
1308  * The expression must be presented as a nodeToString() string.
1309  *
1310  * Caller is responsible for updating the count of constraints
1311  * in the pg_class entry for the relation.
1312  */
1313 static void
1314 StoreRelCheck(Relation rel, char *ccname, char *ccbin)
1315 {
1316         Node       *expr;
1317         char       *ccsrc;
1318         List       *varList;
1319         int                     keycount;
1320         int16      *attNos;
1321
1322         /*
1323          * Convert condition to an expression tree.
1324          */
1325         expr = stringToNode(ccbin);
1326
1327         /*
1328          * deparse it
1329          */
1330         ccsrc = deparse_expression(expr,
1331                                                         deparse_context_for(RelationGetRelationName(rel),
1332                                                                                                 RelationGetRelid(rel)),
1333                                                            false, false);
1334
1335         /*
1336          * Find columns of rel that are used in ccbin
1337          *
1338          * NB: pull_var_clause is okay here only because we don't allow subselects
1339          * in check constraints; it would fail to examine the contents of
1340          * subselects.
1341          */
1342         varList = pull_var_clause(expr, false);
1343         keycount = list_length(varList);
1344
1345         if (keycount > 0)
1346         {
1347                 ListCell   *vl;
1348                 int                     i = 0;
1349
1350                 attNos = (int16 *) palloc(keycount * sizeof(int16));
1351                 foreach(vl, varList)
1352                 {
1353                         Var                *var = (Var *) lfirst(vl);
1354                         int                     j;
1355
1356                         for (j = 0; j < i; j++)
1357                                 if (attNos[j] == var->varattno)
1358                                         break;
1359                         if (j == i)
1360                                 attNos[i++] = var->varattno;
1361                 }
1362                 keycount = i;
1363         }
1364         else
1365                 attNos = NULL;
1366
1367         /*
1368          * Create the Check Constraint
1369          */
1370         CreateConstraintEntry(ccname,           /* Constraint Name */
1371                                                   RelationGetNamespace(rel),    /* namespace */
1372                                                   CONSTRAINT_CHECK,             /* Constraint Type */
1373                                                   false,        /* Is Deferrable */
1374                                                   false,        /* Is Deferred */
1375                                                   RelationGetRelid(rel),                /* relation */
1376                                                   attNos,               /* attrs in the constraint */
1377                                                   keycount,             /* # attrs in the constraint */
1378                                                   InvalidOid,   /* not a domain constraint */
1379                                                   InvalidOid,   /* Foreign key fields */
1380                                                   NULL,
1381                                                   0,
1382                                                   ' ',
1383                                                   ' ',
1384                                                   ' ',
1385                                                   InvalidOid,   /* no associated index */
1386                                                   expr, /* Tree form check constraint */
1387                                                   ccbin,        /* Binary form check constraint */
1388                                                   ccsrc);               /* Source form check constraint */
1389
1390         pfree(ccsrc);
1391 }
1392
1393 /*
1394  * Store defaults and constraints passed in via the tuple constraint struct.
1395  *
1396  * NOTE: only pre-cooked expressions will be passed this way, which is to
1397  * say expressions inherited from an existing relation.  Newly parsed
1398  * expressions can be added later, by direct calls to StoreAttrDefault
1399  * and StoreRelCheck (see AddRelationRawConstraints()).
1400  */
1401 static void
1402 StoreConstraints(Relation rel, TupleDesc tupdesc)
1403 {
1404         TupleConstr *constr = tupdesc->constr;
1405         int                     i;
1406
1407         if (!constr)
1408                 return;                                 /* nothing to do */
1409
1410         /*
1411          * Deparsing of constraint expressions will fail unless the just-created
1412          * pg_attribute tuples for this relation are made visible.      So, bump the
1413          * command counter.  CAUTION: this will cause a relcache entry rebuild.
1414          */
1415         CommandCounterIncrement();
1416
1417         for (i = 0; i < constr->num_defval; i++)
1418                 StoreAttrDefault(rel, constr->defval[i].adnum,
1419                                                  constr->defval[i].adbin);
1420
1421         for (i = 0; i < constr->num_check; i++)
1422                 StoreRelCheck(rel, constr->check[i].ccname,
1423                                           constr->check[i].ccbin);
1424
1425         if (constr->num_check > 0)
1426                 SetRelationNumChecks(rel, constr->num_check);
1427 }
1428
1429 /*
1430  * AddRelationRawConstraints
1431  *
1432  * Add raw (not-yet-transformed) column default expressions and/or constraint
1433  * check expressions to an existing relation.  This is defined to do both
1434  * for efficiency in DefineRelation, but of course you can do just one or
1435  * the other by passing empty lists.
1436  *
1437  * rel: relation to be modified
1438  * rawColDefaults: list of RawColumnDefault structures
1439  * rawConstraints: list of Constraint nodes
1440  *
1441  * All entries in rawColDefaults will be processed.  Entries in rawConstraints
1442  * will be processed only if they are CONSTR_CHECK type and contain a "raw"
1443  * expression.
1444  *
1445  * Returns a list of CookedConstraint nodes that shows the cooked form of
1446  * the default and constraint expressions added to the relation.
1447  *
1448  * NB: caller should have opened rel with AccessExclusiveLock, and should
1449  * hold that lock till end of transaction.      Also, we assume the caller has
1450  * done a CommandCounterIncrement if necessary to make the relation's catalog
1451  * tuples visible.
1452  */
1453 List *
1454 AddRelationRawConstraints(Relation rel,
1455                                                   List *rawColDefaults,
1456                                                   List *rawConstraints)
1457 {
1458         List       *cookedConstraints = NIL;
1459         TupleDesc       tupleDesc;
1460         TupleConstr *oldconstr;
1461         int                     numoldchecks;
1462         ParseState *pstate;
1463         RangeTblEntry *rte;
1464         int                     numchecks;
1465         List       *checknames;
1466         ListCell   *cell;
1467         Node       *expr;
1468         CookedConstraint *cooked;
1469
1470         /*
1471          * Get info about existing constraints.
1472          */
1473         tupleDesc = RelationGetDescr(rel);
1474         oldconstr = tupleDesc->constr;
1475         if (oldconstr)
1476                 numoldchecks = oldconstr->num_check;
1477         else
1478                 numoldchecks = 0;
1479
1480         /*
1481          * Create a dummy ParseState and insert the target relation as its sole
1482          * rangetable entry.  We need a ParseState for transformExpr.
1483          */
1484         pstate = make_parsestate(NULL);
1485         rte = addRangeTableEntryForRelation(pstate,
1486                                                                                 rel,
1487                                                                                 NULL,
1488                                                                                 false,
1489                                                                                 true);
1490         addRTEtoQuery(pstate, rte, true, true, true);
1491
1492         /*
1493          * Process column default expressions.
1494          */
1495         foreach(cell, rawColDefaults)
1496         {
1497                 RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell);
1498                 Form_pg_attribute atp = rel->rd_att->attrs[colDef->attnum - 1];
1499
1500                 expr = cookDefault(pstate, colDef->raw_default,
1501                                                    atp->atttypid, atp->atttypmod,
1502                                                    NameStr(atp->attname));
1503
1504                 StoreAttrDefault(rel, colDef->attnum, nodeToString(expr));
1505
1506                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
1507                 cooked->contype = CONSTR_DEFAULT;
1508                 cooked->name = NULL;
1509                 cooked->attnum = colDef->attnum;
1510                 cooked->expr = expr;
1511                 cookedConstraints = lappend(cookedConstraints, cooked);
1512         }
1513
1514         /*
1515          * Process constraint expressions.
1516          */
1517         numchecks = numoldchecks;
1518         checknames = NIL;
1519         foreach(cell, rawConstraints)
1520         {
1521                 Constraint *cdef = (Constraint *) lfirst(cell);
1522                 char       *ccname;
1523
1524                 if (cdef->contype != CONSTR_CHECK || cdef->raw_expr == NULL)
1525                         continue;
1526                 Assert(cdef->cooked_expr == NULL);
1527
1528                 /*
1529                  * Transform raw parsetree to executable expression.
1530                  */
1531                 expr = transformExpr(pstate, cdef->raw_expr);
1532
1533                 /*
1534                  * Make sure it yields a boolean result.
1535                  */
1536                 expr = coerce_to_boolean(pstate, expr, "CHECK");
1537
1538                 /*
1539                  * Make sure no outside relations are referred to.
1540                  */
1541                 if (list_length(pstate->p_rtable) != 1)
1542                         ereport(ERROR,
1543                                         (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1544                         errmsg("only table \"%s\" can be referenced in check constraint",
1545                                    RelationGetRelationName(rel))));
1546
1547                 /*
1548                  * No subplans or aggregates, either...
1549                  */
1550                 if (pstate->p_hasSubLinks)
1551                         ereport(ERROR,
1552                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1553                                          errmsg("cannot use subquery in check constraint")));
1554                 if (pstate->p_hasAggs)
1555                         ereport(ERROR,
1556                                         (errcode(ERRCODE_GROUPING_ERROR),
1557                            errmsg("cannot use aggregate function in check constraint")));
1558
1559                 /*
1560                  * Check name uniqueness, or generate a name if none was given.
1561                  */
1562                 if (cdef->name != NULL)
1563                 {
1564                         ListCell   *cell2;
1565
1566                         ccname = cdef->name;
1567                         /* Check against pre-existing constraints */
1568                         if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
1569                                                                          RelationGetRelid(rel),
1570                                                                          RelationGetNamespace(rel),
1571                                                                          ccname))
1572                                 ereport(ERROR,
1573                                                 (errcode(ERRCODE_DUPLICATE_OBJECT),
1574                                 errmsg("constraint \"%s\" for relation \"%s\" already exists",
1575                                            ccname, RelationGetRelationName(rel))));
1576                         /* Check against other new constraints */
1577                         /* Needed because we don't do CommandCounterIncrement in loop */
1578                         foreach(cell2, checknames)
1579                         {
1580                                 if (strcmp((char *) lfirst(cell2), ccname) == 0)
1581                                         ereport(ERROR,
1582                                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
1583                                                          errmsg("check constraint \"%s\" already exists",
1584                                                                         ccname)));
1585                         }
1586                 }
1587                 else
1588                 {
1589                         /*
1590                          * When generating a name, we want to create "tab_col_check" for a
1591                          * column constraint and "tab_check" for a table constraint.  We
1592                          * no longer have any info about the syntactic positioning of the
1593                          * constraint phrase, so we approximate this by seeing whether the
1594                          * expression references more than one column.  (If the user
1595                          * played by the rules, the result is the same...)
1596                          *
1597                          * Note: pull_var_clause() doesn't descend into sublinks, but we
1598                          * eliminated those above; and anyway this only needs to be an
1599                          * approximate answer.
1600                          */
1601                         List       *vars;
1602                         char       *colname;
1603
1604                         vars = pull_var_clause(expr, false);
1605
1606                         /* eliminate duplicates */
1607                         vars = list_union(NIL, vars);
1608
1609                         if (list_length(vars) == 1)
1610                                 colname = get_attname(RelationGetRelid(rel),
1611                                                                           ((Var *) linitial(vars))->varattno);
1612                         else
1613                                 colname = NULL;
1614
1615                         ccname = ChooseConstraintName(RelationGetRelationName(rel),
1616                                                                                   colname,
1617                                                                                   "check",
1618                                                                                   RelationGetNamespace(rel),
1619                                                                                   checknames);
1620                 }
1621
1622                 /* save name for future checks */
1623                 checknames = lappend(checknames, ccname);
1624
1625                 /*
1626                  * OK, store it.
1627                  */
1628                 StoreRelCheck(rel, ccname, nodeToString(expr));
1629
1630                 numchecks++;
1631
1632                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
1633                 cooked->contype = CONSTR_CHECK;
1634                 cooked->name = ccname;
1635                 cooked->attnum = 0;
1636                 cooked->expr = expr;
1637                 cookedConstraints = lappend(cookedConstraints, cooked);
1638         }
1639
1640         /*
1641          * Update the count of constraints in the relation's pg_class tuple. We do
1642          * this even if there was no change, in order to ensure that an SI update
1643          * message is sent out for the pg_class tuple, which will force other
1644          * backends to rebuild their relcache entries for the rel. (This is
1645          * critical if we added defaults but not constraints.)
1646          */
1647         SetRelationNumChecks(rel, numchecks);
1648
1649         return cookedConstraints;
1650 }
1651
1652 /*
1653  * Update the count of constraints in the relation's pg_class tuple.
1654  *
1655  * Caller had better hold exclusive lock on the relation.
1656  *
1657  * An important side effect is that a SI update message will be sent out for
1658  * the pg_class tuple, which will force other backends to rebuild their
1659  * relcache entries for the rel.  Also, this backend will rebuild its
1660  * own relcache entry at the next CommandCounterIncrement.
1661  */
1662 static void
1663 SetRelationNumChecks(Relation rel, int numchecks)
1664 {
1665         Relation        relrel;
1666         HeapTuple       reltup;
1667         Form_pg_class relStruct;
1668
1669         relrel = heap_open(RelationRelationId, RowExclusiveLock);
1670         reltup = SearchSysCacheCopy(RELOID,
1671                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
1672                                                                 0, 0, 0);
1673         if (!HeapTupleIsValid(reltup))
1674                 elog(ERROR, "cache lookup failed for relation %u",
1675                          RelationGetRelid(rel));
1676         relStruct = (Form_pg_class) GETSTRUCT(reltup);
1677
1678         if (relStruct->relchecks != numchecks)
1679         {
1680                 relStruct->relchecks = numchecks;
1681
1682                 simple_heap_update(relrel, &reltup->t_self, reltup);
1683
1684                 /* keep catalog indexes current */
1685                 CatalogUpdateIndexes(relrel, reltup);
1686         }
1687         else
1688         {
1689                 /* Skip the disk update, but force relcache inval anyway */
1690                 CacheInvalidateRelcache(rel);
1691         }
1692
1693         heap_freetuple(reltup);
1694         heap_close(relrel, RowExclusiveLock);
1695 }
1696
1697 /*
1698  * Take a raw default and convert it to a cooked format ready for
1699  * storage.
1700  *
1701  * Parse state should be set up to recognize any vars that might appear
1702  * in the expression.  (Even though we plan to reject vars, it's more
1703  * user-friendly to give the correct error message than "unknown var".)
1704  *
1705  * If atttypid is not InvalidOid, coerce the expression to the specified
1706  * type (and typmod atttypmod).   attname is only needed in this case:
1707  * it is used in the error message, if any.
1708  */
1709 Node *
1710 cookDefault(ParseState *pstate,
1711                         Node *raw_default,
1712                         Oid atttypid,
1713                         int32 atttypmod,
1714                         char *attname)
1715 {
1716         Node       *expr;
1717
1718         Assert(raw_default != NULL);
1719
1720         /*
1721          * Transform raw parsetree to executable expression.
1722          */
1723         expr = transformExpr(pstate, raw_default);
1724
1725         /*
1726          * Make sure default expr does not refer to any vars.
1727          */
1728         if (contain_var_clause(expr))
1729                 ereport(ERROR,
1730                                 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1731                           errmsg("cannot use column references in default expression")));
1732
1733         /*
1734          * It can't return a set either.
1735          */
1736         if (expression_returns_set(expr))
1737                 ereport(ERROR,
1738                                 (errcode(ERRCODE_DATATYPE_MISMATCH),
1739                                  errmsg("default expression must not return a set")));
1740
1741         /*
1742          * No subplans or aggregates, either...
1743          */
1744         if (pstate->p_hasSubLinks)
1745                 ereport(ERROR,
1746                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1747                                  errmsg("cannot use subquery in default expression")));
1748         if (pstate->p_hasAggs)
1749                 ereport(ERROR,
1750                                 (errcode(ERRCODE_GROUPING_ERROR),
1751                          errmsg("cannot use aggregate function in default expression")));
1752
1753         /*
1754          * Coerce the expression to the correct type and typmod, if given. This
1755          * should match the parser's processing of non-defaulted expressions ---
1756          * see updateTargetListEntry().
1757          */
1758         if (OidIsValid(atttypid))
1759         {
1760                 Oid                     type_id = exprType(expr);
1761
1762                 expr = coerce_to_target_type(pstate, expr, type_id,
1763                                                                          atttypid, atttypmod,
1764                                                                          COERCION_ASSIGNMENT,
1765                                                                          COERCE_IMPLICIT_CAST);
1766                 if (expr == NULL)
1767                         ereport(ERROR,
1768                                         (errcode(ERRCODE_DATATYPE_MISMATCH),
1769                                          errmsg("column \"%s\" is of type %s"
1770                                                         " but default expression is of type %s",
1771                                                         attname,
1772                                                         format_type_be(atttypid),
1773                                                         format_type_be(type_id)),
1774                            errhint("You will need to rewrite or cast the expression.")));
1775         }
1776
1777         return expr;
1778 }
1779
1780
1781 /*
1782  * Removes all constraints on a relation that match the given name.
1783  *
1784  * It is the responsibility of the calling function to acquire a suitable
1785  * lock on the relation.
1786  *
1787  * Returns: The number of constraints removed.
1788  */
1789 int
1790 RemoveRelConstraints(Relation rel, const char *constrName,
1791                                          DropBehavior behavior)
1792 {
1793         int                     ndeleted = 0;
1794         Relation        conrel;
1795         SysScanDesc conscan;
1796         ScanKeyData key[1];
1797         HeapTuple       contup;
1798
1799         /* Grab an appropriate lock on the pg_constraint relation */
1800         conrel = heap_open(ConstraintRelationId, RowExclusiveLock);
1801
1802         /* Use the index to scan only constraints of the target relation */
1803         ScanKeyInit(&key[0],
1804                                 Anum_pg_constraint_conrelid,
1805                                 BTEqualStrategyNumber, F_OIDEQ,
1806                                 ObjectIdGetDatum(RelationGetRelid(rel)));
1807
1808         conscan = systable_beginscan(conrel, ConstraintRelidIndexId, true,
1809                                                                  SnapshotNow, 1, key);
1810
1811         /*
1812          * Scan over the result set, removing any matching entries.
1813          */
1814         while ((contup = systable_getnext(conscan)) != NULL)
1815         {
1816                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(contup);
1817
1818                 if (strcmp(NameStr(con->conname), constrName) == 0)
1819                 {
1820                         ObjectAddress conobj;
1821
1822                         conobj.classId = ConstraintRelationId;
1823                         conobj.objectId = HeapTupleGetOid(contup);
1824                         conobj.objectSubId = 0;
1825
1826                         performDeletion(&conobj, behavior);
1827
1828                         ndeleted++;
1829                 }
1830         }
1831
1832         /* Clean up after the scan */
1833         systable_endscan(conscan);
1834         heap_close(conrel, RowExclusiveLock);
1835
1836         return ndeleted;
1837 }
1838
1839
1840 /*
1841  * RemoveStatistics --- remove entries in pg_statistic for a rel or column
1842  *
1843  * If attnum is zero, remove all entries for rel; else remove only the one
1844  * for that column.
1845  */
1846 void
1847 RemoveStatistics(Oid relid, AttrNumber attnum)
1848 {
1849         Relation        pgstatistic;
1850         SysScanDesc scan;
1851         ScanKeyData key[2];
1852         int                     nkeys;
1853         HeapTuple       tuple;
1854
1855         pgstatistic = heap_open(StatisticRelationId, RowExclusiveLock);
1856
1857         ScanKeyInit(&key[0],
1858                                 Anum_pg_statistic_starelid,
1859                                 BTEqualStrategyNumber, F_OIDEQ,
1860                                 ObjectIdGetDatum(relid));
1861
1862         if (attnum == 0)
1863                 nkeys = 1;
1864         else
1865         {
1866                 ScanKeyInit(&key[1],
1867                                         Anum_pg_statistic_staattnum,
1868                                         BTEqualStrategyNumber, F_INT2EQ,
1869                                         Int16GetDatum(attnum));
1870                 nkeys = 2;
1871         }
1872
1873         scan = systable_beginscan(pgstatistic, StatisticRelidAttnumIndexId, true,
1874                                                           SnapshotNow, nkeys, key);
1875
1876         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1877                 simple_heap_delete(pgstatistic, &tuple->t_self);
1878
1879         systable_endscan(scan);
1880
1881         heap_close(pgstatistic, RowExclusiveLock);
1882 }
1883
1884
1885 /*
1886  * RelationTruncateIndexes - truncate all
1887  * indexes associated with the heap relation to zero tuples.
1888  *
1889  * The routine will truncate and then reconstruct the indexes on
1890  * the relation specified by the heapId parameter.
1891  */
1892 static void
1893 RelationTruncateIndexes(Oid heapId)
1894 {
1895         Relation        heapRelation;
1896         ListCell   *indlist;
1897
1898         /*
1899          * Open the heap rel.  We need grab no lock because we assume
1900          * heap_truncate is holding an exclusive lock on the heap rel.
1901          */
1902         heapRelation = heap_open(heapId, NoLock);
1903
1904         /* Ask the relcache to produce a list of the indexes of the rel */
1905         foreach(indlist, RelationGetIndexList(heapRelation))
1906         {
1907                 Oid                     indexId = lfirst_oid(indlist);
1908                 Relation        currentIndex;
1909                 IndexInfo  *indexInfo;
1910
1911                 /* Open the index relation */
1912                 currentIndex = index_open(indexId);
1913
1914                 /* Obtain exclusive lock on it, just to be sure */
1915                 LockRelation(currentIndex, AccessExclusiveLock);
1916
1917                 /* Fetch info needed for index_build */
1918                 indexInfo = BuildIndexInfo(currentIndex);
1919
1920                 /* Now truncate the actual file (and discard buffers) */
1921                 RelationTruncate(currentIndex, 0);
1922
1923                 /* Initialize the index and rebuild */
1924                 /* Note: we do not need to re-establish pkey or toast settings */
1925                 index_build(heapRelation, currentIndex, indexInfo, false, false);
1926
1927                 /* We're done with this index */
1928                 index_close(currentIndex);
1929         }
1930
1931         /* And now done with the heap; but keep lock until commit */
1932         heap_close(heapRelation, NoLock);
1933 }
1934
1935 /*
1936  *       heap_truncate
1937  *
1938  *       This routine deletes all data within all the specified relations.
1939  *
1940  * This is not transaction-safe!  There is another, transaction-safe
1941  * implementation in commands/tablecmds.c.      We now use this only for
1942  * ON COMMIT truncation of temporary tables, where it doesn't matter.
1943  */
1944 void
1945 heap_truncate(List *relids)
1946 {
1947         List       *relations = NIL;
1948         ListCell   *cell;
1949
1950         /* Open relations for processing, and grab exclusive access on each */
1951         foreach(cell, relids)
1952         {
1953                 Oid                     rid = lfirst_oid(cell);
1954                 Relation        rel;
1955                 Oid                     toastrelid;
1956
1957                 rel = heap_open(rid, AccessExclusiveLock);
1958                 relations = lappend(relations, rel);
1959
1960                 /* If there is a toast table, add it to the list too */
1961                 toastrelid = rel->rd_rel->reltoastrelid;
1962                 if (OidIsValid(toastrelid))
1963                 {
1964                         rel = heap_open(toastrelid, AccessExclusiveLock);
1965                         relations = lappend(relations, rel);
1966                 }
1967         }
1968
1969         /* Don't allow truncate on tables that are referenced by foreign keys */
1970         heap_truncate_check_FKs(relations, true);
1971
1972         /* OK to do it */
1973         foreach(cell, relations)
1974         {
1975                 Relation        rel = lfirst(cell);
1976
1977                 /* Truncate the actual file (and discard buffers) */
1978                 RelationTruncate(rel, 0);
1979
1980                 /* If this relation has indexes, truncate the indexes too */
1981                 RelationTruncateIndexes(RelationGetRelid(rel));
1982
1983                 /*
1984                  * Close the relation, but keep exclusive lock on it until commit.
1985                  */
1986                 heap_close(rel, NoLock);
1987         }
1988 }
1989
1990 /*
1991  * heap_truncate_check_FKs
1992  *              Check for foreign keys referencing a list of relations that
1993  *              are to be truncated
1994  *
1995  * We disallow such FKs (except self-referential ones) since the whole point
1996  * of TRUNCATE is to not scan the individual rows to be thrown away.
1997  *
1998  * This is split out so it can be shared by both implementations of truncate.
1999  * Caller should already hold a suitable lock on the relations.
2000  *
2001  * tempTables is only used to select an appropriate error message.
2002  */
2003 void
2004 heap_truncate_check_FKs(List *relations, bool tempTables)
2005 {
2006         List       *oids = NIL;
2007         ListCell   *cell;
2008         Relation        fkeyRel;
2009         SysScanDesc fkeyScan;
2010         HeapTuple       tuple;
2011
2012         /*
2013          * Build a list of OIDs of the interesting relations.
2014          *
2015          * If a relation has no triggers, then it can neither have FKs nor be
2016          * referenced by a FK from another table, so we can ignore it.
2017          */
2018         foreach(cell, relations)
2019         {
2020                 Relation        rel = lfirst(cell);
2021
2022                 if (rel->rd_rel->reltriggers != 0)
2023                         oids = lappend_oid(oids, RelationGetRelid(rel));
2024         }
2025
2026         /*
2027          * Fast path: if no relation has triggers, none has FKs either.
2028          */
2029         if (oids == NIL)
2030                 return;
2031
2032         /*
2033          * Otherwise, must scan pg_constraint.  Right now, it is a seqscan because
2034          * there is no available index on confrelid.
2035          */
2036         fkeyRel = heap_open(ConstraintRelationId, AccessShareLock);
2037
2038         fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
2039                                                                   SnapshotNow, 0, NULL);
2040
2041         while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
2042         {
2043                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
2044
2045                 /* Not a foreign key */
2046                 if (con->contype != CONSTRAINT_FOREIGN)
2047                         continue;
2048
2049                 /* Not referencing one of our list of tables */
2050                 if (!list_member_oid(oids, con->confrelid))
2051                         continue;
2052
2053                 /* The referencer should be in our list too */
2054                 if (!list_member_oid(oids, con->conrelid))
2055                 {
2056                         if (tempTables)
2057                                 ereport(ERROR,
2058                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2059                                  errmsg("unsupported ON COMMIT and foreign key combination"),
2060                                                  errdetail("Table \"%s\" references \"%s\" via foreign key constraint \"%s\", but they do not have the same ON COMMIT setting.",
2061                                                                    get_rel_name(con->conrelid),
2062                                                                    get_rel_name(con->confrelid),
2063                                                                    NameStr(con->conname))));
2064                         else
2065                                 ereport(ERROR,
2066                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2067                                                  errmsg("cannot truncate a table referenced in a foreign key constraint"),
2068                                                  errdetail("Table \"%s\" references \"%s\" via foreign key constraint \"%s\".",
2069                                                                    get_rel_name(con->conrelid),
2070                                                                    get_rel_name(con->confrelid),
2071                                                                    NameStr(con->conname)),
2072                                                  errhint("Truncate table \"%s\" at the same time, "
2073                                                                  "or use TRUNCATE ... CASCADE.",
2074                                                                  get_rel_name(con->conrelid))));
2075                 }
2076         }
2077
2078         systable_endscan(fkeyScan);
2079         heap_close(fkeyRel, AccessShareLock);
2080 }
2081
2082 /*
2083  * heap_truncate_find_FKs
2084  *              Find relations having foreign keys referencing any relations that
2085  *              are to be truncated
2086  *
2087  * This is almost the same code as heap_truncate_check_FKs, but we don't
2088  * raise an error if we find such relations; instead we return a list of
2089  * their OIDs.  Also note that the input is a list of OIDs not a list
2090  * of Relations.  The result list does *not* include any rels that are
2091  * already in the input list.
2092  *
2093  * Note: caller should already have exclusive lock on all rels mentioned
2094  * in relationIds.  Since adding or dropping an FK requires exclusive lock
2095  * on both rels, this ensures that the answer will be stable.
2096  */
2097 List *
2098 heap_truncate_find_FKs(List *relationIds)
2099 {
2100         List       *result = NIL;
2101         Relation        fkeyRel;
2102         SysScanDesc fkeyScan;
2103         HeapTuple       tuple;
2104
2105         /*
2106          * Must scan pg_constraint.  Right now, it is a seqscan because
2107          * there is no available index on confrelid.
2108          */
2109         fkeyRel = heap_open(ConstraintRelationId, AccessShareLock);
2110
2111         fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
2112                                                                   SnapshotNow, 0, NULL);
2113
2114         while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
2115         {
2116                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
2117
2118                 /* Not a foreign key */
2119                 if (con->contype != CONSTRAINT_FOREIGN)
2120                         continue;
2121
2122                 /* Not referencing one of our list of tables */
2123                 if (!list_member_oid(relationIds, con->confrelid))
2124                         continue;
2125
2126                 /* Add referencer unless already in input or result list */
2127                 if (!list_member_oid(relationIds, con->conrelid))
2128                         result = list_append_unique_oid(result, con->conrelid);
2129         }
2130
2131         systable_endscan(fkeyScan);
2132         heap_close(fkeyRel, AccessShareLock);
2133
2134         return result;
2135 }