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