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

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