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