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