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.314 2006/11/05 22:42:08 tgl Exp $
  12  *
  13  *
  14  * INTERFACE ROUTINES
  15  *              heap_create()                   - Create an uncataloged heap relation
  16  *              heap_create_with_catalog() - Create a cataloged relation
  17  *              heap_drop_with_catalog() - Removes named relation from catalogs
  18  *
  19  * NOTES
  20  *        this code taken from access/heap/create.c, which contains
  21  *        the old heap_create_with_catalog, amcreate, and amdestroy.
  22  *        those routines will soon call these routines using the function
  23  *        manager,
  24  *        just like the poorly named "NewXXX" routines do.      The
  25  *        "New" routines are all going to die soon, once and for all!
  26  *              -cim 1/13/91
  27  *
  28  *-------------------------------------------------------------------------
  29  */
  30 #include "postgres.h"
  31
  32 #include "access/genam.h"
  33 #include "access/heapam.h"
  34 #include "access/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_relfrozenxid - 1] = TransactionIdGetDatum(rd_rel->relfrozenxid);
 599         /* start out with empty permissions */
 600         nulls[Anum_pg_class_relacl - 1] = 'n';
 601         if (reloptions != (Datum) 0)
 602                 values[Anum_pg_class_reloptions - 1] = reloptions;
 603         else
 604                 nulls[Anum_pg_class_reloptions - 1] = 'n';
 605
 606         tup = heap_formtuple(RelationGetDescr(pg_class_desc), values, nulls);
 607
 608         /*
 609          * The new tuple must have the oid already chosen for the rel.  Sure would
 610          * be embarrassing to do this sort of thing in polite company.
 611          */
 612         HeapTupleSetOid(tup, new_rel_oid);
 613
 614         /* finally insert the new tuple, update the indexes, and clean up */
 615         simple_heap_insert(pg_class_desc, tup);
 616
 617         CatalogUpdateIndexes(pg_class_desc, tup);
 618
 619         heap_freetuple(tup);
 620 }
 621
 622 /* --------------------------------
 623  *              AddNewRelationTuple
 624  *
 625  *              this registers the new relation in the catalogs by
 626  *              adding a tuple to pg_class.
 627  * --------------------------------
 628  */
 629 static void
 630 AddNewRelationTuple(Relation pg_class_desc,
 631                                         Relation new_rel_desc,
 632                                         Oid new_rel_oid,
 633                                         Oid new_type_oid,
 634                                         Oid relowner,
 635                                         char relkind,
 636                                         Datum reloptions)
 637 {
 638         Form_pg_class new_rel_reltup;
 639
 640         /*
 641          * first we update some of the information in our uncataloged relation's
 642          * relation descriptor.
 643          */
 644         new_rel_reltup = new_rel_desc->rd_rel;
 645
 646         switch (relkind)
 647         {
 648                 case RELKIND_RELATION:
 649                 case RELKIND_INDEX:
 650                 case RELKIND_TOASTVALUE:
 651                         /* The relation is real, but as yet empty */
 652                         new_rel_reltup->relpages = 0;
 653                         new_rel_reltup->reltuples = 0;
 654                         break;
 655                 case RELKIND_SEQUENCE:
 656                         /* Sequences always have a known size */
 657                         new_rel_reltup->relpages = 1;
 658                         new_rel_reltup->reltuples = 1;
 659                         break;
 660                 default:
 661                         /* Views, etc, have no disk storage */
 662                         new_rel_reltup->relpages = 0;
 663                         new_rel_reltup->reltuples = 0;
 664                         break;
 665         }
 666
 667         /* Initialize relfrozenxid */
 668         if (relkind == RELKIND_RELATION ||
 669                 relkind == RELKIND_TOASTVALUE)
 670         {
 671                 /*
 672                  * Initialize to the minimum XID that could put tuples in the table.
 673                  * We know that no xacts older than RecentXmin are still running,
 674                  * so that will do.
 675                  */
 676                 if (!IsBootstrapProcessingMode())
 677                         new_rel_reltup->relfrozenxid = RecentXmin;
 678                 else
 679                         new_rel_reltup->relfrozenxid = FirstNormalTransactionId;
 680         }
 681         else
 682         {
 683                 /*
 684                  * Other relation types will not contain XIDs, so set relfrozenxid
 685                  * to InvalidTransactionId.  (Note: a sequence does contain a tuple,
 686                  * but we force its xmin to be FrozenTransactionId always; see
 687                  * commands/sequence.c.)
 688                  */
 689                 new_rel_reltup->relfrozenxid = InvalidTransactionId;
 690         }
 691
 692         new_rel_reltup->relowner = relowner;
 693         new_rel_reltup->reltype = new_type_oid;
 694         new_rel_reltup->relkind = relkind;
 695
 696         new_rel_desc->rd_att->tdtypeid = new_type_oid;
 697
 698         /* Now build and insert the tuple */
 699         InsertPgClassTuple(pg_class_desc, new_rel_desc, new_rel_oid, reloptions);
 700 }
 701
 702
 703 /* --------------------------------
 704  *              AddNewRelationType -
 705  *
 706  *              define a composite type corresponding to the new relation
 707  * --------------------------------
 708  */
 709 static Oid
 710 AddNewRelationType(const char *typeName,
 711                                    Oid typeNamespace,
 712                                    Oid new_rel_oid,
 713                                    char new_rel_kind)
 714 {
 715         return
 716                 TypeCreate(typeName,    /* type name */
 717                                    typeNamespace,               /* type namespace */
 718                                    new_rel_oid, /* relation oid */
 719                                    new_rel_kind,        /* relation kind */
 720                                    -1,                  /* internal size (varlena) */
 721                                    'c',                 /* type-type (complex) */
 722                                    ',',                 /* default array delimiter */
 723                                    F_RECORD_IN, /* input procedure */
 724                                    F_RECORD_OUT,        /* output procedure */
 725                                    F_RECORD_RECV,               /* receive procedure */
 726                                    F_RECORD_SEND,               /* send procedure */
 727                                    InvalidOid,  /* analyze procedure - default */
 728                                    InvalidOid,  /* array element type - irrelevant */
 729                                    InvalidOid,  /* domain base type - irrelevant */
 730                                    NULL,                /* default value - none */
 731                                    NULL,                /* default binary representation */
 732                                    false,               /* passed by reference */
 733                                    'd',                 /* alignment - must be the largest! */
 734                                    'x',                 /* fully TOASTable */
 735                                    -1,                  /* typmod */
 736                                    0,                   /* array dimensions for typBaseType */
 737                                    false);              /* Type NOT NULL */
 738 }
 739
 740 /* --------------------------------
 741  *              heap_create_with_catalog
 742  *
 743  *              creates a new cataloged relation.  see comments above.
 744  * --------------------------------
 745  */
 746 Oid
 747 heap_create_with_catalog(const char *relname,
 748                                                  Oid relnamespace,
 749                                                  Oid reltablespace,
 750                                                  Oid relid,
 751                                                  Oid ownerid,
 752                                                  TupleDesc tupdesc,
 753                                                  char relkind,
 754                                                  bool shared_relation,
 755                                                  bool oidislocal,
 756                                                  int oidinhcount,
 757                                                  OnCommitAction oncommit,
 758                                                  Datum reloptions,
 759                                                  bool allow_system_table_mods)
 760 {
 761         Relation        pg_class_desc;
 762         Relation        new_rel_desc;
 763         Oid                     new_type_oid;
 764
 765         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
 766
 767         /*
 768          * sanity checks
 769          */
 770         Assert(IsNormalProcessingMode() || IsBootstrapProcessingMode());
 771
 772         CheckAttributeNamesTypes(tupdesc, relkind);
 773
 774         if (get_relname_relid(relname, relnamespace))
 775                 ereport(ERROR,
 776                                 (errcode(ERRCODE_DUPLICATE_TABLE),
 777                                  errmsg("relation \"%s\" already exists", relname)));
 778
 779         /*
 780          * Allocate an OID for the relation, unless we were told what to use.
 781          *
 782          * The OID will be the relfilenode as well, so make sure it doesn't
 783          * collide with either pg_class OIDs or existing physical files.
 784          */
 785         if (!OidIsValid(relid))
 786                 relid = GetNewRelFileNode(reltablespace, shared_relation,
 787                                                                   pg_class_desc);
 788
 789         /*
 790          * Create the relcache entry (mostly dummy at this point) and the physical
 791          * disk file.  (If we fail further down, it's the smgr's responsibility to
 792          * remove the disk file again.)
 793          */
 794         new_rel_desc = heap_create(relname,
 795                                                            relnamespace,
 796                                                            reltablespace,
 797                                                            relid,
 798                                                            tupdesc,
 799                                                            relkind,
 800                                                            shared_relation,
 801                                                            allow_system_table_mods);
 802
 803         Assert(relid == RelationGetRelid(new_rel_desc));
 804
 805         /*
 806          * since defining a relation also defines a complex type, we add a new
 807          * system type corresponding to the new relation.
 808          *
 809          * NOTE: we could get a unique-index failure here, in case the same name
 810          * has already been used for a type.
 811          */
 812         new_type_oid = AddNewRelationType(relname,
 813                                                                           relnamespace,
 814                                                                           relid,
 815                                                                           relkind);
 816
 817         /*
 818          * now create an entry in pg_class for the relation.
 819          *
 820          * NOTE: we could get a unique-index failure here, in case someone else is
 821          * creating the same relation name in parallel but hadn't committed yet
 822          * when we checked for a duplicate name above.
 823          */
 824         AddNewRelationTuple(pg_class_desc,
 825                                                 new_rel_desc,
 826                                                 relid,
 827                                                 new_type_oid,
 828                                                 ownerid,
 829                                                 relkind,
 830                                                 reloptions);
 831
 832         /*
 833          * now add tuples to pg_attribute for the attributes in our new relation.
 834          */
 835         AddNewAttributeTuples(relid, new_rel_desc->rd_att, relkind,
 836                                                   oidislocal, oidinhcount);
 837
 838         /*
 839          * make a dependency link to force the relation to be deleted if its
 840          * namespace is.  Skip this in bootstrap mode, since we don't make
 841          * dependencies while bootstrapping.
 842          *
 843          * Also make a dependency link to its owner.
 844          */
 845         if (!IsBootstrapProcessingMode())
 846         {
 847                 ObjectAddress myself,
 848                                         referenced;
 849
 850                 myself.classId = RelationRelationId;
 851                 myself.objectId = relid;
 852                 myself.objectSubId = 0;
 853                 referenced.classId = NamespaceRelationId;
 854                 referenced.objectId = relnamespace;
 855                 referenced.objectSubId = 0;
 856                 recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 857
 858                 /*
 859                  * For composite types, the dependency on owner is tracked for the
 860                  * pg_type entry, so don't record it here.  All other relkinds need
 861                  * their ownership tracked.
 862                  */
 863                 if (relkind != RELKIND_COMPOSITE_TYPE)
 864                         recordDependencyOnOwner(RelationRelationId, relid, ownerid);
 865         }
 866
 867         /*
 868          * store constraints and defaults passed in the tupdesc, if any.
 869          *
 870          * NB: this may do a CommandCounterIncrement and rebuild the relcache
 871          * entry, so the relation must be valid and self-consistent at this point.
 872          * In particular, there are not yet constraints and defaults anywhere.
 873          */
 874         StoreConstraints(new_rel_desc, tupdesc);
 875
 876         /*
 877          * If there's a special on-commit action, remember it
 878          */
 879         if (oncommit != ONCOMMIT_NOOP)
 880                 register_on_commit_action(relid, oncommit);
 881
 882         /*
 883          * ok, the relation has been cataloged, so close our relations and return
 884          * the OID of the newly created relation.
 885          */
 886         heap_close(new_rel_desc, NoLock);       /* do not unlock till end of xact */
 887         heap_close(pg_class_desc, RowExclusiveLock);
 888
 889         return relid;
 890 }
 891
 892
 893 /*
 894  *              RelationRemoveInheritance
 895  *
 896  * Formerly, this routine checked for child relations and aborted the
 897  * deletion if any were found.  Now we rely on the dependency mechanism
 898  * to check for or delete child relations.      By the time we get here,
 899  * there are no children and we need only remove any pg_inherits rows
 900  * linking this relation to its parent(s).
 901  */
 902 static void
 903 RelationRemoveInheritance(Oid relid)
 904 {
 905         Relation        catalogRelation;
 906         SysScanDesc scan;
 907         ScanKeyData key;
 908         HeapTuple       tuple;
 909
 910         catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
 911
 912         ScanKeyInit(&key,
 913                                 Anum_pg_inherits_inhrelid,
 914                                 BTEqualStrategyNumber, F_OIDEQ,
 915                                 ObjectIdGetDatum(relid));
 916
 917         scan = systable_beginscan(catalogRelation, InheritsRelidSeqnoIndexId, true,
 918                                                           SnapshotNow, 1, &key);
 919
 920         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
 921                 simple_heap_delete(catalogRelation, &tuple->t_self);
 922
 923         systable_endscan(scan);
 924         heap_close(catalogRelation, RowExclusiveLock);
 925 }
 926
 927 /*
 928  *              DeleteRelationTuple
 929  *
 930  * Remove pg_class row for the given relid.
 931  *
 932  * Note: this is shared by relation deletion and index deletion.  It's
 933  * not intended for use anyplace else.
 934  */
 935 void
 936 DeleteRelationTuple(Oid relid)
 937 {
 938         Relation        pg_class_desc;
 939         HeapTuple       tup;
 940
 941         /* Grab an appropriate lock on the pg_class relation */
 942         pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);
 943
 944         tup = SearchSysCache(RELOID,
 945                                                  ObjectIdGetDatum(relid),
 946                                                  0, 0, 0);
 947         if (!HeapTupleIsValid(tup))
 948                 elog(ERROR, "cache lookup failed for relation %u", relid);
 949
 950         /* delete the relation tuple from pg_class, and finish up */
 951         simple_heap_delete(pg_class_desc, &tup->t_self);
 952
 953         ReleaseSysCache(tup);
 954
 955         heap_close(pg_class_desc, RowExclusiveLock);
 956 }
 957
 958 /*
 959  *              DeleteAttributeTuples
 960  *
 961  * Remove pg_attribute rows for the given relid.
 962  *
 963  * Note: this is shared by relation deletion and index deletion.  It's
 964  * not intended for use anyplace else.
 965  */
 966 void
 967 DeleteAttributeTuples(Oid relid)
 968 {
 969         Relation        attrel;
 970         SysScanDesc scan;
 971         ScanKeyData key[1];
 972         HeapTuple       atttup;
 973
 974         /* Grab an appropriate lock on the pg_attribute relation */
 975         attrel = heap_open(AttributeRelationId, RowExclusiveLock);
 976
 977         /* Use the index to scan only attributes of the target relation */
 978         ScanKeyInit(&key[0],
 979                                 Anum_pg_attribute_attrelid,
 980                                 BTEqualStrategyNumber, F_OIDEQ,
 981                                 ObjectIdGetDatum(relid));
 982
 983         scan = systable_beginscan(attrel, AttributeRelidNumIndexId, true,
 984                                                           SnapshotNow, 1, key);
 985
 986         /* Delete all the matching tuples */
 987         while ((atttup = systable_getnext(scan)) != NULL)
 988                 simple_heap_delete(attrel, &atttup->t_self);
 989
 990         /* Clean up after the scan */
 991         systable_endscan(scan);
 992         heap_close(attrel, RowExclusiveLock);
 993 }
 994
 995 /*
 996  *              RemoveAttributeById
 997  *
 998  * This is the guts of ALTER TABLE DROP COLUMN: actually mark the attribute
 999  * deleted in pg_attribute.  We also remove pg_statistic entries for it.
1000  * (Everything else needed, such as getting rid of any pg_attrdef entry,
1001  * is handled by dependency.c.)
1002  */
1003 void
1004 RemoveAttributeById(Oid relid, AttrNumber attnum)
1005 {
1006         Relation        rel;
1007         Relation        attr_rel;
1008         HeapTuple       tuple;
1009         Form_pg_attribute attStruct;
1010         char            newattname[NAMEDATALEN];
1011
1012         /*
1013          * Grab an exclusive lock on the target table, which we will NOT release
1014          * until end of transaction.  (In the simple case where we are directly
1015          * dropping this column, AlterTableDropColumn already did this ... but
1016          * when cascading from a drop of some other object, we may not have any
1017          * lock.)
1018          */
1019         rel = relation_open(relid, AccessExclusiveLock);
1020
1021         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1022
1023         tuple = SearchSysCacheCopy(ATTNUM,
1024                                                            ObjectIdGetDatum(relid),
1025                                                            Int16GetDatum(attnum),
1026                                                            0, 0);
1027         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
1028                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1029                          attnum, relid);
1030         attStruct = (Form_pg_attribute) GETSTRUCT(tuple);
1031
1032         if (attnum < 0)
1033         {
1034                 /* System attribute (probably OID) ... just delete the row */
1035
1036                 simple_heap_delete(attr_rel, &tuple->t_self);
1037         }
1038         else
1039         {
1040                 /* Dropping user attributes is lots harder */
1041
1042                 /* Mark the attribute as dropped */
1043                 attStruct->attisdropped = true;
1044
1045                 /*
1046                  * Set the type OID to invalid.  A dropped attribute's type link
1047                  * cannot be relied on (once the attribute is dropped, the type might
1048                  * be too). Fortunately we do not need the type row --- the only
1049                  * really essential information is the type's typlen and typalign,
1050                  * which are preserved in the attribute's attlen and attalign.  We set
1051                  * atttypid to zero here as a means of catching code that incorrectly
1052                  * expects it to be valid.
1053                  */
1054                 attStruct->atttypid = InvalidOid;
1055
1056                 /* Remove any NOT NULL constraint the column may have */
1057                 attStruct->attnotnull = false;
1058
1059                 /* We don't want to keep stats for it anymore */
1060                 attStruct->attstattarget = 0;
1061
1062                 /*
1063                  * Change the column name to something that isn't likely to conflict
1064                  */
1065                 snprintf(newattname, sizeof(newattname),
1066                                  "........pg.dropped.%d........", attnum);
1067                 namestrcpy(&(attStruct->attname), newattname);
1068
1069                 simple_heap_update(attr_rel, &tuple->t_self, tuple);
1070
1071                 /* keep the system catalog indexes current */
1072                 CatalogUpdateIndexes(attr_rel, tuple);
1073         }
1074
1075         /*
1076          * Because updating the pg_attribute row will trigger a relcache flush for
1077          * the target relation, we need not do anything else to notify other
1078          * backends of the change.
1079          */
1080
1081         heap_close(attr_rel, RowExclusiveLock);
1082
1083         if (attnum > 0)
1084                 RemoveStatistics(relid, attnum);
1085
1086         relation_close(rel, NoLock);
1087 }
1088
1089 /*
1090  *              RemoveAttrDefault
1091  *
1092  * If the specified relation/attribute has a default, remove it.
1093  * (If no default, raise error if complain is true, else return quietly.)
1094  */
1095 void
1096 RemoveAttrDefault(Oid relid, AttrNumber attnum,
1097                                   DropBehavior behavior, bool complain)
1098 {
1099         Relation        attrdef_rel;
1100         ScanKeyData scankeys[2];
1101         SysScanDesc scan;
1102         HeapTuple       tuple;
1103         bool            found = false;
1104
1105         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1106
1107         ScanKeyInit(&scankeys[0],
1108                                 Anum_pg_attrdef_adrelid,
1109                                 BTEqualStrategyNumber, F_OIDEQ,
1110                                 ObjectIdGetDatum(relid));
1111         ScanKeyInit(&scankeys[1],
1112                                 Anum_pg_attrdef_adnum,
1113                                 BTEqualStrategyNumber, F_INT2EQ,
1114                                 Int16GetDatum(attnum));
1115
1116         scan = systable_beginscan(attrdef_rel, AttrDefaultIndexId, true,
1117                                                           SnapshotNow, 2, scankeys);
1118
1119         /* There should be at most one matching tuple, but we loop anyway */
1120         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1121         {
1122                 ObjectAddress object;
1123
1124                 object.classId = AttrDefaultRelationId;
1125                 object.objectId = HeapTupleGetOid(tuple);
1126                 object.objectSubId = 0;
1127
1128                 performDeletion(&object, behavior);
1129
1130                 found = true;
1131         }
1132
1133         systable_endscan(scan);
1134         heap_close(attrdef_rel, RowExclusiveLock);
1135
1136         if (complain && !found)
1137                 elog(ERROR, "could not find attrdef tuple for relation %u attnum %d",
1138                          relid, attnum);
1139 }
1140
1141 /*
1142  *              RemoveAttrDefaultById
1143  *
1144  * Remove a pg_attrdef entry specified by OID.  This is the guts of
1145  * attribute-default removal.  Note it should be called via performDeletion,
1146  * not directly.
1147  */
1148 void
1149 RemoveAttrDefaultById(Oid attrdefId)
1150 {
1151         Relation        attrdef_rel;
1152         Relation        attr_rel;
1153         Relation        myrel;
1154         ScanKeyData scankeys[1];
1155         SysScanDesc scan;
1156         HeapTuple       tuple;
1157         Oid                     myrelid;
1158         AttrNumber      myattnum;
1159
1160         /* Grab an appropriate lock on the pg_attrdef relation */
1161         attrdef_rel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1162
1163         /* Find the pg_attrdef tuple */
1164         ScanKeyInit(&scankeys[0],
1165                                 ObjectIdAttributeNumber,
1166                                 BTEqualStrategyNumber, F_OIDEQ,
1167                                 ObjectIdGetDatum(attrdefId));
1168
1169         scan = systable_beginscan(attrdef_rel, AttrDefaultOidIndexId, true,
1170                                                           SnapshotNow, 1, scankeys);
1171
1172         tuple = systable_getnext(scan);
1173         if (!HeapTupleIsValid(tuple))
1174                 elog(ERROR, "could not find tuple for attrdef %u", attrdefId);
1175
1176         myrelid = ((Form_pg_attrdef) GETSTRUCT(tuple))->adrelid;
1177         myattnum = ((Form_pg_attrdef) GETSTRUCT(tuple))->adnum;
1178
1179         /* Get an exclusive lock on the relation owning the attribute */
1180         myrel = relation_open(myrelid, AccessExclusiveLock);
1181
1182         /* Now we can delete the pg_attrdef row */
1183         simple_heap_delete(attrdef_rel, &tuple->t_self);
1184
1185         systable_endscan(scan);
1186         heap_close(attrdef_rel, RowExclusiveLock);
1187
1188         /* Fix the pg_attribute row */
1189         attr_rel = heap_open(AttributeRelationId, RowExclusiveLock);
1190
1191         tuple = SearchSysCacheCopy(ATTNUM,
1192                                                            ObjectIdGetDatum(myrelid),
1193                                                            Int16GetDatum(myattnum),
1194                                                            0, 0);
1195         if (!HeapTupleIsValid(tuple))           /* shouldn't happen */
1196                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1197                          myattnum, myrelid);
1198
1199         ((Form_pg_attribute) GETSTRUCT(tuple))->atthasdef = false;
1200
1201         simple_heap_update(attr_rel, &tuple->t_self, tuple);
1202
1203         /* keep the system catalog indexes current */
1204         CatalogUpdateIndexes(attr_rel, tuple);
1205
1206         /*
1207          * Our update of the pg_attribute row will force a relcache rebuild, so
1208          * there's nothing else to do here.
1209          */
1210         heap_close(attr_rel, RowExclusiveLock);
1211
1212         /* Keep lock on attribute's rel until end of xact */
1213         relation_close(myrel, NoLock);
1214 }
1215
1216 /*
1217  * heap_drop_with_catalog       - removes specified relation from catalogs
1218  *
1219  * Note that this routine is not responsible for dropping objects that are
1220  * linked to the pg_class entry via dependencies (for example, indexes and
1221  * constraints).  Those are deleted by the dependency-tracing logic in
1222  * dependency.c before control gets here.  In general, therefore, this routine
1223  * should never be called directly; go through performDeletion() instead.
1224  */
1225 void
1226 heap_drop_with_catalog(Oid relid)
1227 {
1228         Relation        rel;
1229
1230         /*
1231          * Open and lock the relation.
1232          */
1233         rel = relation_open(relid, AccessExclusiveLock);
1234
1235         /*
1236          * Schedule unlinking of the relation's physical file at commit.
1237          */
1238         if (rel->rd_rel->relkind != RELKIND_VIEW &&
1239                 rel->rd_rel->relkind != RELKIND_COMPOSITE_TYPE)
1240         {
1241                 RelationOpenSmgr(rel);
1242                 smgrscheduleunlink(rel->rd_smgr, rel->rd_istemp);
1243         }
1244
1245         /*
1246          * Close relcache entry, but *keep* AccessExclusiveLock on the relation
1247          * until transaction commit.  This ensures no one else will try to do
1248          * something with the doomed relation.
1249          */
1250         relation_close(rel, NoLock);
1251
1252         /*
1253          * Forget any ON COMMIT action for the rel
1254          */
1255         remove_on_commit_action(relid);
1256
1257         /*
1258          * Flush the relation from the relcache.  We want to do this before
1259          * starting to remove catalog entries, just to be certain that no relcache
1260          * entry rebuild will happen partway through.  (That should not really
1261          * matter, since we don't do CommandCounterIncrement here, but let's be
1262          * safe.)
1263          */
1264         RelationForgetRelation(relid);
1265
1266         /*
1267          * remove inheritance information
1268          */
1269         RelationRemoveInheritance(relid);
1270
1271         /*
1272          * delete statistics
1273          */
1274         RemoveStatistics(relid, 0);
1275
1276         /*
1277          * delete attribute tuples
1278          */
1279         DeleteAttributeTuples(relid);
1280
1281         /*
1282          * delete relation tuple
1283          */
1284         DeleteRelationTuple(relid);
1285 }
1286
1287
1288 /*
1289  * Store a default expression for column attnum of relation rel.
1290  * The expression must be presented as a nodeToString() string.
1291  */
1292 void
1293 StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
1294 {
1295         Node       *expr;
1296         char       *adsrc;
1297         Relation        adrel;
1298         HeapTuple       tuple;
1299         Datum           values[4];
1300         static char nulls[4] = {' ', ' ', ' ', ' '};
1301         Relation        attrrel;
1302         HeapTuple       atttup;
1303         Form_pg_attribute attStruct;
1304         Oid                     attrdefOid;
1305         ObjectAddress colobject,
1306                                 defobject;
1307
1308         /*
1309          * Need to construct source equivalent of given node-string.
1310          */
1311         expr = stringToNode(adbin);
1312
1313         /*
1314          * deparse it
1315          */
1316         adsrc = deparse_expression(expr,
1317                                                         deparse_context_for(RelationGetRelationName(rel),
1318                                                                                                 RelationGetRelid(rel)),
1319                                                            false, false);
1320
1321         /*
1322          * Make the pg_attrdef entry.
1323          */
1324         values[Anum_pg_attrdef_adrelid - 1] = RelationGetRelid(rel);
1325         values[Anum_pg_attrdef_adnum - 1] = attnum;
1326         values[Anum_pg_attrdef_adbin - 1] = DirectFunctionCall1(textin,
1327                                                                                                          CStringGetDatum(adbin));
1328         values[Anum_pg_attrdef_adsrc - 1] = DirectFunctionCall1(textin,
1329                                                                                                          CStringGetDatum(adsrc));
1330
1331         adrel = heap_open(AttrDefaultRelationId, RowExclusiveLock);
1332
1333         tuple = heap_formtuple(adrel->rd_att, values, nulls);
1334         attrdefOid = simple_heap_insert(adrel, tuple);
1335
1336         CatalogUpdateIndexes(adrel, tuple);
1337
1338         defobject.classId = AttrDefaultRelationId;
1339         defobject.objectId = attrdefOid;
1340         defobject.objectSubId = 0;
1341
1342         heap_close(adrel, RowExclusiveLock);
1343
1344         /* now can free some of the stuff allocated above */
1345         pfree(DatumGetPointer(values[Anum_pg_attrdef_adbin - 1]));
1346         pfree(DatumGetPointer(values[Anum_pg_attrdef_adsrc - 1]));
1347         heap_freetuple(tuple);
1348         pfree(adsrc);
1349
1350         /*
1351          * Update the pg_attribute entry for the column to show that a default
1352          * exists.
1353          */
1354         attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
1355         atttup = SearchSysCacheCopy(ATTNUM,
1356                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
1357                                                                 Int16GetDatum(attnum),
1358                                                                 0, 0);
1359         if (!HeapTupleIsValid(atttup))
1360                 elog(ERROR, "cache lookup failed for attribute %d of relation %u",
1361                          attnum, RelationGetRelid(rel));
1362         attStruct = (Form_pg_attribute) GETSTRUCT(atttup);
1363         if (!attStruct->atthasdef)
1364         {
1365                 attStruct->atthasdef = true;
1366                 simple_heap_update(attrrel, &atttup->t_self, atttup);
1367                 /* keep catalog indexes current */
1368                 CatalogUpdateIndexes(attrrel, atttup);
1369         }
1370         heap_close(attrrel, RowExclusiveLock);
1371         heap_freetuple(atttup);
1372
1373         /*
1374          * Make a dependency so that the pg_attrdef entry goes away if the column
1375          * (or whole table) is deleted.
1376          */
1377         colobject.classId = RelationRelationId;
1378         colobject.objectId = RelationGetRelid(rel);
1379         colobject.objectSubId = attnum;
1380
1381         recordDependencyOn(&defobject, &colobject, DEPENDENCY_AUTO);
1382
1383         /*
1384          * Record dependencies on objects used in the expression, too.
1385          */
1386         recordDependencyOnExpr(&defobject, expr, NIL, DEPENDENCY_NORMAL);
1387 }
1388
1389 /*
1390  * Store a check-constraint expression for the given relation.
1391  * The expression must be presented as a nodeToString() string.
1392  *
1393  * Caller is responsible for updating the count of constraints
1394  * in the pg_class entry for the relation.
1395  */
1396 static void
1397 StoreRelCheck(Relation rel, char *ccname, char *ccbin)
1398 {
1399         Node       *expr;
1400         char       *ccsrc;
1401         List       *varList;
1402         int                     keycount;
1403         int16      *attNos;
1404
1405         /*
1406          * Convert condition to an expression tree.
1407          */
1408         expr = stringToNode(ccbin);
1409
1410         /*
1411          * deparse it
1412          */
1413         ccsrc = deparse_expression(expr,
1414                                                         deparse_context_for(RelationGetRelationName(rel),
1415                                                                                                 RelationGetRelid(rel)),
1416                                                            false, false);
1417
1418         /*
1419          * Find columns of rel that are used in ccbin
1420          *
1421          * NB: pull_var_clause is okay here only because we don't allow subselects
1422          * in check constraints; it would fail to examine the contents of
1423          * subselects.
1424          */
1425         varList = pull_var_clause(expr, false);
1426         keycount = list_length(varList);
1427
1428         if (keycount > 0)
1429         {
1430                 ListCell   *vl;
1431                 int                     i = 0;
1432
1433                 attNos = (int16 *) palloc(keycount * sizeof(int16));
1434                 foreach(vl, varList)
1435                 {
1436                         Var                *var = (Var *) lfirst(vl);
1437                         int                     j;
1438
1439                         for (j = 0; j < i; j++)
1440                                 if (attNos[j] == var->varattno)
1441                                         break;
1442                         if (j == i)
1443                                 attNos[i++] = var->varattno;
1444                 }
1445                 keycount = i;
1446         }
1447         else
1448                 attNos = NULL;
1449
1450         /*
1451          * Create the Check Constraint
1452          */
1453         CreateConstraintEntry(ccname,           /* Constraint Name */
1454                                                   RelationGetNamespace(rel),    /* namespace */
1455                                                   CONSTRAINT_CHECK,             /* Constraint Type */
1456                                                   false,        /* Is Deferrable */
1457                                                   false,        /* Is Deferred */
1458                                                   RelationGetRelid(rel),                /* relation */
1459                                                   attNos,               /* attrs in the constraint */
1460                                                   keycount,             /* # attrs in the constraint */
1461                                                   InvalidOid,   /* not a domain constraint */
1462                                                   InvalidOid,   /* Foreign key fields */
1463                                                   NULL,
1464                                                   0,
1465                                                   ' ',
1466                                                   ' ',
1467                                                   ' ',
1468                                                   InvalidOid,   /* no associated index */
1469                                                   expr, /* Tree form check constraint */
1470                                                   ccbin,        /* Binary form check constraint */
1471                                                   ccsrc);               /* Source form check constraint */
1472
1473         pfree(ccsrc);
1474 }
1475
1476 /*
1477  * Store defaults and constraints passed in via the tuple constraint struct.
1478  *
1479  * NOTE: only pre-cooked expressions will be passed this way, which is to
1480  * say expressions inherited from an existing relation.  Newly parsed
1481  * expressions can be added later, by direct calls to StoreAttrDefault
1482  * and StoreRelCheck (see AddRelationRawConstraints()).
1483  */
1484 static void
1485 StoreConstraints(Relation rel, TupleDesc tupdesc)
1486 {
1487         TupleConstr *constr = tupdesc->constr;
1488         int                     i;
1489
1490         if (!constr)
1491                 return;                                 /* nothing to do */
1492
1493         /*
1494          * Deparsing of constraint expressions will fail unless the just-created
1495          * pg_attribute tuples for this relation are made visible.      So, bump the
1496          * command counter.  CAUTION: this will cause a relcache entry rebuild.
1497          */
1498         CommandCounterIncrement();
1499
1500         for (i = 0; i < constr->num_defval; i++)
1501                 StoreAttrDefault(rel, constr->defval[i].adnum,
1502                                                  constr->defval[i].adbin);
1503
1504         for (i = 0; i < constr->num_check; i++)
1505                 StoreRelCheck(rel, constr->check[i].ccname,
1506                                           constr->check[i].ccbin);
1507
1508         if (constr->num_check > 0)
1509                 SetRelationNumChecks(rel, constr->num_check);
1510 }
1511
1512 /*
1513  * AddRelationRawConstraints
1514  *
1515  * Add raw (not-yet-transformed) column default expressions and/or constraint
1516  * check expressions to an existing relation.  This is defined to do both
1517  * for efficiency in DefineRelation, but of course you can do just one or
1518  * the other by passing empty lists.
1519  *
1520  * rel: relation to be modified
1521  * rawColDefaults: list of RawColumnDefault structures
1522  * rawConstraints: list of Constraint nodes
1523  *
1524  * All entries in rawColDefaults will be processed.  Entries in rawConstraints
1525  * will be processed only if they are CONSTR_CHECK type and contain a "raw"
1526  * expression.
1527  *
1528  * Returns a list of CookedConstraint nodes that shows the cooked form of
1529  * the default and constraint expressions added to the relation.
1530  *
1531  * NB: caller should have opened rel with AccessExclusiveLock, and should
1532  * hold that lock till end of transaction.      Also, we assume the caller has
1533  * done a CommandCounterIncrement if necessary to make the relation's catalog
1534  * tuples visible.
1535  */
1536 List *
1537 AddRelationRawConstraints(Relation rel,
1538                                                   List *rawColDefaults,
1539                                                   List *rawConstraints)
1540 {
1541         List       *cookedConstraints = NIL;
1542         TupleDesc       tupleDesc;
1543         TupleConstr *oldconstr;
1544         int                     numoldchecks;
1545         ParseState *pstate;
1546         RangeTblEntry *rte;
1547         int                     numchecks;
1548         List       *checknames;
1549         ListCell   *cell;
1550         Node       *expr;
1551         CookedConstraint *cooked;
1552
1553         /*
1554          * Get info about existing constraints.
1555          */
1556         tupleDesc = RelationGetDescr(rel);
1557         oldconstr = tupleDesc->constr;
1558         if (oldconstr)
1559                 numoldchecks = oldconstr->num_check;
1560         else
1561                 numoldchecks = 0;
1562
1563         /*
1564          * Create a dummy ParseState and insert the target relation as its sole
1565          * rangetable entry.  We need a ParseState for transformExpr.
1566          */
1567         pstate = make_parsestate(NULL);
1568         rte = addRangeTableEntryForRelation(pstate,
1569                                                                                 rel,
1570                                                                                 NULL,
1571                                                                                 false,
1572                                                                                 true);
1573         addRTEtoQuery(pstate, rte, true, true, true);
1574
1575         /*
1576          * Process column default expressions.
1577          */
1578         foreach(cell, rawColDefaults)
1579         {
1580                 RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell);
1581                 Form_pg_attribute atp = rel->rd_att->attrs[colDef->attnum - 1];
1582
1583                 expr = cookDefault(pstate, colDef->raw_default,
1584                                                    atp->atttypid, atp->atttypmod,
1585                                                    NameStr(atp->attname));
1586
1587                 StoreAttrDefault(rel, colDef->attnum, nodeToString(expr));
1588
1589                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
1590                 cooked->contype = CONSTR_DEFAULT;
1591                 cooked->name = NULL;
1592                 cooked->attnum = colDef->attnum;
1593                 cooked->expr = expr;
1594                 cookedConstraints = lappend(cookedConstraints, cooked);
1595         }
1596
1597         /*
1598          * Process constraint expressions.
1599          */
1600         numchecks = numoldchecks;
1601         checknames = NIL;
1602         foreach(cell, rawConstraints)
1603         {
1604                 Constraint *cdef = (Constraint *) lfirst(cell);
1605                 char       *ccname;
1606
1607                 if (cdef->contype != CONSTR_CHECK || cdef->raw_expr == NULL)
1608                         continue;
1609                 Assert(cdef->cooked_expr == NULL);
1610
1611                 /*
1612                  * Transform raw parsetree to executable expression.
1613                  */
1614                 expr = transformExpr(pstate, cdef->raw_expr);
1615
1616                 /*
1617                  * Make sure it yields a boolean result.
1618                  */
1619                 expr = coerce_to_boolean(pstate, expr, "CHECK");
1620
1621                 /*
1622                  * Make sure no outside relations are referred to.
1623                  */
1624                 if (list_length(pstate->p_rtable) != 1)
1625                         ereport(ERROR,
1626                                         (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1627                         errmsg("only table \"%s\" can be referenced in check constraint",
1628                                    RelationGetRelationName(rel))));
1629
1630                 /*
1631                  * No subplans or aggregates, either...
1632                  */
1633                 if (pstate->p_hasSubLinks)
1634                         ereport(ERROR,
1635                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1636                                          errmsg("cannot use subquery in check constraint")));
1637                 if (pstate->p_hasAggs)
1638                         ereport(ERROR,
1639                                         (errcode(ERRCODE_GROUPING_ERROR),
1640                            errmsg("cannot use aggregate function in check constraint")));
1641
1642                 /*
1643                  * Check name uniqueness, or generate a name if none was given.
1644                  */
1645                 if (cdef->name != NULL)
1646                 {
1647                         ListCell   *cell2;
1648
1649                         ccname = cdef->name;
1650                         /* Check against pre-existing constraints */
1651                         if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
1652                                                                          RelationGetRelid(rel),
1653                                                                          RelationGetNamespace(rel),
1654                                                                          ccname))
1655                                 ereport(ERROR,
1656                                                 (errcode(ERRCODE_DUPLICATE_OBJECT),
1657                                 errmsg("constraint \"%s\" for relation \"%s\" already exists",
1658                                            ccname, RelationGetRelationName(rel))));
1659                         /* Check against other new constraints */
1660                         /* Needed because we don't do CommandCounterIncrement in loop */
1661                         foreach(cell2, checknames)
1662                         {
1663                                 if (strcmp((char *) lfirst(cell2), ccname) == 0)
1664                                         ereport(ERROR,
1665                                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
1666                                                          errmsg("check constraint \"%s\" already exists",
1667                                                                         ccname)));
1668                         }
1669                 }
1670                 else
1671                 {
1672                         /*
1673                          * When generating a name, we want to create "tab_col_check" for a
1674                          * column constraint and "tab_check" for a table constraint.  We
1675                          * no longer have any info about the syntactic positioning of the
1676                          * constraint phrase, so we approximate this by seeing whether the
1677                          * expression references more than one column.  (If the user
1678                          * played by the rules, the result is the same...)
1679                          *
1680                          * Note: pull_var_clause() doesn't descend into sublinks, but we
1681                          * eliminated those above; and anyway this only needs to be an
1682                          * approximate answer.
1683                          */
1684                         List       *vars;
1685                         char       *colname;
1686
1687                         vars = pull_var_clause(expr, false);
1688
1689                         /* eliminate duplicates */
1690                         vars = list_union(NIL, vars);
1691
1692                         if (list_length(vars) == 1)
1693                                 colname = get_attname(RelationGetRelid(rel),
1694                                                                           ((Var *) linitial(vars))->varattno);
1695                         else
1696                                 colname = NULL;
1697
1698                         ccname = ChooseConstraintName(RelationGetRelationName(rel),
1699                                                                                   colname,
1700                                                                                   "check",
1701                                                                                   RelationGetNamespace(rel),
1702                                                                                   checknames);
1703                 }
1704
1705                 /* save name for future checks */
1706                 checknames = lappend(checknames, ccname);
1707
1708                 /*
1709                  * OK, store it.
1710                  */
1711                 StoreRelCheck(rel, ccname, nodeToString(expr));
1712
1713                 numchecks++;
1714
1715                 cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
1716                 cooked->contype = CONSTR_CHECK;
1717                 cooked->name = ccname;
1718                 cooked->attnum = 0;
1719                 cooked->expr = expr;
1720                 cookedConstraints = lappend(cookedConstraints, cooked);
1721         }
1722
1723         /*
1724          * Update the count of constraints in the relation's pg_class tuple. We do
1725          * this even if there was no change, in order to ensure that an SI update
1726          * message is sent out for the pg_class tuple, which will force other
1727          * backends to rebuild their relcache entries for the rel. (This is
1728          * critical if we added defaults but not constraints.)
1729          */
1730         SetRelationNumChecks(rel, numchecks);
1731
1732         return cookedConstraints;
1733 }
1734
1735 /*
1736  * Update the count of constraints in the relation's pg_class tuple.
1737  *
1738  * Caller had better hold exclusive lock on the relation.
1739  *
1740  * An important side effect is that a SI update message will be sent out for
1741  * the pg_class tuple, which will force other backends to rebuild their
1742  * relcache entries for the rel.  Also, this backend will rebuild its
1743  * own relcache entry at the next CommandCounterIncrement.
1744  */
1745 static void
1746 SetRelationNumChecks(Relation rel, int numchecks)
1747 {
1748         Relation        relrel;
1749         HeapTuple       reltup;
1750         Form_pg_class relStruct;
1751
1752         relrel = heap_open(RelationRelationId, RowExclusiveLock);
1753         reltup = SearchSysCacheCopy(RELOID,
1754                                                                 ObjectIdGetDatum(RelationGetRelid(rel)),
1755                                                                 0, 0, 0);
1756         if (!HeapTupleIsValid(reltup))
1757                 elog(ERROR, "cache lookup failed for relation %u",
1758                          RelationGetRelid(rel));
1759         relStruct = (Form_pg_class) GETSTRUCT(reltup);
1760
1761         if (relStruct->relchecks != numchecks)
1762         {
1763                 relStruct->relchecks = numchecks;
1764
1765                 simple_heap_update(relrel, &reltup->t_self, reltup);
1766
1767                 /* keep catalog indexes current */
1768                 CatalogUpdateIndexes(relrel, reltup);
1769         }
1770         else
1771         {
1772                 /* Skip the disk update, but force relcache inval anyway */
1773                 CacheInvalidateRelcache(rel);
1774         }
1775
1776         heap_freetuple(reltup);
1777         heap_close(relrel, RowExclusiveLock);
1778 }
1779
1780 /*
1781  * Take a raw default and convert it to a cooked format ready for
1782  * storage.
1783  *
1784  * Parse state should be set up to recognize any vars that might appear
1785  * in the expression.  (Even though we plan to reject vars, it's more
1786  * user-friendly to give the correct error message than "unknown var".)
1787  *
1788  * If atttypid is not InvalidOid, coerce the expression to the specified
1789  * type (and typmod atttypmod).   attname is only needed in this case:
1790  * it is used in the error message, if any.
1791  */
1792 Node *
1793 cookDefault(ParseState *pstate,
1794                         Node *raw_default,
1795                         Oid atttypid,
1796                         int32 atttypmod,
1797                         char *attname)
1798 {
1799         Node       *expr;
1800
1801         Assert(raw_default != NULL);
1802
1803         /*
1804          * Transform raw parsetree to executable expression.
1805          */
1806         expr = transformExpr(pstate, raw_default);
1807
1808         /*
1809          * Make sure default expr does not refer to any vars.
1810          */
1811         if (contain_var_clause(expr))
1812                 ereport(ERROR,
1813                                 (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1814                           errmsg("cannot use column references in default expression")));
1815
1816         /*
1817          * It can't return a set either.
1818          */
1819         if (expression_returns_set(expr))
1820                 ereport(ERROR,
1821                                 (errcode(ERRCODE_DATATYPE_MISMATCH),
1822                                  errmsg("default expression must not return a set")));
1823
1824         /*
1825          * No subplans or aggregates, either...
1826          */
1827         if (pstate->p_hasSubLinks)
1828                 ereport(ERROR,
1829                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1830                                  errmsg("cannot use subquery in default expression")));
1831         if (pstate->p_hasAggs)
1832                 ereport(ERROR,
1833                                 (errcode(ERRCODE_GROUPING_ERROR),
1834                          errmsg("cannot use aggregate function in default expression")));
1835
1836         /*
1837          * Coerce the expression to the correct type and typmod, if given. This
1838          * should match the parser's processing of non-defaulted expressions ---
1839          * see transformAssignedExpr().
1840          */
1841         if (OidIsValid(atttypid))
1842         {
1843                 Oid                     type_id = exprType(expr);
1844
1845                 expr = coerce_to_target_type(pstate, expr, type_id,
1846                                                                          atttypid, atttypmod,
1847                                                                          COERCION_ASSIGNMENT,
1848                                                                          COERCE_IMPLICIT_CAST);
1849                 if (expr == NULL)
1850                         ereport(ERROR,
1851                                         (errcode(ERRCODE_DATATYPE_MISMATCH),
1852                                          errmsg("column \"%s\" is of type %s"
1853                                                         " but default expression is of type %s",
1854                                                         attname,
1855                                                         format_type_be(atttypid),
1856                                                         format_type_be(type_id)),
1857                            errhint("You will need to rewrite or cast the expression.")));
1858         }
1859
1860         return expr;
1861 }
1862
1863
1864 /*
1865  * Removes all constraints on a relation that match the given name.
1866  *
1867  * It is the responsibility of the calling function to acquire a suitable
1868  * lock on the relation.
1869  *
1870  * Returns: The number of constraints removed.
1871  */
1872 int
1873 RemoveRelConstraints(Relation rel, const char *constrName,
1874                                          DropBehavior behavior)
1875 {
1876         int                     ndeleted = 0;
1877         Relation        conrel;
1878         SysScanDesc conscan;
1879         ScanKeyData key[1];
1880         HeapTuple       contup;
1881
1882         /* Grab an appropriate lock on the pg_constraint relation */
1883         conrel = heap_open(ConstraintRelationId, RowExclusiveLock);
1884
1885         /* Use the index to scan only constraints of the target relation */
1886         ScanKeyInit(&key[0],
1887                                 Anum_pg_constraint_conrelid,
1888                                 BTEqualStrategyNumber, F_OIDEQ,
1889                                 ObjectIdGetDatum(RelationGetRelid(rel)));
1890
1891         conscan = systable_beginscan(conrel, ConstraintRelidIndexId, true,
1892                                                                  SnapshotNow, 1, key);
1893
1894         /*
1895          * Scan over the result set, removing any matching entries.
1896          */
1897         while ((contup = systable_getnext(conscan)) != NULL)
1898         {
1899                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(contup);
1900
1901                 if (strcmp(NameStr(con->conname), constrName) == 0)
1902                 {
1903                         ObjectAddress conobj;
1904
1905                         conobj.classId = ConstraintRelationId;
1906                         conobj.objectId = HeapTupleGetOid(contup);
1907                         conobj.objectSubId = 0;
1908
1909                         performDeletion(&conobj, behavior);
1910
1911                         ndeleted++;
1912                 }
1913         }
1914
1915         /* Clean up after the scan */
1916         systable_endscan(conscan);
1917         heap_close(conrel, RowExclusiveLock);
1918
1919         return ndeleted;
1920 }
1921
1922
1923 /*
1924  * RemoveStatistics --- remove entries in pg_statistic for a rel or column
1925  *
1926  * If attnum is zero, remove all entries for rel; else remove only the one
1927  * for that column.
1928  */
1929 void
1930 RemoveStatistics(Oid relid, AttrNumber attnum)
1931 {
1932         Relation        pgstatistic;
1933         SysScanDesc scan;
1934         ScanKeyData key[2];
1935         int                     nkeys;
1936         HeapTuple       tuple;
1937
1938         pgstatistic = heap_open(StatisticRelationId, RowExclusiveLock);
1939
1940         ScanKeyInit(&key[0],
1941                                 Anum_pg_statistic_starelid,
1942                                 BTEqualStrategyNumber, F_OIDEQ,
1943                                 ObjectIdGetDatum(relid));
1944
1945         if (attnum == 0)
1946                 nkeys = 1;
1947         else
1948         {
1949                 ScanKeyInit(&key[1],
1950                                         Anum_pg_statistic_staattnum,
1951                                         BTEqualStrategyNumber, F_INT2EQ,
1952                                         Int16GetDatum(attnum));
1953                 nkeys = 2;
1954         }
1955
1956         scan = systable_beginscan(pgstatistic, StatisticRelidAttnumIndexId, true,
1957                                                           SnapshotNow, nkeys, key);
1958
1959         while (HeapTupleIsValid(tuple = systable_getnext(scan)))
1960                 simple_heap_delete(pgstatistic, &tuple->t_self);
1961
1962         systable_endscan(scan);
1963
1964         heap_close(pgstatistic, RowExclusiveLock);
1965 }
1966
1967
1968 /*
1969  * RelationTruncateIndexes - truncate all indexes associated
1970  * with the heap relation to zero tuples.
1971  *
1972  * The routine will truncate and then reconstruct the indexes on
1973  * the specified relation.      Caller must hold exclusive lock on rel.
1974  */
1975 static void
1976 RelationTruncateIndexes(Relation heapRelation)
1977 {
1978         ListCell   *indlist;
1979
1980         /* Ask the relcache to produce a list of the indexes of the rel */
1981         foreach(indlist, RelationGetIndexList(heapRelation))
1982         {
1983                 Oid                     indexId = lfirst_oid(indlist);
1984                 Relation        currentIndex;
1985                 IndexInfo  *indexInfo;
1986
1987                 /* Open the index relation; use exclusive lock, just to be sure */
1988                 currentIndex = index_open(indexId, AccessExclusiveLock);
1989
1990                 /* Fetch info needed for index_build */
1991                 indexInfo = BuildIndexInfo(currentIndex);
1992
1993                 /* Now truncate the actual file (and discard buffers) */
1994                 RelationTruncate(currentIndex, 0);
1995
1996                 /* Initialize the index and rebuild */
1997                 /* Note: we do not need to re-establish pkey setting */
1998                 index_build(heapRelation, currentIndex, indexInfo, false);
1999
2000                 /* We're done with this index */
2001                 index_close(currentIndex, NoLock);
2002         }
2003 }
2004
2005 /*
2006  *       heap_truncate
2007  *
2008  *       This routine deletes all data within all the specified relations.
2009  *
2010  * This is not transaction-safe!  There is another, transaction-safe
2011  * implementation in commands/tablecmds.c.      We now use this only for
2012  * ON COMMIT truncation of temporary tables, where it doesn't matter.
2013  */
2014 void
2015 heap_truncate(List *relids)
2016 {
2017         List       *relations = NIL;
2018         ListCell   *cell;
2019
2020         /* Open relations for processing, and grab exclusive access on each */
2021         foreach(cell, relids)
2022         {
2023                 Oid                     rid = lfirst_oid(cell);
2024                 Relation        rel;
2025                 Oid                     toastrelid;
2026
2027                 rel = heap_open(rid, AccessExclusiveLock);
2028                 relations = lappend(relations, rel);
2029
2030                 /* If there is a toast table, add it to the list too */
2031                 toastrelid = rel->rd_rel->reltoastrelid;
2032                 if (OidIsValid(toastrelid))
2033                 {
2034                         rel = heap_open(toastrelid, AccessExclusiveLock);
2035                         relations = lappend(relations, rel);
2036                 }
2037         }
2038
2039         /* Don't allow truncate on tables that are referenced by foreign keys */
2040         heap_truncate_check_FKs(relations, true);
2041
2042         /* OK to do it */
2043         foreach(cell, relations)
2044         {
2045                 Relation        rel = lfirst(cell);
2046
2047                 /* Truncate the actual file (and discard buffers) */
2048                 RelationTruncate(rel, 0);
2049
2050                 /* If this relation has indexes, truncate the indexes too */
2051                 RelationTruncateIndexes(rel);
2052
2053                 /*
2054                  * Close the relation, but keep exclusive lock on it until commit.
2055                  */
2056                 heap_close(rel, NoLock);
2057         }
2058 }
2059
2060 /*
2061  * heap_truncate_check_FKs
2062  *              Check for foreign keys referencing a list of relations that
2063  *              are to be truncated, and raise error if there are any
2064  *
2065  * We disallow such FKs (except self-referential ones) since the whole point
2066  * of TRUNCATE is to not scan the individual rows to be thrown away.
2067  *
2068  * This is split out so it can be shared by both implementations of truncate.
2069  * Caller should already hold a suitable lock on the relations.
2070  *
2071  * tempTables is only used to select an appropriate error message.
2072  */
2073 void
2074 heap_truncate_check_FKs(List *relations, bool tempTables)
2075 {
2076         List       *oids = NIL;
2077         List       *dependents;
2078         ListCell   *cell;
2079
2080         /*
2081          * Build a list of OIDs of the interesting relations.
2082          *
2083          * If a relation has no triggers, then it can neither have FKs nor be
2084          * referenced by a FK from another table, so we can ignore it.
2085          */
2086         foreach(cell, relations)
2087         {
2088                 Relation        rel = lfirst(cell);
2089
2090                 if (rel->rd_rel->reltriggers != 0)
2091                         oids = lappend_oid(oids, RelationGetRelid(rel));
2092         }
2093
2094         /*
2095          * Fast path: if no relation has triggers, none has FKs either.
2096          */
2097         if (oids == NIL)
2098                 return;
2099
2100         /*
2101          * Otherwise, must scan pg_constraint.  We make one pass with all the
2102          * relations considered; if this finds nothing, then all is well.
2103          */
2104         dependents = heap_truncate_find_FKs(oids);
2105         if (dependents == NIL)
2106                 return;
2107
2108         /*
2109          * Otherwise we repeat the scan once per relation to identify a particular
2110          * pair of relations to complain about.  This is pretty slow, but
2111          * performance shouldn't matter much in a failure path.  The reason for
2112          * doing things this way is to ensure that the message produced is not
2113          * dependent on chance row locations within pg_constraint.
2114          */
2115         foreach(cell, oids)
2116         {
2117                 Oid                     relid = lfirst_oid(cell);
2118                 ListCell   *cell2;
2119
2120                 dependents = heap_truncate_find_FKs(list_make1_oid(relid));
2121
2122                 foreach(cell2, dependents)
2123                 {
2124                         Oid                     relid2 = lfirst_oid(cell2);
2125
2126                         if (!list_member_oid(oids, relid2))
2127                         {
2128                                 char       *relname = get_rel_name(relid);
2129                                 char       *relname2 = get_rel_name(relid2);
2130
2131                                 if (tempTables)
2132                                         ereport(ERROR,
2133                                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2134                                                          errmsg("unsupported ON COMMIT and foreign key combination"),
2135                                                          errdetail("Table \"%s\" references \"%s\", but they do not have the same ON COMMIT setting.",
2136                                                                            relname2, relname)));
2137                                 else
2138                                         ereport(ERROR,
2139                                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2140                                                          errmsg("cannot truncate a table referenced in a foreign key constraint"),
2141                                                          errdetail("Table \"%s\" references \"%s\".",
2142                                                                            relname2, relname),
2143                                                    errhint("Truncate table \"%s\" at the same time, "
2144                                                                    "or use TRUNCATE ... CASCADE.",
2145                                                                    relname2)));
2146                         }
2147                 }
2148         }
2149 }
2150
2151 /*
2152  * heap_truncate_find_FKs
2153  *              Find relations having foreign keys referencing any of the given rels
2154  *
2155  * Input and result are both lists of relation OIDs.  The result contains
2156  * no duplicates, does *not* include any rels that were already in the input
2157  * list, and is sorted in OID order.  (The last property is enforced mainly
2158  * to guarantee consistent behavior in the regression tests; we don't want
2159  * behavior to change depending on chance locations of rows in pg_constraint.)
2160  *
2161  * Note: caller should already have appropriate lock on all rels mentioned
2162  * in relationIds.      Since adding or dropping an FK requires exclusive lock
2163  * on both rels, this ensures that the answer will be stable.
2164  */
2165 List *
2166 heap_truncate_find_FKs(List *relationIds)
2167 {
2168         List       *result = NIL;
2169         Relation        fkeyRel;
2170         SysScanDesc fkeyScan;
2171         HeapTuple       tuple;
2172
2173         /*
2174          * Must scan pg_constraint.  Right now, it is a seqscan because there is
2175          * no available index on confrelid.
2176          */
2177         fkeyRel = heap_open(ConstraintRelationId, AccessShareLock);
2178
2179         fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
2180                                                                   SnapshotNow, 0, NULL);
2181
2182         while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
2183         {
2184                 Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
2185
2186                 /* Not a foreign key */
2187                 if (con->contype != CONSTRAINT_FOREIGN)
2188                         continue;
2189
2190                 /* Not referencing one of our list of tables */
2191                 if (!list_member_oid(relationIds, con->confrelid))
2192                         continue;
2193
2194                 /* Add referencer unless already in input or result list */
2195                 if (!list_member_oid(relationIds, con->conrelid))
2196                         result = insert_ordered_unique_oid(result, con->conrelid);
2197         }
2198
2199         systable_endscan(fkeyScan);
2200         heap_close(fkeyRel, AccessShareLock);
2201
2202         return result;
2203 }
2204
2205 /*
2206  * insert_ordered_unique_oid
2207  *              Insert a new Oid into a sorted list of Oids, preserving ordering,
2208  *              and eliminating duplicates
2209  *
2210  * Building the ordered list this way is O(N^2), but with a pretty small
2211  * constant, so for the number of entries we expect it will probably be
2212  * faster than trying to apply qsort().  It seems unlikely someone would be
2213  * trying to truncate a table with thousands of dependent tables ...
2214  */
2215 static List *
2216 insert_ordered_unique_oid(List *list, Oid datum)
2217 {
2218         ListCell   *prev;
2219
2220         /* Does the datum belong at the front? */
2221         if (list == NIL || datum < linitial_oid(list))
2222                 return lcons_oid(datum, list);
2223         /* Does it match the first entry? */
2224         if (datum == linitial_oid(list))
2225                 return list;                    /* duplicate, so don't insert */
2226         /* No, so find the entry it belongs after */
2227         prev = list_head(list);
2228         for (;;)
2229         {
2230                 ListCell   *curr = lnext(prev);
2231
2232                 if (curr == NULL || datum < lfirst_oid(curr))
2233                         break;                          /* it belongs after 'prev', before 'curr' */
2234
2235                 if (datum == lfirst_oid(curr))
2236                         return list;            /* duplicate, so don't insert */
2237
2238                 prev = curr;
2239         }
2240         /* Insert datum into list after 'prev' */
2241         lappend_cell_oid(list, prev, datum);
2242         return list;
2243 }