granicus.if.org Git - postgresql/blob - src/backend/catalog/dependency.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * dependency.c
   4  *        Routines to support inter-object dependencies.
   5  *
   6  *
   7  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
   8  * Portions Copyright (c) 1994, Regents of the University of California
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL: pgsql/src/backend/catalog/dependency.c,v 1.36 2004/05/26 04:41:06 neilc Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/genam.h"
  18 #include "access/heapam.h"
  19 #include "catalog/catname.h"
  20 #include "catalog/dependency.h"
  21 #include "catalog/heap.h"
  22 #include "catalog/index.h"
  23 #include "catalog/indexing.h"
  24 #include "catalog/namespace.h"
  25 #include "catalog/pg_attrdef.h"
  26 #include "catalog/pg_cast.h"
  27 #include "catalog/pg_constraint.h"
  28 #include "catalog/pg_conversion.h"
  29 #include "catalog/pg_depend.h"
  30 #include "catalog/pg_language.h"
  31 #include "catalog/pg_opclass.h"
  32 #include "catalog/pg_rewrite.h"
  33 #include "catalog/pg_trigger.h"
  34 #include "commands/comment.h"
  35 #include "commands/defrem.h"
  36 #include "commands/proclang.h"
  37 #include "commands/schemacmds.h"
  38 #include "commands/trigger.h"
  39 #include "commands/typecmds.h"
  40 #include "lib/stringinfo.h"
  41 #include "miscadmin.h"
  42 #include "optimizer/clauses.h"
  43 #include "parser/parsetree.h"
  44 #include "rewrite/rewriteRemove.h"
  45 #include "utils/builtins.h"
  46 #include "utils/fmgroids.h"
  47 #include "utils/lsyscache.h"
  48 #include "utils/syscache.h"
  49
  50
  51 /* expansible list of ObjectAddresses */
  52 typedef struct
  53 {
  54         ObjectAddress *refs;            /* => palloc'd array */
  55         int                     numrefs;                /* current number of references */
  56         int                     maxrefs;                /* current size of palloc'd array */
  57 } ObjectAddresses;
  58
  59 /* for find_expr_references_walker */
  60 typedef struct
  61 {
  62         ObjectAddresses addrs;          /* addresses being accumulated */
  63         List       *rtables;            /* list of rangetables to resolve Vars */
  64 } find_expr_references_context;
  65
  66
  67 /*
  68  * Because not all system catalogs have predetermined OIDs, we build a table
  69  * mapping between ObjectClasses and OIDs.      This is done at most once per
  70  * backend run, to minimize lookup overhead.
  71  */
  72 static bool object_classes_initialized = false;
  73 static Oid      object_classes[MAX_OCLASS];
  74
  75
  76 static void findAutoDeletableObjects(const ObjectAddress *object,
  77                                                  ObjectAddresses *oktodelete,
  78                                                  Relation depRel);
  79 static bool recursiveDeletion(const ObjectAddress *object,
  80                                   DropBehavior behavior,
  81                                   int msglevel,
  82                                   const ObjectAddress *callingObject,
  83                                   ObjectAddresses *oktodelete,
  84                                   Relation depRel);
  85 static bool deleteDependentObjects(const ObjectAddress *object,
  86                                            const char *objDescription,
  87                                            DropBehavior behavior,
  88                                            int msglevel,
  89                                            ObjectAddresses *oktodelete,
  90                                            Relation depRel);
  91 static void doDeletion(const ObjectAddress *object);
  92 static bool find_expr_references_walker(Node *node,
  93                                                         find_expr_references_context *context);
  94 static void eliminate_duplicate_dependencies(ObjectAddresses *addrs);
  95 static int      object_address_comparator(const void *a, const void *b);
  96 static void init_object_addresses(ObjectAddresses *addrs);
  97 static void add_object_address(ObjectClass oclass, Oid objectId, int32 subId,
  98                                    ObjectAddresses *addrs);
  99 static void add_exact_object_address(const ObjectAddress *object,
 100                                                  ObjectAddresses *addrs);
 101 static bool object_address_present(const ObjectAddress *object,
 102                                            ObjectAddresses *addrs);
 103 static void term_object_addresses(ObjectAddresses *addrs);
 104 static void init_object_classes(void);
 105 static void getRelationDescription(StringInfo buffer, Oid relid);
 106
 107
 108 /*
 109  * performDeletion: attempt to drop the specified object.  If CASCADE
 110  * behavior is specified, also drop any dependent objects (recursively).
 111  * If RESTRICT behavior is specified, error out if there are any dependent
 112  * objects, except for those that should be implicitly dropped anyway
 113  * according to the dependency type.
 114  *
 115  * This is the outer control routine for all forms of DROP that drop objects
 116  * that can participate in dependencies.
 117  */
 118 void
 119 performDeletion(const ObjectAddress *object,
 120                                 DropBehavior behavior)
 121 {
 122         char       *objDescription;
 123         Relation        depRel;
 124         ObjectAddresses oktodelete;
 125
 126         /*
 127          * Get object description for possible use in failure message. Must do
 128          * this before deleting it ...
 129          */
 130         objDescription = getObjectDescription(object);
 131
 132         /*
 133          * We save some cycles by opening pg_depend just once and passing the
 134          * Relation pointer down to all the recursive deletion steps.
 135          */
 136         depRel = heap_openr(DependRelationName, RowExclusiveLock);
 137
 138         /*
 139          * Construct a list of objects that are reachable by AUTO or INTERNAL
 140          * dependencies from the target object.  These should be deleted
 141          * silently, even if the actual deletion pass first reaches one of
 142          * them via a non-auto dependency.
 143          */
 144         init_object_addresses(&oktodelete);
 145
 146         findAutoDeletableObjects(object, &oktodelete, depRel);
 147
 148         if (!recursiveDeletion(object, behavior, NOTICE,
 149                                                    NULL, &oktodelete, depRel))
 150                 ereport(ERROR,
 151                                 (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 152                           errmsg("cannot drop %s because other objects depend on it",
 153                                          objDescription),
 154                                  errhint("Use DROP ... CASCADE to drop the dependent objects too.")));
 155
 156         term_object_addresses(&oktodelete);
 157
 158         heap_close(depRel, RowExclusiveLock);
 159
 160         pfree(objDescription);
 161 }
 162
 163
 164 /*
 165  * deleteWhatDependsOn: attempt to drop everything that depends on the
 166  * specified object, though not the object itself.      Behavior is always
 167  * CASCADE.
 168  *
 169  * This is currently used only to clean out the contents of a schema
 170  * (namespace): the passed object is a namespace.  We normally want this
 171  * to be done silently, so there's an option to suppress NOTICE messages.
 172  */
 173 void
 174 deleteWhatDependsOn(const ObjectAddress *object,
 175                                         bool showNotices)
 176 {
 177         char       *objDescription;
 178         Relation        depRel;
 179         ObjectAddresses oktodelete;
 180
 181         /*
 182          * Get object description for possible use in failure messages
 183          */
 184         objDescription = getObjectDescription(object);
 185
 186         /*
 187          * We save some cycles by opening pg_depend just once and passing the
 188          * Relation pointer down to all the recursive deletion steps.
 189          */
 190         depRel = heap_openr(DependRelationName, RowExclusiveLock);
 191
 192         /*
 193          * Construct a list of objects that are reachable by AUTO or INTERNAL
 194          * dependencies from the target object.  These should be deleted
 195          * silently, even if the actual deletion pass first reaches one of
 196          * them via a non-auto dependency.
 197          */
 198         init_object_addresses(&oktodelete);
 199
 200         findAutoDeletableObjects(object, &oktodelete, depRel);
 201
 202         /*
 203          * Now invoke only step 2 of recursiveDeletion: just recurse to the
 204          * stuff dependent on the given object.
 205          */
 206         if (!deleteDependentObjects(object, objDescription,
 207                                                                 DROP_CASCADE,
 208                                                                 showNotices ? NOTICE : DEBUG2,
 209                                                                 &oktodelete, depRel))
 210                 ereport(ERROR,
 211                                 (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 212                                  errmsg("failed to drop all objects depending on %s",
 213                                                 objDescription)));
 214
 215         /*
 216          * We do not need CommandCounterIncrement here, since if step 2 did
 217          * anything then each recursive call will have ended with one.
 218          */
 219
 220         term_object_addresses(&oktodelete);
 221
 222         heap_close(depRel, RowExclusiveLock);
 223
 224         pfree(objDescription);
 225 }
 226
 227
 228 /*
 229  * findAutoDeletableObjects: find all objects that are reachable by AUTO or
 230  * INTERNAL dependency paths from the given object.  Add them all to the
 231  * oktodelete list.  Note that the originally given object will also be
 232  * added to the list.
 233  *
 234  * depRel is the already-open pg_depend relation.
 235  */
 236 static void
 237 findAutoDeletableObjects(const ObjectAddress *object,
 238                                                  ObjectAddresses *oktodelete,
 239                                                  Relation depRel)
 240 {
 241         ScanKeyData key[3];
 242         int                     nkeys;
 243         SysScanDesc scan;
 244         HeapTuple       tup;
 245         ObjectAddress otherObject;
 246
 247         /*
 248          * If this object is already in oktodelete, then we already visited
 249          * it; don't do so again (this prevents infinite recursion if there's
 250          * a loop in pg_depend).  Otherwise, add it.
 251          */
 252         if (object_address_present(object, oktodelete))
 253                 return;
 254         add_exact_object_address(object, oktodelete);
 255
 256         /*
 257          * Scan pg_depend records that link to this object, showing the things
 258          * that depend on it.  For each one that is AUTO or INTERNAL, visit
 259          * the referencing object.
 260          *
 261          * When dropping a whole object (subId = 0), find pg_depend records for
 262          * its sub-objects too.
 263          */
 264         ScanKeyInit(&key[0],
 265                                 Anum_pg_depend_refclassid,
 266                                 BTEqualStrategyNumber, F_OIDEQ,
 267                                 ObjectIdGetDatum(object->classId));
 268         ScanKeyInit(&key[1],
 269                                 Anum_pg_depend_refobjid,
 270                                 BTEqualStrategyNumber, F_OIDEQ,
 271                                 ObjectIdGetDatum(object->objectId));
 272         if (object->objectSubId != 0)
 273         {
 274                 ScanKeyInit(&key[2],
 275                                         Anum_pg_depend_refobjsubid,
 276                                         BTEqualStrategyNumber, F_INT4EQ,
 277                                         Int32GetDatum(object->objectSubId));
 278                 nkeys = 3;
 279         }
 280         else
 281                 nkeys = 2;
 282
 283         scan = systable_beginscan(depRel, DependReferenceIndex, true,
 284                                                           SnapshotNow, nkeys, key);
 285
 286         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 287         {
 288                 Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup);
 289
 290                 switch (foundDep->deptype)
 291                 {
 292                         case DEPENDENCY_NORMAL:
 293                                 /* ignore */
 294                                 break;
 295                         case DEPENDENCY_AUTO:
 296                         case DEPENDENCY_INTERNAL:
 297                                 /* recurse */
 298                                 otherObject.classId = foundDep->classid;
 299                                 otherObject.objectId = foundDep->objid;
 300                                 otherObject.objectSubId = foundDep->objsubid;
 301                                 findAutoDeletableObjects(&otherObject, oktodelete, depRel);
 302                                 break;
 303                         case DEPENDENCY_PIN:
 304
 305                                 /*
 306                                  * For a PIN dependency we just ereport immediately; there
 307                                  * won't be any others to examine, and we aren't ever
 308                                  * going to let the user delete it.
 309                                  */
 310                                 ereport(ERROR,
 311                                                 (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 312                                                  errmsg("cannot drop %s because it is required by the database system",
 313                                                                 getObjectDescription(object))));
 314                                 break;
 315                         default:
 316                                 elog(ERROR, "unrecognized dependency type '%c' for %s",
 317                                          foundDep->deptype, getObjectDescription(object));
 318                                 break;
 319                 }
 320         }
 321
 322         systable_endscan(scan);
 323 }
 324
 325
 326 /*
 327  * recursiveDeletion: delete a single object for performDeletion, plus
 328  * (recursively) anything that depends on it.
 329  *
 330  * Returns TRUE if successful, FALSE if not.
 331  *
 332  * callingObject is NULL at the outer level, else identifies the object that
 333  * we recursed from (the reference object that someone else needs to delete).
 334  *
 335  * oktodelete is a list of objects verified deletable (ie, reachable by one
 336  * or more AUTO or INTERNAL dependencies from the original target).
 337  *
 338  * depRel is the already-open pg_depend relation.
 339  *
 340  *
 341  * In RESTRICT mode, we perform all the deletions anyway, but ereport a message
 342  * and return FALSE if we find a restriction violation.  performDeletion
 343  * will then abort the transaction to nullify the deletions.  We have to
 344  * do it this way to (a) report all the direct and indirect dependencies
 345  * while (b) not going into infinite recursion if there's a cycle.
 346  *
 347  * This is even more complex than one could wish, because it is possible for
 348  * the same pair of objects to be related by both NORMAL and AUTO/INTERNAL
 349  * dependencies.  Also, we might have a situation where we've been asked to
 350  * delete object A, and objects B and C both have AUTO dependencies on A,
 351  * but B also has a NORMAL dependency on C.  (Since any of these paths might
 352  * be indirect, we can't prevent these scenarios, but must cope instead.)
 353  * If we visit C before B then we would mistakenly decide that the B->C link
 354  * should prevent the restricted drop from occurring.  To handle this, we make
 355  * a pre-scan to find all the objects that are auto-deletable from A.  If we
 356  * visit C first, but B is present in the oktodelete list, then we make no
 357  * complaint but recurse to delete B anyway.  (Note that in general we must
 358  * delete B before deleting C; the drop routine for B may try to access C.)
 359  *
 360  * Note: in the case where the path to B is traversed first, we will not
 361  * see the NORMAL dependency when we reach C, because of the pg_depend
 362  * removals done in step 1.  The oktodelete list is necessary just
 363  * to make the behavior independent of the order in which pg_depend
 364  * entries are visited.
 365  */
 366 static bool
 367 recursiveDeletion(const ObjectAddress *object,
 368                                   DropBehavior behavior,
 369                                   int msglevel,
 370                                   const ObjectAddress *callingObject,
 371                                   ObjectAddresses *oktodelete,
 372                                   Relation depRel)
 373 {
 374         bool            ok = true;
 375         char       *objDescription;
 376         ScanKeyData key[3];
 377         int                     nkeys;
 378         SysScanDesc scan;
 379         HeapTuple       tup;
 380         ObjectAddress otherObject;
 381         ObjectAddress owningObject;
 382         bool            amOwned = false;
 383
 384         /*
 385          * Get object description for possible use in messages.  Must do this
 386          * before deleting it ...
 387          */
 388         objDescription = getObjectDescription(object);
 389
 390         /*
 391          * Step 1: find and remove pg_depend records that link from this
 392          * object to others.  We have to do this anyway, and doing it first
 393          * ensures that we avoid infinite recursion in the case of cycles.
 394          * Also, some dependency types require extra processing here.
 395          *
 396          * When dropping a whole object (subId = 0), remove all pg_depend records
 397          * for its sub-objects too.
 398          */
 399         ScanKeyInit(&key[0],
 400                                 Anum_pg_depend_classid,
 401                                 BTEqualStrategyNumber, F_OIDEQ,
 402                                 ObjectIdGetDatum(object->classId));
 403         ScanKeyInit(&key[1],
 404                                 Anum_pg_depend_objid,
 405                                 BTEqualStrategyNumber, F_OIDEQ,
 406                                 ObjectIdGetDatum(object->objectId));
 407         if (object->objectSubId != 0)
 408         {
 409                 ScanKeyInit(&key[2],
 410                                         Anum_pg_depend_objsubid,
 411                                         BTEqualStrategyNumber, F_INT4EQ,
 412                                         Int32GetDatum(object->objectSubId));
 413                 nkeys = 3;
 414         }
 415         else
 416                 nkeys = 2;
 417
 418         scan = systable_beginscan(depRel, DependDependerIndex, true,
 419                                                           SnapshotNow, nkeys, key);
 420
 421         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 422         {
 423                 Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup);
 424
 425                 otherObject.classId = foundDep->refclassid;
 426                 otherObject.objectId = foundDep->refobjid;
 427                 otherObject.objectSubId = foundDep->refobjsubid;
 428
 429                 switch (foundDep->deptype)
 430                 {
 431                         case DEPENDENCY_NORMAL:
 432                         case DEPENDENCY_AUTO:
 433                                 /* no problem */
 434                                 break;
 435                         case DEPENDENCY_INTERNAL:
 436
 437                                 /*
 438                                  * This object is part of the internal implementation of
 439                                  * another object.      We have three cases:
 440                                  *
 441                                  * 1. At the outermost recursion level, disallow the DROP.
 442                                  * (We just ereport here, rather than proceeding, since no
 443                                  * other dependencies are likely to be interesting.)
 444                                  */
 445                                 if (callingObject == NULL)
 446                                 {
 447                                         char       *otherObjDesc = getObjectDescription(&otherObject);
 448
 449                                         ereport(ERROR,
 450                                                  (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 451                                                   errmsg("cannot drop %s because %s requires it",
 452                                                                  objDescription, otherObjDesc),
 453                                                   errhint("You may drop %s instead.",
 454                                                                   otherObjDesc)));
 455                                 }
 456
 457                                 /*
 458                                  * 2. When recursing from the other end of this
 459                                  * dependency, it's okay to continue with the deletion.
 460                                  * This holds when recursing from a whole object that
 461                                  * includes the nominal other end as a component, too.
 462                                  */
 463                                 if (callingObject->classId == otherObject.classId &&
 464                                         callingObject->objectId == otherObject.objectId &&
 465                                 (callingObject->objectSubId == otherObject.objectSubId ||
 466                                  callingObject->objectSubId == 0))
 467                                         break;
 468
 469                                 /*
 470                                  * 3. When recursing from anyplace else, transform this
 471                                  * deletion request into a delete of the other object.
 472                                  * (This will be an error condition iff RESTRICT mode.) In
 473                                  * this case we finish deleting my dependencies except for
 474                                  * the INTERNAL link, which will be needed to cause the
 475                                  * owning object to recurse back to me.
 476                                  */
 477                                 if (amOwned)    /* shouldn't happen */
 478                                         elog(ERROR, "multiple INTERNAL dependencies for %s",
 479                                                  objDescription);
 480                                 owningObject = otherObject;
 481                                 amOwned = true;
 482                                 /* "continue" bypasses the simple_heap_delete call below */
 483                                 continue;
 484                         case DEPENDENCY_PIN:
 485
 486                                 /*
 487                                  * Should not happen; PIN dependencies should have zeroes
 488                                  * in the depender fields...
 489                                  */
 490                                 elog(ERROR, "incorrect use of PIN dependency with %s",
 491                                          objDescription);
 492                                 break;
 493                         default:
 494                                 elog(ERROR, "unrecognized dependency type '%c' for %s",
 495                                          foundDep->deptype, objDescription);
 496                                 break;
 497                 }
 498
 499                 simple_heap_delete(depRel, &tup->t_self);
 500         }
 501
 502         systable_endscan(scan);
 503
 504         /*
 505          * CommandCounterIncrement here to ensure that preceding changes are
 506          * all visible; in particular, that the above deletions of pg_depend
 507          * entries are visible.  That prevents infinite recursion in case of a
 508          * dependency loop (which is perfectly legal).
 509          */
 510         CommandCounterIncrement();
 511
 512         /*
 513          * If we found we are owned by another object, ask it to delete itself
 514          * instead of proceeding.  Complain if RESTRICT mode, unless the other
 515          * object is in oktodelete.
 516          */
 517         if (amOwned)
 518         {
 519                 if (object_address_present(&owningObject, oktodelete))
 520                         ereport(DEBUG2,
 521                                         (errmsg("drop auto-cascades to %s",
 522                                                         getObjectDescription(&owningObject))));
 523                 else if (behavior == DROP_RESTRICT)
 524                 {
 525                         ereport(msglevel,
 526                                         (errmsg("%s depends on %s",
 527                                                         getObjectDescription(&owningObject),
 528                                                         objDescription)));
 529                         ok = false;
 530                 }
 531                 else
 532                         ereport(msglevel,
 533                                         (errmsg("drop cascades to %s",
 534                                                         getObjectDescription(&owningObject))));
 535
 536                 if (!recursiveDeletion(&owningObject, behavior, msglevel,
 537                                                            object, oktodelete, depRel))
 538                         ok = false;
 539
 540                 pfree(objDescription);
 541
 542                 return ok;
 543         }
 544
 545         /*
 546          * Step 2: scan pg_depend records that link to this object, showing
 547          * the things that depend on it.  Recursively delete those things.
 548          * Note it's important to delete the dependent objects before the
 549          * referenced one, since the deletion routines might do things like
 550          * try to update the pg_class record when deleting a check constraint.
 551          */
 552         if (!deleteDependentObjects(object, objDescription,
 553                                                                 behavior, msglevel,
 554                                                                 oktodelete, depRel))
 555                 ok = false;
 556
 557         /*
 558          * We do not need CommandCounterIncrement here, since if step 2 did
 559          * anything then each recursive call will have ended with one.
 560          */
 561
 562         /*
 563          * Step 3: delete the object itself.
 564          */
 565         doDeletion(object);
 566
 567         /*
 568          * Delete any comments associated with this object.  (This is a
 569          * convenient place to do it instead of having every object type know
 570          * to do it.)
 571          */
 572         DeleteComments(object->objectId, object->classId, object->objectSubId);
 573
 574         /*
 575          * CommandCounterIncrement here to ensure that preceding changes are
 576          * all visible.
 577          */
 578         CommandCounterIncrement();
 579
 580         /*
 581          * And we're done!
 582          */
 583         pfree(objDescription);
 584
 585         return ok;
 586 }
 587
 588
 589 /*
 590  * deleteDependentObjects - find and delete objects that depend on 'object'
 591  *
 592  * Scan pg_depend records that link to the given object, showing
 593  * the things that depend on it.  Recursively delete those things. (We
 594  * don't delete the pg_depend records here, as the recursive call will
 595  * do that.)  Note it's important to delete the dependent objects
 596  * before the referenced one, since the deletion routines might do
 597  * things like try to update the pg_class record when deleting a check
 598  * constraint.
 599  *
 600  * When dropping a whole object (subId = 0), find pg_depend records for
 601  * its sub-objects too.
 602  *
 603  *      object: the object to find dependencies on
 604  *      objDescription: description of object (only used for error messages)
 605  *      behavior: desired drop behavior
 606  *      oktodelete: stuff that's AUTO-deletable
 607  *      depRel: already opened pg_depend relation
 608  *
 609  * Returns TRUE if all is well, false if any problem found.
 610  *
 611  * NOTE: because we are using SnapshotNow, if a recursive call deletes
 612  * any pg_depend tuples that our scan hasn't yet visited, we will not
 613  * see them as good when we do visit them.      This is essential for
 614  * correct behavior if there are multiple dependency paths between two
 615  * objects --- else we might try to delete an already-deleted object.
 616  */
 617 static bool
 618 deleteDependentObjects(const ObjectAddress *object,
 619                                            const char *objDescription,
 620                                            DropBehavior behavior,
 621                                            int msglevel,
 622                                            ObjectAddresses *oktodelete,
 623                                            Relation depRel)
 624 {
 625         bool            ok = true;
 626         ScanKeyData key[3];
 627         int                     nkeys;
 628         SysScanDesc scan;
 629         HeapTuple       tup;
 630         ObjectAddress otherObject;
 631
 632         ScanKeyInit(&key[0],
 633                                 Anum_pg_depend_refclassid,
 634                                 BTEqualStrategyNumber, F_OIDEQ,
 635                                 ObjectIdGetDatum(object->classId));
 636         ScanKeyInit(&key[1],
 637                                 Anum_pg_depend_refobjid,
 638                                 BTEqualStrategyNumber, F_OIDEQ,
 639                                 ObjectIdGetDatum(object->objectId));
 640         if (object->objectSubId != 0)
 641         {
 642                 ScanKeyInit(&key[2],
 643                                         Anum_pg_depend_refobjsubid,
 644                                         BTEqualStrategyNumber, F_INT4EQ,
 645                                         Int32GetDatum(object->objectSubId));
 646                 nkeys = 3;
 647         }
 648         else
 649                 nkeys = 2;
 650
 651         scan = systable_beginscan(depRel, DependReferenceIndex, true,
 652                                                           SnapshotNow, nkeys, key);
 653
 654         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 655         {
 656                 Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup);
 657
 658                 otherObject.classId = foundDep->classid;
 659                 otherObject.objectId = foundDep->objid;
 660                 otherObject.objectSubId = foundDep->objsubid;
 661
 662                 switch (foundDep->deptype)
 663                 {
 664                         case DEPENDENCY_NORMAL:
 665
 666                                 /*
 667                                  * Perhaps there was another dependency path that would
 668                                  * have allowed silent deletion of the otherObject, had we
 669                                  * only taken that path first. In that case, act like this
 670                                  * link is AUTO, too.
 671                                  */
 672                                 if (object_address_present(&otherObject, oktodelete))
 673                                         ereport(DEBUG2,
 674                                                         (errmsg("drop auto-cascades to %s",
 675                                                                         getObjectDescription(&otherObject))));
 676                                 else if (behavior == DROP_RESTRICT)
 677                                 {
 678                                         ereport(msglevel,
 679                                                         (errmsg("%s depends on %s",
 680                                                                         getObjectDescription(&otherObject),
 681                                                                         objDescription)));
 682                                         ok = false;
 683                                 }
 684                                 else
 685                                         ereport(msglevel,
 686                                                         (errmsg("drop cascades to %s",
 687                                                                         getObjectDescription(&otherObject))));
 688
 689                                 if (!recursiveDeletion(&otherObject, behavior, msglevel,
 690                                                                            object, oktodelete, depRel))
 691                                         ok = false;
 692                                 break;
 693                         case DEPENDENCY_AUTO:
 694                         case DEPENDENCY_INTERNAL:
 695
 696                                 /*
 697                                  * We propagate the DROP without complaint even in the
 698                                  * RESTRICT case.  (However, normal dependencies on the
 699                                  * component object could still cause failure.)
 700                                  */
 701                                 ereport(DEBUG2,
 702                                                 (errmsg("drop auto-cascades to %s",
 703                                                                 getObjectDescription(&otherObject))));
 704
 705                                 if (!recursiveDeletion(&otherObject, behavior, msglevel,
 706                                                                            object, oktodelete, depRel))
 707                                         ok = false;
 708                                 break;
 709                         case DEPENDENCY_PIN:
 710
 711                                 /*
 712                                  * For a PIN dependency we just ereport immediately; there
 713                                  * won't be any others to report.
 714                                  */
 715                                 ereport(ERROR,
 716                                                 (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 717                                                  errmsg("cannot drop %s because it is required by the database system",
 718                                                                 objDescription)));
 719                                 break;
 720                         default:
 721                                 elog(ERROR, "unrecognized dependency type '%c' for %s",
 722                                          foundDep->deptype, objDescription);
 723                                 break;
 724                 }
 725         }
 726
 727         systable_endscan(scan);
 728
 729         return ok;
 730 }
 731
 732
 733 /*
 734  * doDeletion: actually delete a single object
 735  */
 736 static void
 737 doDeletion(const ObjectAddress *object)
 738 {
 739         switch (getObjectClass(object))
 740         {
 741                 case OCLASS_CLASS:
 742                         {
 743                                 char            relKind = get_rel_relkind(object->objectId);
 744
 745                                 if (relKind == RELKIND_INDEX)
 746                                 {
 747                                         Assert(object->objectSubId == 0);
 748                                         index_drop(object->objectId);
 749                                 }
 750                                 else
 751                                 {
 752                                         if (object->objectSubId != 0)
 753                                                 RemoveAttributeById(object->objectId,
 754                                                                                         object->objectSubId);
 755                                         else
 756                                                 heap_drop_with_catalog(object->objectId);
 757                                 }
 758                                 break;
 759                         }
 760
 761                 case OCLASS_PROC:
 762                         RemoveFunctionById(object->objectId);
 763                         break;
 764
 765                 case OCLASS_TYPE:
 766                         RemoveTypeById(object->objectId);
 767                         break;
 768
 769                 case OCLASS_CAST:
 770                         DropCastById(object->objectId);
 771                         break;
 772
 773                 case OCLASS_CONSTRAINT:
 774                         RemoveConstraintById(object->objectId);
 775                         break;
 776
 777                 case OCLASS_CONVERSION:
 778                         RemoveConversionById(object->objectId);
 779                         break;
 780
 781                 case OCLASS_DEFAULT:
 782                         RemoveAttrDefaultById(object->objectId);
 783                         break;
 784
 785                 case OCLASS_LANGUAGE:
 786                         DropProceduralLanguageById(object->objectId);
 787                         break;
 788
 789                 case OCLASS_OPERATOR:
 790                         RemoveOperatorById(object->objectId);
 791                         break;
 792
 793                 case OCLASS_OPCLASS:
 794                         RemoveOpClassById(object->objectId);
 795                         break;
 796
 797                 case OCLASS_REWRITE:
 798                         RemoveRewriteRuleById(object->objectId);
 799                         break;
 800
 801                 case OCLASS_TRIGGER:
 802                         RemoveTriggerById(object->objectId);
 803                         break;
 804
 805                 case OCLASS_SCHEMA:
 806                         RemoveSchemaById(object->objectId);
 807                         break;
 808
 809                 default:
 810                         elog(ERROR, "unrecognized object class: %u",
 811                                  object->classId);
 812         }
 813 }
 814
 815 /*
 816  * recordDependencyOnExpr - find expression dependencies
 817  *
 818  * This is used to find the dependencies of rules, constraint expressions,
 819  * etc.
 820  *
 821  * Given an expression or query in node-tree form, find all the objects
 822  * it refers to (tables, columns, operators, functions, etc).  Record
 823  * a dependency of the specified type from the given depender object
 824  * to each object mentioned in the expression.
 825  *
 826  * rtable is the rangetable to be used to interpret Vars with varlevelsup=0.
 827  * It can be NIL if no such variables are expected.
 828  *
 829  * XXX is it important to create dependencies on the datatypes mentioned in
 830  * the expression?      In most cases this would be redundant (eg, a ref to an
 831  * operator indirectly references its input and output datatypes), but I'm
 832  * not quite convinced there are no cases where we need it.
 833  */
 834 void
 835 recordDependencyOnExpr(const ObjectAddress *depender,
 836                                            Node *expr, List *rtable,
 837                                            DependencyType behavior)
 838 {
 839         find_expr_references_context context;
 840
 841         init_object_addresses(&context.addrs);
 842
 843         /* Set up interpretation for Vars at varlevelsup = 0 */
 844         context.rtables = list_make1(rtable);
 845
 846         /* Scan the expression tree for referenceable objects */
 847         find_expr_references_walker(expr, &context);
 848
 849         /* Remove any duplicates */
 850         eliminate_duplicate_dependencies(&context.addrs);
 851
 852         /* And record 'em */
 853         recordMultipleDependencies(depender,
 854                                                            context.addrs.refs, context.addrs.numrefs,
 855                                                            behavior);
 856
 857         term_object_addresses(&context.addrs);
 858 }
 859
 860 /*
 861  * recordDependencyOnSingleRelExpr - find expression dependencies
 862  *
 863  * As above, but only one relation is expected to be referenced (with
 864  * varno = 1 and varlevelsup = 0).      Pass the relation OID instead of a
 865  * range table.  An additional frammish is that dependencies on that
 866  * relation (or its component columns) will be marked with 'self_behavior',
 867  * whereas 'behavior' is used for everything else.
 868  */
 869 void
 870 recordDependencyOnSingleRelExpr(const ObjectAddress *depender,
 871                                                                 Node *expr, Oid relId,
 872                                                                 DependencyType behavior,
 873                                                                 DependencyType self_behavior)
 874 {
 875         find_expr_references_context context;
 876         RangeTblEntry rte;
 877
 878         init_object_addresses(&context.addrs);
 879
 880         /* We gin up a rather bogus rangetable list to handle Vars */
 881         MemSet(&rte, 0, sizeof(rte));
 882         rte.type = T_RangeTblEntry;
 883         rte.rtekind = RTE_RELATION;
 884         rte.relid = relId;
 885
 886         context.rtables = list_make1(list_make1(&rte));
 887
 888         /* Scan the expression tree for referenceable objects */
 889         find_expr_references_walker(expr, &context);
 890
 891         /* Remove any duplicates */
 892         eliminate_duplicate_dependencies(&context.addrs);
 893
 894         /* Separate self-dependencies if necessary */
 895         if (behavior != self_behavior && context.addrs.numrefs > 0)
 896         {
 897                 ObjectAddresses self_addrs;
 898                 ObjectAddress *outobj;
 899                 int                     oldref,
 900                                         outrefs;
 901
 902                 init_object_addresses(&self_addrs);
 903
 904                 outobj = context.addrs.refs;
 905                 outrefs = 0;
 906                 for (oldref = 0; oldref < context.addrs.numrefs; oldref++)
 907                 {
 908                         ObjectAddress *thisobj = context.addrs.refs + oldref;
 909
 910                         if (thisobj->classId == RelOid_pg_class &&
 911                                 thisobj->objectId == relId)
 912                         {
 913                                 /* Move this ref into self_addrs */
 914                                 add_object_address(OCLASS_CLASS, relId, thisobj->objectSubId,
 915                                                                    &self_addrs);
 916                         }
 917                         else
 918                         {
 919                                 /* Keep it in context.addrs */
 920                                 outobj->classId = thisobj->classId;
 921                                 outobj->objectId = thisobj->objectId;
 922                                 outobj->objectSubId = thisobj->objectSubId;
 923                                 outobj++;
 924                                 outrefs++;
 925                         }
 926                 }
 927                 context.addrs.numrefs = outrefs;
 928
 929                 /* Record the self-dependencies */
 930                 recordMultipleDependencies(depender,
 931                                                                    self_addrs.refs, self_addrs.numrefs,
 932                                                                    self_behavior);
 933
 934                 term_object_addresses(&self_addrs);
 935         }
 936
 937         /* Record the external dependencies */
 938         recordMultipleDependencies(depender,
 939                                                            context.addrs.refs, context.addrs.numrefs,
 940                                                            behavior);
 941
 942         term_object_addresses(&context.addrs);
 943 }
 944
 945 /*
 946  * Recursively search an expression tree for object references.
 947  *
 948  * Note: we avoid creating references to columns of tables that participate
 949  * in an SQL JOIN construct, but are not actually used anywhere in the query.
 950  * To do so, we do not scan the joinaliasvars list of a join RTE while
 951  * scanning the query rangetable, but instead scan each individual entry
 952  * of the alias list when we find a reference to it.
 953  */
 954 static bool
 955 find_expr_references_walker(Node *node,
 956                                                         find_expr_references_context *context)
 957 {
 958         if (node == NULL)
 959                 return false;
 960         if (IsA(node, Var))
 961         {
 962                 Var                *var = (Var *) node;
 963                 List       *rtable;
 964                 RangeTblEntry *rte;
 965
 966                 /* Find matching rtable entry, or complain if not found */
 967                 if (var->varlevelsup >= list_length(context->rtables))
 968                         elog(ERROR, "invalid varlevelsup %d", var->varlevelsup);
 969                 rtable = (List *) list_nth(context->rtables, var->varlevelsup);
 970                 if (var->varno <= 0 || var->varno > list_length(rtable))
 971                         elog(ERROR, "invalid varno %d", var->varno);
 972                 rte = rt_fetch(var->varno, rtable);
 973                 if (rte->rtekind == RTE_RELATION)
 974                 {
 975                         /* If it's a plain relation, reference this column */
 976                         /* NB: this code works for whole-row Var with attno 0, too */
 977                         add_object_address(OCLASS_CLASS, rte->relid, var->varattno,
 978                                                            &context->addrs);
 979                 }
 980                 else if (rte->rtekind == RTE_JOIN)
 981                 {
 982                         /* Scan join output column to add references to join inputs */
 983                         List       *save_rtables;
 984
 985                         /* We must make the context appropriate for join's level */
 986                         save_rtables = context->rtables;
 987                         context->rtables = list_copy_tail(context->rtables,
 988                                                                                           var->varlevelsup);
 989                         if (var->varattno <= 0 ||
 990                                 var->varattno > list_length(rte->joinaliasvars))
 991                                 elog(ERROR, "invalid varattno %d", var->varattno);
 992                         find_expr_references_walker((Node *) list_nth(rte->joinaliasvars,
 993                                                                                                                   var->varattno - 1),
 994                                                                                 context);
 995                         list_free(context->rtables);
 996                         context->rtables = save_rtables;
 997                 }
 998                 return false;
 999         }
1000         if (IsA(node, FuncExpr))
1001         {
1002                 FuncExpr   *funcexpr = (FuncExpr *) node;
1003
1004                 add_object_address(OCLASS_PROC, funcexpr->funcid, 0,
1005                                                    &context->addrs);
1006                 /* fall through to examine arguments */
1007         }
1008         if (IsA(node, OpExpr))
1009         {
1010                 OpExpr     *opexpr = (OpExpr *) node;
1011
1012                 add_object_address(OCLASS_OPERATOR, opexpr->opno, 0,
1013                                                    &context->addrs);
1014                 /* fall through to examine arguments */
1015         }
1016         if (IsA(node, DistinctExpr))
1017         {
1018                 DistinctExpr *distinctexpr = (DistinctExpr *) node;
1019
1020                 add_object_address(OCLASS_OPERATOR, distinctexpr->opno, 0,
1021                                                    &context->addrs);
1022                 /* fall through to examine arguments */
1023         }
1024         if (IsA(node, ScalarArrayOpExpr))
1025         {
1026                 ScalarArrayOpExpr *opexpr = (ScalarArrayOpExpr *) node;
1027
1028                 add_object_address(OCLASS_OPERATOR, opexpr->opno, 0,
1029                                                    &context->addrs);
1030                 /* fall through to examine arguments */
1031         }
1032         if (IsA(node, NullIfExpr))
1033         {
1034                 NullIfExpr *nullifexpr = (NullIfExpr *) node;
1035
1036                 add_object_address(OCLASS_OPERATOR, nullifexpr->opno, 0,
1037                                                    &context->addrs);
1038                 /* fall through to examine arguments */
1039         }
1040         if (IsA(node, Aggref))
1041         {
1042                 Aggref     *aggref = (Aggref *) node;
1043
1044                 add_object_address(OCLASS_PROC, aggref->aggfnoid, 0,
1045                                                    &context->addrs);
1046                 /* fall through to examine arguments */
1047         }
1048         if (IsA(node, SubLink))
1049         {
1050                 SubLink    *sublink = (SubLink *) node;
1051                 ListCell   *opid;
1052
1053                 foreach(opid, sublink->operOids)
1054                 {
1055                         add_object_address(OCLASS_OPERATOR, lfirst_oid(opid), 0,
1056                                                            &context->addrs);
1057                 }
1058                 /* fall through to examine arguments */
1059         }
1060         if (is_subplan(node))
1061         {
1062                 /* Extra work needed here if we ever need this case */
1063                 elog(ERROR, "already-planned subqueries not supported");
1064         }
1065         if (IsA(node, Query))
1066         {
1067                 /* Recurse into RTE subquery or not-yet-planned sublink subquery */
1068                 Query      *query = (Query *) node;
1069                 ListCell   *rtable;
1070                 bool            result;
1071
1072                 /*
1073                  * Add whole-relation refs for each plain relation mentioned in
1074                  * the subquery's rtable.  (Note: query_tree_walker takes care of
1075                  * recursing into RTE_FUNCTION and RTE_SUBQUERY RTEs, so no need
1076                  * to do that here.  But keep it from looking at join alias
1077                  * lists.)
1078                  */
1079                 foreach(rtable, query->rtable)
1080                 {
1081                         RangeTblEntry *rte = (RangeTblEntry *) lfirst(rtable);
1082
1083                         if (rte->rtekind == RTE_RELATION)
1084                                 add_object_address(OCLASS_CLASS, rte->relid, 0,
1085                                                                    &context->addrs);
1086                 }
1087
1088                 /* Examine substructure of query */
1089                 context->rtables = lcons(query->rtable, context->rtables);
1090                 result = query_tree_walker(query,
1091                                                                    find_expr_references_walker,
1092                                                                    (void *) context,
1093                                                                    QTW_IGNORE_JOINALIASES);
1094                 context->rtables = list_delete_first(context->rtables);
1095                 return result;
1096         }
1097         return expression_tree_walker(node, find_expr_references_walker,
1098                                                                   (void *) context);
1099 }
1100
1101 /*
1102  * Given an array of dependency references, eliminate any duplicates.
1103  */
1104 static void
1105 eliminate_duplicate_dependencies(ObjectAddresses *addrs)
1106 {
1107         ObjectAddress *priorobj;
1108         int                     oldref,
1109                                 newrefs;
1110
1111         if (addrs->numrefs <= 1)
1112                 return;                                 /* nothing to do */
1113
1114         /* Sort the refs so that duplicates are adjacent */
1115         qsort((void *) addrs->refs, addrs->numrefs, sizeof(ObjectAddress),
1116                   object_address_comparator);
1117
1118         /* Remove dups */
1119         priorobj = addrs->refs;
1120         newrefs = 1;
1121         for (oldref = 1; oldref < addrs->numrefs; oldref++)
1122         {
1123                 ObjectAddress *thisobj = addrs->refs + oldref;
1124
1125                 if (priorobj->classId == thisobj->classId &&
1126                         priorobj->objectId == thisobj->objectId)
1127                 {
1128                         if (priorobj->objectSubId == thisobj->objectSubId)
1129                                 continue;               /* identical, so drop thisobj */
1130
1131                         /*
1132                          * If we have a whole-object reference and a reference to a
1133                          * part of the same object, we don't need the whole-object
1134                          * reference (for example, we don't need to reference both
1135                          * table foo and column foo.bar).  The whole-object reference
1136                          * will always appear first in the sorted list.
1137                          */
1138                         if (priorobj->objectSubId == 0)
1139                         {
1140                                 /* replace whole ref with partial */
1141                                 priorobj->objectSubId = thisobj->objectSubId;
1142                                 continue;
1143                         }
1144                 }
1145                 /* Not identical, so add thisobj to output set */
1146                 priorobj++;
1147                 priorobj->classId = thisobj->classId;
1148                 priorobj->objectId = thisobj->objectId;
1149                 priorobj->objectSubId = thisobj->objectSubId;
1150                 newrefs++;
1151         }
1152
1153         addrs->numrefs = newrefs;
1154 }
1155
1156 /*
1157  * qsort comparator for ObjectAddress items
1158  */
1159 static int
1160 object_address_comparator(const void *a, const void *b)
1161 {
1162         const ObjectAddress *obja = (const ObjectAddress *) a;
1163         const ObjectAddress *objb = (const ObjectAddress *) b;
1164
1165         if (obja->classId < objb->classId)
1166                 return -1;
1167         if (obja->classId > objb->classId)
1168                 return 1;
1169         if (obja->objectId < objb->objectId)
1170                 return -1;
1171         if (obja->objectId > objb->objectId)
1172                 return 1;
1173
1174         /*
1175          * We sort the subId as an unsigned int so that 0 will come first. See
1176          * logic in eliminate_duplicate_dependencies.
1177          */
1178         if ((unsigned int) obja->objectSubId < (unsigned int) objb->objectSubId)
1179                 return -1;
1180         if ((unsigned int) obja->objectSubId > (unsigned int) objb->objectSubId)
1181                 return 1;
1182         return 0;
1183 }
1184
1185 /*
1186  * Routines for handling an expansible array of ObjectAddress items.
1187  *
1188  * init_object_addresses: initialize an ObjectAddresses array.
1189  */
1190 static void
1191 init_object_addresses(ObjectAddresses *addrs)
1192 {
1193         /* Initialize array to empty */
1194         addrs->numrefs = 0;
1195         addrs->maxrefs = 32;            /* arbitrary initial array size */
1196         addrs->refs = (ObjectAddress *)
1197                 palloc(addrs->maxrefs * sizeof(ObjectAddress));
1198
1199         /* Initialize object_classes[] if not done yet */
1200         /* This will be needed by add_object_address() */
1201         if (!object_classes_initialized)
1202                 init_object_classes();
1203 }
1204
1205 /*
1206  * Add an entry to an ObjectAddresses array.
1207  *
1208  * It is convenient to specify the class by ObjectClass rather than directly
1209  * by catalog OID.
1210  */
1211 static void
1212 add_object_address(ObjectClass oclass, Oid objectId, int32 subId,
1213                                    ObjectAddresses *addrs)
1214 {
1215         ObjectAddress *item;
1216
1217         /* enlarge array if needed */
1218         if (addrs->numrefs >= addrs->maxrefs)
1219         {
1220                 addrs->maxrefs *= 2;
1221                 addrs->refs = (ObjectAddress *)
1222                         repalloc(addrs->refs, addrs->maxrefs * sizeof(ObjectAddress));
1223         }
1224         /* record this item */
1225         item = addrs->refs + addrs->numrefs;
1226         item->classId = object_classes[oclass];
1227         item->objectId = objectId;
1228         item->objectSubId = subId;
1229         addrs->numrefs++;
1230 }
1231
1232 /*
1233  * Add an entry to an ObjectAddresses array.
1234  *
1235  * As above, but specify entry exactly.
1236  */
1237 static void
1238 add_exact_object_address(const ObjectAddress *object,
1239                                                  ObjectAddresses *addrs)
1240 {
1241         ObjectAddress *item;
1242
1243         /* enlarge array if needed */
1244         if (addrs->numrefs >= addrs->maxrefs)
1245         {
1246                 addrs->maxrefs *= 2;
1247                 addrs->refs = (ObjectAddress *)
1248                         repalloc(addrs->refs, addrs->maxrefs * sizeof(ObjectAddress));
1249         }
1250         /* record this item */
1251         item = addrs->refs + addrs->numrefs;
1252         *item = *object;
1253         addrs->numrefs++;
1254 }
1255
1256 /*
1257  * Test whether an object is present in an ObjectAddresses array.
1258  *
1259  * We return "true" if object is a subobject of something in the array, too.
1260  */
1261 static bool
1262 object_address_present(const ObjectAddress *object,
1263                                            ObjectAddresses *addrs)
1264 {
1265         int                     i;
1266
1267         for (i = addrs->numrefs - 1; i >= 0; i--)
1268         {
1269                 ObjectAddress *thisobj = addrs->refs + i;
1270
1271                 if (object->classId == thisobj->classId &&
1272                         object->objectId == thisobj->objectId)
1273                 {
1274                         if (object->objectSubId == thisobj->objectSubId ||
1275                                 thisobj->objectSubId == 0)
1276                                 return true;
1277                 }
1278         }
1279
1280         return false;
1281 }
1282
1283 /*
1284  * Clean up when done with an ObjectAddresses array.
1285  */
1286 static void
1287 term_object_addresses(ObjectAddresses *addrs)
1288 {
1289         pfree(addrs->refs);
1290 }
1291
1292 /*
1293  * Initialize the object_classes[] table.
1294  *
1295  * Although some of these OIDs aren't compile-time constants, they surely
1296  * shouldn't change during a backend's run.  So, we look them up the
1297  * first time through and then cache them.
1298  */
1299 static void
1300 init_object_classes(void)
1301 {
1302         object_classes[OCLASS_CLASS] = RelOid_pg_class;
1303         object_classes[OCLASS_PROC] = RelOid_pg_proc;
1304         object_classes[OCLASS_TYPE] = RelOid_pg_type;
1305         object_classes[OCLASS_CAST] = get_system_catalog_relid(CastRelationName);
1306         object_classes[OCLASS_CONSTRAINT] = get_system_catalog_relid(ConstraintRelationName);
1307         object_classes[OCLASS_CONVERSION] = get_system_catalog_relid(ConversionRelationName);
1308         object_classes[OCLASS_DEFAULT] = get_system_catalog_relid(AttrDefaultRelationName);
1309         object_classes[OCLASS_LANGUAGE] = get_system_catalog_relid(LanguageRelationName);
1310         object_classes[OCLASS_OPERATOR] = get_system_catalog_relid(OperatorRelationName);
1311         object_classes[OCLASS_OPCLASS] = get_system_catalog_relid(OperatorClassRelationName);
1312         object_classes[OCLASS_REWRITE] = get_system_catalog_relid(RewriteRelationName);
1313         object_classes[OCLASS_TRIGGER] = get_system_catalog_relid(TriggerRelationName);
1314         object_classes[OCLASS_SCHEMA] = get_system_catalog_relid(NamespaceRelationName);
1315         object_classes_initialized = true;
1316 }
1317
1318 /*
1319  * Determine the class of a given object identified by objectAddress.
1320  *
1321  * This function is needed just because some of the system catalogs do
1322  * not have hardwired-at-compile-time OIDs.
1323  */
1324 ObjectClass
1325 getObjectClass(const ObjectAddress *object)
1326 {
1327         /* Easy for the bootstrapped catalogs... */
1328         switch (object->classId)
1329         {
1330                 case RelOid_pg_class:
1331                         /* caller must check objectSubId */
1332                         return OCLASS_CLASS;
1333
1334                 case RelOid_pg_proc:
1335                         Assert(object->objectSubId == 0);
1336                         return OCLASS_PROC;
1337
1338                 case RelOid_pg_type:
1339                         Assert(object->objectSubId == 0);
1340                         return OCLASS_TYPE;
1341         }
1342
1343         /*
1344          * Handle cases where catalog's OID is not hardwired.
1345          */
1346         if (!object_classes_initialized)
1347                 init_object_classes();
1348
1349         if (object->classId == object_classes[OCLASS_CAST])
1350         {
1351                 Assert(object->objectSubId == 0);
1352                 return OCLASS_CAST;
1353         }
1354         if (object->classId == object_classes[OCLASS_CONSTRAINT])
1355         {
1356                 Assert(object->objectSubId == 0);
1357                 return OCLASS_CONSTRAINT;
1358         }
1359         if (object->classId == object_classes[OCLASS_CONVERSION])
1360         {
1361                 Assert(object->objectSubId == 0);
1362                 return OCLASS_CONVERSION;
1363         }
1364         if (object->classId == object_classes[OCLASS_DEFAULT])
1365         {
1366                 Assert(object->objectSubId == 0);
1367                 return OCLASS_DEFAULT;
1368         }
1369         if (object->classId == object_classes[OCLASS_LANGUAGE])
1370         {
1371                 Assert(object->objectSubId == 0);
1372                 return OCLASS_LANGUAGE;
1373         }
1374         if (object->classId == object_classes[OCLASS_OPERATOR])
1375         {
1376                 Assert(object->objectSubId == 0);
1377                 return OCLASS_OPERATOR;
1378         }
1379         if (object->classId == object_classes[OCLASS_OPCLASS])
1380         {
1381                 Assert(object->objectSubId == 0);
1382                 return OCLASS_OPCLASS;
1383         }
1384         if (object->classId == object_classes[OCLASS_REWRITE])
1385         {
1386                 Assert(object->objectSubId == 0);
1387                 return OCLASS_REWRITE;
1388         }
1389         if (object->classId == object_classes[OCLASS_TRIGGER])
1390         {
1391                 Assert(object->objectSubId == 0);
1392                 return OCLASS_TRIGGER;
1393         }
1394         if (object->classId == object_classes[OCLASS_SCHEMA])
1395         {
1396                 Assert(object->objectSubId == 0);
1397                 return OCLASS_SCHEMA;
1398         }
1399
1400         elog(ERROR, "unrecognized object class: %u", object->classId);
1401         return OCLASS_CLASS;            /* keep compiler quiet */
1402 }
1403
1404 /*
1405  * getObjectDescription: build an object description for messages
1406  *
1407  * The result is a palloc'd string.
1408  */
1409 char *
1410 getObjectDescription(const ObjectAddress *object)
1411 {
1412         StringInfoData buffer;
1413
1414         initStringInfo(&buffer);
1415
1416         switch (getObjectClass(object))
1417         {
1418                 case OCLASS_CLASS:
1419                         getRelationDescription(&buffer, object->objectId);
1420                         if (object->objectSubId != 0)
1421                                 appendStringInfo(&buffer, gettext(" column %s"),
1422                                                                  get_relid_attribute_name(object->objectId,
1423                                                                                                                   object->objectSubId));
1424                         break;
1425
1426                 case OCLASS_PROC:
1427                         appendStringInfo(&buffer, gettext("function %s"),
1428                                                          format_procedure(object->objectId));
1429                         break;
1430
1431                 case OCLASS_TYPE:
1432                         appendStringInfo(&buffer, gettext("type %s"),
1433                                                          format_type_be(object->objectId));
1434                         break;
1435
1436                 case OCLASS_CAST:
1437                         {
1438                                 Relation        castDesc;
1439                                 ScanKeyData skey[1];
1440                                 SysScanDesc rcscan;
1441                                 HeapTuple       tup;
1442                                 Form_pg_cast castForm;
1443
1444                                 castDesc = heap_openr(CastRelationName, AccessShareLock);
1445
1446                                 ScanKeyInit(&skey[0],
1447                                                         ObjectIdAttributeNumber,
1448                                                         BTEqualStrategyNumber, F_OIDEQ,
1449                                                         ObjectIdGetDatum(object->objectId));
1450
1451                                 rcscan = systable_beginscan(castDesc, CastOidIndex, true,
1452                                                                                         SnapshotNow, 1, skey);
1453
1454                                 tup = systable_getnext(rcscan);
1455
1456                                 if (!HeapTupleIsValid(tup))
1457                                         elog(ERROR, "could not find tuple for cast %u",
1458                                                  object->objectId);
1459
1460                                 castForm = (Form_pg_cast) GETSTRUCT(tup);
1461
1462                                 appendStringInfo(&buffer, gettext("cast from %s to %s"),
1463                                                                  format_type_be(castForm->castsource),
1464                                                                  format_type_be(castForm->casttarget));
1465
1466                                 systable_endscan(rcscan);
1467                                 heap_close(castDesc, AccessShareLock);
1468                                 break;
1469                         }
1470
1471                 case OCLASS_CONSTRAINT:
1472                         {
1473                                 Relation        conDesc;
1474                                 ScanKeyData skey[1];
1475                                 SysScanDesc rcscan;
1476                                 HeapTuple       tup;
1477                                 Form_pg_constraint con;
1478
1479                                 conDesc = heap_openr(ConstraintRelationName, AccessShareLock);
1480
1481                                 ScanKeyInit(&skey[0],
1482                                                         ObjectIdAttributeNumber,
1483                                                         BTEqualStrategyNumber, F_OIDEQ,
1484                                                         ObjectIdGetDatum(object->objectId));
1485
1486                                 rcscan = systable_beginscan(conDesc, ConstraintOidIndex, true,
1487                                                                                         SnapshotNow, 1, skey);
1488
1489                                 tup = systable_getnext(rcscan);
1490
1491                                 if (!HeapTupleIsValid(tup))
1492                                         elog(ERROR, "could not find tuple for constraint %u",
1493                                                  object->objectId);
1494
1495                                 con = (Form_pg_constraint) GETSTRUCT(tup);
1496
1497                                 if (OidIsValid(con->conrelid))
1498                                 {
1499                                         appendStringInfo(&buffer, gettext("constraint %s on "),
1500                                                                          NameStr(con->conname));
1501                                         getRelationDescription(&buffer, con->conrelid);
1502                                 }
1503                                 else
1504                                 {
1505                                         appendStringInfo(&buffer, gettext("constraint %s"),
1506                                                                          NameStr(con->conname));
1507                                 }
1508
1509                                 systable_endscan(rcscan);
1510                                 heap_close(conDesc, AccessShareLock);
1511                                 break;
1512                         }
1513
1514                 case OCLASS_CONVERSION:
1515                         {
1516                                 HeapTuple       conTup;
1517
1518                                 conTup = SearchSysCache(CONOID,
1519                                                                           ObjectIdGetDatum(object->objectId),
1520                                                                                 0, 0, 0);
1521                                 if (!HeapTupleIsValid(conTup))
1522                                         elog(ERROR, "cache lookup failed for conversion %u",
1523                                                  object->objectId);
1524                                 appendStringInfo(&buffer, gettext("conversion %s"),
1525                                                                  NameStr(((Form_pg_conversion) GETSTRUCT(conTup))->conname));
1526                                 ReleaseSysCache(conTup);
1527                                 break;
1528                         }
1529
1530                 case OCLASS_DEFAULT:
1531                         {
1532                                 Relation        attrdefDesc;
1533                                 ScanKeyData skey[1];
1534                                 SysScanDesc adscan;
1535                                 HeapTuple       tup;
1536                                 Form_pg_attrdef attrdef;
1537                                 ObjectAddress colobject;
1538
1539                                 attrdefDesc = heap_openr(AttrDefaultRelationName, AccessShareLock);
1540
1541                                 ScanKeyInit(&skey[0],
1542                                                         ObjectIdAttributeNumber,
1543                                                         BTEqualStrategyNumber, F_OIDEQ,
1544                                                         ObjectIdGetDatum(object->objectId));
1545
1546                                 adscan = systable_beginscan(attrdefDesc, AttrDefaultOidIndex,
1547                                                                                         true, SnapshotNow, 1, skey);
1548
1549                                 tup = systable_getnext(adscan);
1550
1551                                 if (!HeapTupleIsValid(tup))
1552                                         elog(ERROR, "could not find tuple for attrdef %u",
1553                                                  object->objectId);
1554
1555                                 attrdef = (Form_pg_attrdef) GETSTRUCT(tup);
1556
1557                                 colobject.classId = RelOid_pg_class;
1558                                 colobject.objectId = attrdef->adrelid;
1559                                 colobject.objectSubId = attrdef->adnum;
1560
1561                                 appendStringInfo(&buffer, gettext("default for %s"),
1562                                                                  getObjectDescription(&colobject));
1563
1564                                 systable_endscan(adscan);
1565                                 heap_close(attrdefDesc, AccessShareLock);
1566                                 break;
1567                         }
1568
1569                 case OCLASS_LANGUAGE:
1570                         {
1571                                 HeapTuple       langTup;
1572
1573                                 langTup = SearchSysCache(LANGOID,
1574                                                                           ObjectIdGetDatum(object->objectId),
1575                                                                                  0, 0, 0);
1576                                 if (!HeapTupleIsValid(langTup))
1577                                         elog(ERROR, "cache lookup failed for language %u",
1578                                                  object->objectId);
1579                                 appendStringInfo(&buffer, gettext("language %s"),
1580                                                                  NameStr(((Form_pg_language) GETSTRUCT(langTup))->lanname));
1581                                 ReleaseSysCache(langTup);
1582                                 break;
1583                         }
1584
1585                 case OCLASS_OPERATOR:
1586                         appendStringInfo(&buffer, gettext("operator %s"),
1587                                                          format_operator(object->objectId));
1588                         break;
1589
1590                 case OCLASS_OPCLASS:
1591                         {
1592                                 HeapTuple       opcTup;
1593                                 Form_pg_opclass opcForm;
1594                                 HeapTuple       amTup;
1595                                 Form_pg_am      amForm;
1596                                 char       *nspname;
1597
1598                                 opcTup = SearchSysCache(CLAOID,
1599                                                                           ObjectIdGetDatum(object->objectId),
1600                                                                                 0, 0, 0);
1601                                 if (!HeapTupleIsValid(opcTup))
1602                                         elog(ERROR, "cache lookup failed for opclass %u",
1603                                                  object->objectId);
1604                                 opcForm = (Form_pg_opclass) GETSTRUCT(opcTup);
1605
1606                                 amTup = SearchSysCache(AMOID,
1607                                                                            ObjectIdGetDatum(opcForm->opcamid),
1608                                                                            0, 0, 0);
1609                                 if (!HeapTupleIsValid(amTup))
1610                                         elog(ERROR, "cache lookup failed for access method %u",
1611                                                  opcForm->opcamid);
1612                                 amForm = (Form_pg_am) GETSTRUCT(amTup);
1613
1614                                 /* Qualify the name if not visible in search path */
1615                                 if (OpclassIsVisible(object->objectId))
1616                                         nspname = NULL;
1617                                 else
1618                                         nspname = get_namespace_name(opcForm->opcnamespace);
1619
1620                                 appendStringInfo(&buffer, gettext("operator class %s for %s"),
1621                                                                  quote_qualified_identifier(nspname,
1622                                                                                          NameStr(opcForm->opcname)),
1623                                                                  NameStr(amForm->amname));
1624
1625                                 ReleaseSysCache(amTup);
1626                                 ReleaseSysCache(opcTup);
1627                                 break;
1628                         }
1629
1630                 case OCLASS_REWRITE:
1631                         {
1632                                 Relation        ruleDesc;
1633                                 ScanKeyData skey[1];
1634                                 SysScanDesc rcscan;
1635                                 HeapTuple       tup;
1636                                 Form_pg_rewrite rule;
1637
1638                                 ruleDesc = heap_openr(RewriteRelationName, AccessShareLock);
1639
1640                                 ScanKeyInit(&skey[0],
1641                                                         ObjectIdAttributeNumber,
1642                                                         BTEqualStrategyNumber, F_OIDEQ,
1643                                                         ObjectIdGetDatum(object->objectId));
1644
1645                                 rcscan = systable_beginscan(ruleDesc, RewriteOidIndex, true,
1646                                                                                         SnapshotNow, 1, skey);
1647
1648                                 tup = systable_getnext(rcscan);
1649
1650                                 if (!HeapTupleIsValid(tup))
1651                                         elog(ERROR, "could not find tuple for rule %u",
1652                                                  object->objectId);
1653
1654                                 rule = (Form_pg_rewrite) GETSTRUCT(tup);
1655
1656                                 appendStringInfo(&buffer, gettext("rule %s on "),
1657                                                                  NameStr(rule->rulename));
1658                                 getRelationDescription(&buffer, rule->ev_class);
1659
1660                                 systable_endscan(rcscan);
1661                                 heap_close(ruleDesc, AccessShareLock);
1662                                 break;
1663                         }
1664
1665                 case OCLASS_TRIGGER:
1666                         {
1667                                 Relation        trigDesc;
1668                                 ScanKeyData skey[1];
1669                                 SysScanDesc tgscan;
1670                                 HeapTuple       tup;
1671                                 Form_pg_trigger trig;
1672
1673                                 trigDesc = heap_openr(TriggerRelationName, AccessShareLock);
1674
1675                                 ScanKeyInit(&skey[0],
1676                                                         ObjectIdAttributeNumber,
1677                                                         BTEqualStrategyNumber, F_OIDEQ,
1678                                                         ObjectIdGetDatum(object->objectId));
1679
1680                                 tgscan = systable_beginscan(trigDesc, TriggerOidIndex, true,
1681                                                                                         SnapshotNow, 1, skey);
1682
1683                                 tup = systable_getnext(tgscan);
1684
1685                                 if (!HeapTupleIsValid(tup))
1686                                         elog(ERROR, "could not find tuple for trigger %u",
1687                                                  object->objectId);
1688
1689                                 trig = (Form_pg_trigger) GETSTRUCT(tup);
1690
1691                                 appendStringInfo(&buffer, gettext("trigger %s on "),
1692                                                                  NameStr(trig->tgname));
1693                                 getRelationDescription(&buffer, trig->tgrelid);
1694
1695                                 systable_endscan(tgscan);
1696                                 heap_close(trigDesc, AccessShareLock);
1697                                 break;
1698                         }
1699
1700                 case OCLASS_SCHEMA:
1701                         {
1702                                 char       *nspname;
1703
1704                                 nspname = get_namespace_name(object->objectId);
1705                                 if (!nspname)
1706                                         elog(ERROR, "cache lookup failed for namespace %u",
1707                                                  object->objectId);
1708                                 appendStringInfo(&buffer, gettext("schema %s"), nspname);
1709                                 break;
1710                         }
1711
1712                 default:
1713                         appendStringInfo(&buffer, "unrecognized object %u %u %d",
1714                                                          object->classId,
1715                                                          object->objectId,
1716                                                          object->objectSubId);
1717                         break;
1718         }
1719
1720         return buffer.data;
1721 }
1722
1723 /*
1724  * subroutine for getObjectDescription: describe a relation
1725  */
1726 static void
1727 getRelationDescription(StringInfo buffer, Oid relid)
1728 {
1729         HeapTuple       relTup;
1730         Form_pg_class relForm;
1731         char       *nspname;
1732         char       *relname;
1733
1734         relTup = SearchSysCache(RELOID,
1735                                                         ObjectIdGetDatum(relid),
1736                                                         0, 0, 0);
1737         if (!HeapTupleIsValid(relTup))
1738                 elog(ERROR, "cache lookup failed for relation %u", relid);
1739         relForm = (Form_pg_class) GETSTRUCT(relTup);
1740
1741         /* Qualify the name if not visible in search path */
1742         if (RelationIsVisible(relid))
1743                 nspname = NULL;
1744         else
1745                 nspname = get_namespace_name(relForm->relnamespace);
1746
1747         relname = quote_qualified_identifier(nspname, NameStr(relForm->relname));
1748
1749         switch (relForm->relkind)
1750         {
1751                 case RELKIND_RELATION:
1752                         appendStringInfo(buffer, gettext("table %s"),
1753                                                          relname);
1754                         break;
1755                 case RELKIND_INDEX:
1756                         appendStringInfo(buffer, gettext("index %s"),
1757                                                          relname);
1758                         break;
1759                 case RELKIND_SPECIAL:
1760                         appendStringInfo(buffer, gettext("special system relation %s"),
1761                                                          relname);
1762                         break;
1763                 case RELKIND_SEQUENCE:
1764                         appendStringInfo(buffer, gettext("sequence %s"),
1765                                                          relname);
1766                         break;
1767                 case RELKIND_UNCATALOGED:
1768                         appendStringInfo(buffer, gettext("uncataloged table %s"),
1769                                                          relname);
1770                         break;
1771                 case RELKIND_TOASTVALUE:
1772                         appendStringInfo(buffer, gettext("toast table %s"),
1773                                                          relname);
1774                         break;
1775                 case RELKIND_VIEW:
1776                         appendStringInfo(buffer, gettext("view %s"),
1777                                                          relname);
1778                         break;
1779                 case RELKIND_COMPOSITE_TYPE:
1780                         appendStringInfo(buffer, gettext("composite type %s"),
1781                                                          relname);
1782                         break;
1783                 default:
1784                         /* shouldn't get here */
1785                         appendStringInfo(buffer, gettext("relation %s"),
1786                                                          relname);
1787                         break;
1788         }
1789
1790         ReleaseSysCache(relTup);
1791 }