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