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-2010, 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.96 2010/02/26 02:00:36 momjian Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/genam.h"
  18 #include "access/heapam.h"
  19 #include "access/sysattr.h"
  20 #include "access/xact.h"
  21 #include "catalog/dependency.h"
  22 #include "catalog/heap.h"
  23 #include "catalog/index.h"
  24 #include "catalog/indexing.h"
  25 #include "catalog/namespace.h"
  26 #include "catalog/pg_amop.h"
  27 #include "catalog/pg_amproc.h"
  28 #include "catalog/pg_attrdef.h"
  29 #include "catalog/pg_authid.h"
  30 #include "catalog/pg_cast.h"
  31 #include "catalog/pg_constraint.h"
  32 #include "catalog/pg_conversion.h"
  33 #include "catalog/pg_conversion_fn.h"
  34 #include "catalog/pg_database.h"
  35 #include "catalog/pg_default_acl.h"
  36 #include "catalog/pg_depend.h"
  37 #include "catalog/pg_foreign_data_wrapper.h"
  38 #include "catalog/pg_foreign_server.h"
  39 #include "catalog/pg_language.h"
  40 #include "catalog/pg_largeobject.h"
  41 #include "catalog/pg_namespace.h"
  42 #include "catalog/pg_opclass.h"
  43 #include "catalog/pg_operator.h"
  44 #include "catalog/pg_opfamily.h"
  45 #include "catalog/pg_proc.h"
  46 #include "catalog/pg_rewrite.h"
  47 #include "catalog/pg_tablespace.h"
  48 #include "catalog/pg_trigger.h"
  49 #include "catalog/pg_ts_config.h"
  50 #include "catalog/pg_ts_dict.h"
  51 #include "catalog/pg_ts_parser.h"
  52 #include "catalog/pg_ts_template.h"
  53 #include "catalog/pg_type.h"
  54 #include "catalog/pg_user_mapping.h"
  55 #include "commands/comment.h"
  56 #include "commands/dbcommands.h"
  57 #include "commands/defrem.h"
  58 #include "commands/proclang.h"
  59 #include "commands/schemacmds.h"
  60 #include "commands/tablespace.h"
  61 #include "commands/trigger.h"
  62 #include "commands/typecmds.h"
  63 #include "foreign/foreign.h"
  64 #include "miscadmin.h"
  65 #include "nodes/nodeFuncs.h"
  66 #include "parser/parsetree.h"
  67 #include "rewrite/rewriteRemove.h"
  68 #include "storage/lmgr.h"
  69 #include "utils/acl.h"
  70 #include "utils/builtins.h"
  71 #include "utils/fmgroids.h"
  72 #include "utils/guc.h"
  73 #include "utils/lsyscache.h"
  74 #include "utils/syscache.h"
  75 #include "utils/tqual.h"
  76
  77
  78 /*
  79  * Deletion processing requires additional state for each ObjectAddress that
  80  * it's planning to delete.  For simplicity and code-sharing we make the
  81  * ObjectAddresses code support arrays with or without this extra state.
  82  */
  83 typedef struct
  84 {
  85         int                     flags;                  /* bitmask, see bit definitions below */
  86         ObjectAddress dependee;         /* object whose deletion forced this one */
  87 } ObjectAddressExtra;
  88
  89 /* ObjectAddressExtra flag bits */
  90 #define DEPFLAG_ORIGINAL        0x0001          /* an original deletion target */
  91 #define DEPFLAG_NORMAL          0x0002          /* reached via normal dependency */
  92 #define DEPFLAG_AUTO            0x0004          /* reached via auto dependency */
  93 #define DEPFLAG_INTERNAL        0x0008          /* reached via internal dependency */
  94
  95
  96 /* expansible list of ObjectAddresses */
  97 struct ObjectAddresses
  98 {
  99         ObjectAddress *refs;            /* => palloc'd array */
 100         ObjectAddressExtra *extras; /* => palloc'd array, or NULL if not used */
 101         int                     numrefs;                /* current number of references */
 102         int                     maxrefs;                /* current size of palloc'd array(s) */
 103 };
 104
 105 /* typedef ObjectAddresses appears in dependency.h */
 106
 107 /* threaded list of ObjectAddresses, for recursion detection */
 108 typedef struct ObjectAddressStack
 109 {
 110         const ObjectAddress *object;    /* object being visited */
 111         int                     flags;                  /* its current flag bits */
 112         struct ObjectAddressStack *next;        /* next outer stack level */
 113 } ObjectAddressStack;
 114
 115 /* for find_expr_references_walker */
 116 typedef struct
 117 {
 118         ObjectAddresses *addrs;         /* addresses being accumulated */
 119         List       *rtables;            /* list of rangetables to resolve Vars */
 120 } find_expr_references_context;
 121
 122 /*
 123  * This constant table maps ObjectClasses to the corresponding catalog OIDs.
 124  * See also getObjectClass().
 125  */
 126 static const Oid object_classes[MAX_OCLASS] = {
 127         RelationRelationId,                     /* OCLASS_CLASS */
 128         ProcedureRelationId,            /* OCLASS_PROC */
 129         TypeRelationId,                         /* OCLASS_TYPE */
 130         CastRelationId,                         /* OCLASS_CAST */
 131         ConstraintRelationId,           /* OCLASS_CONSTRAINT */
 132         ConversionRelationId,           /* OCLASS_CONVERSION */
 133         AttrDefaultRelationId,          /* OCLASS_DEFAULT */
 134         LanguageRelationId,                     /* OCLASS_LANGUAGE */
 135         LargeObjectRelationId,          /* OCLASS_LARGEOBJECT */
 136         OperatorRelationId,                     /* OCLASS_OPERATOR */
 137         OperatorClassRelationId,        /* OCLASS_OPCLASS */
 138         OperatorFamilyRelationId,       /* OCLASS_OPFAMILY */
 139         AccessMethodOperatorRelationId,         /* OCLASS_AMOP */
 140         AccessMethodProcedureRelationId,        /* OCLASS_AMPROC */
 141         RewriteRelationId,                      /* OCLASS_REWRITE */
 142         TriggerRelationId,                      /* OCLASS_TRIGGER */
 143         NamespaceRelationId,            /* OCLASS_SCHEMA */
 144         TSParserRelationId,                     /* OCLASS_TSPARSER */
 145         TSDictionaryRelationId,         /* OCLASS_TSDICT */
 146         TSTemplateRelationId,           /* OCLASS_TSTEMPLATE */
 147         TSConfigRelationId,                     /* OCLASS_TSCONFIG */
 148         AuthIdRelationId,                       /* OCLASS_ROLE */
 149         DatabaseRelationId,                     /* OCLASS_DATABASE */
 150         TableSpaceRelationId,           /* OCLASS_TBLSPACE */
 151         ForeignDataWrapperRelationId,           /* OCLASS_FDW */
 152         ForeignServerRelationId,        /* OCLASS_FOREIGN_SERVER */
 153         UserMappingRelationId,          /* OCLASS_USER_MAPPING */
 154         DefaultAclRelationId            /* OCLASS_DEFACL */
 155 };
 156
 157
 158 static void findDependentObjects(const ObjectAddress *object,
 159                                          int flags,
 160                                          ObjectAddressStack *stack,
 161                                          ObjectAddresses *targetObjects,
 162                                          const ObjectAddresses *pendingObjects,
 163                                          Relation depRel);
 164 static void reportDependentObjects(const ObjectAddresses *targetObjects,
 165                                            DropBehavior behavior,
 166                                            int msglevel,
 167                                            const ObjectAddress *origObject);
 168 static void deleteOneObject(const ObjectAddress *object, Relation depRel);
 169 static void doDeletion(const ObjectAddress *object);
 170 static void AcquireDeletionLock(const ObjectAddress *object);
 171 static void ReleaseDeletionLock(const ObjectAddress *object);
 172 static bool find_expr_references_walker(Node *node,
 173                                                         find_expr_references_context *context);
 174 static void eliminate_duplicate_dependencies(ObjectAddresses *addrs);
 175 static int      object_address_comparator(const void *a, const void *b);
 176 static void add_object_address(ObjectClass oclass, Oid objectId, int32 subId,
 177                                    ObjectAddresses *addrs);
 178 static void add_exact_object_address_extra(const ObjectAddress *object,
 179                                                            const ObjectAddressExtra *extra,
 180                                                            ObjectAddresses *addrs);
 181 static bool object_address_present_add_flags(const ObjectAddress *object,
 182                                                                  int flags,
 183                                                                  ObjectAddresses *addrs);
 184 static void getRelationDescription(StringInfo buffer, Oid relid);
 185 static void getOpFamilyDescription(StringInfo buffer, Oid opfid);
 186
 187
 188 /*
 189  * performDeletion: attempt to drop the specified object.  If CASCADE
 190  * behavior is specified, also drop any dependent objects (recursively).
 191  * If RESTRICT behavior is specified, error out if there are any dependent
 192  * objects, except for those that should be implicitly dropped anyway
 193  * according to the dependency type.
 194  *
 195  * This is the outer control routine for all forms of DROP that drop objects
 196  * that can participate in dependencies.  Note that the next two routines
 197  * are variants on the same theme; if you change anything here you'll likely
 198  * need to fix them too.
 199  */
 200 void
 201 performDeletion(const ObjectAddress *object,
 202                                 DropBehavior behavior)
 203 {
 204         Relation        depRel;
 205         ObjectAddresses *targetObjects;
 206         int                     i;
 207
 208         /*
 209          * We save some cycles by opening pg_depend just once and passing the
 210          * Relation pointer down to all the recursive deletion steps.
 211          */
 212         depRel = heap_open(DependRelationId, RowExclusiveLock);
 213
 214         /*
 215          * Acquire deletion lock on the target object.  (Ideally the caller has
 216          * done this already, but many places are sloppy about it.)
 217          */
 218         AcquireDeletionLock(object);
 219
 220         /*
 221          * Construct a list of objects to delete (ie, the given object plus
 222          * everything directly or indirectly dependent on it).
 223          */
 224         targetObjects = new_object_addresses();
 225
 226         findDependentObjects(object,
 227                                                  DEPFLAG_ORIGINAL,
 228                                                  NULL,  /* empty stack */
 229                                                  targetObjects,
 230                                                  NULL,  /* no pendingObjects */
 231                                                  depRel);
 232
 233         /*
 234          * Check if deletion is allowed, and report about cascaded deletes.
 235          */
 236         reportDependentObjects(targetObjects,
 237                                                    behavior,
 238                                                    NOTICE,
 239                                                    object);
 240
 241         /*
 242          * Delete all the objects in the proper order.
 243          */
 244         for (i = 0; i < targetObjects->numrefs; i++)
 245         {
 246                 ObjectAddress *thisobj = targetObjects->refs + i;
 247
 248                 deleteOneObject(thisobj, depRel);
 249         }
 250
 251         /* And clean up */
 252         free_object_addresses(targetObjects);
 253
 254         heap_close(depRel, RowExclusiveLock);
 255 }
 256
 257 /*
 258  * performMultipleDeletions: Similar to performDeletion, but act on multiple
 259  * objects at once.
 260  *
 261  * The main difference from issuing multiple performDeletion calls is that the
 262  * list of objects that would be implicitly dropped, for each object to be
 263  * dropped, is the union of the implicit-object list for all objects.  This
 264  * makes each check be more relaxed.
 265  */
 266 void
 267 performMultipleDeletions(const ObjectAddresses *objects,
 268                                                  DropBehavior behavior)
 269 {
 270         Relation        depRel;
 271         ObjectAddresses *targetObjects;
 272         int                     i;
 273
 274         /* No work if no objects... */
 275         if (objects->numrefs <= 0)
 276                 return;
 277
 278         /*
 279          * We save some cycles by opening pg_depend just once and passing the
 280          * Relation pointer down to all the recursive deletion steps.
 281          */
 282         depRel = heap_open(DependRelationId, RowExclusiveLock);
 283
 284         /*
 285          * Construct a list of objects to delete (ie, the given objects plus
 286          * everything directly or indirectly dependent on them).  Note that
 287          * because we pass the whole objects list as pendingObjects context, we
 288          * won't get a failure from trying to delete an object that is internally
 289          * dependent on another one in the list; we'll just skip that object and
 290          * delete it when we reach its owner.
 291          */
 292         targetObjects = new_object_addresses();
 293
 294         for (i = 0; i < objects->numrefs; i++)
 295         {
 296                 const ObjectAddress *thisobj = objects->refs + i;
 297
 298                 /*
 299                  * Acquire deletion lock on each target object.  (Ideally the caller
 300                  * has done this already, but many places are sloppy about it.)
 301                  */
 302                 AcquireDeletionLock(thisobj);
 303
 304                 findDependentObjects(thisobj,
 305                                                          DEPFLAG_ORIGINAL,
 306                                                          NULL,          /* empty stack */
 307                                                          targetObjects,
 308                                                          objects,
 309                                                          depRel);
 310         }
 311
 312         /*
 313          * Check if deletion is allowed, and report about cascaded deletes.
 314          *
 315          * If there's exactly one object being deleted, report it the same way as
 316          * in performDeletion(), else we have to be vaguer.
 317          */
 318         reportDependentObjects(targetObjects,
 319                                                    behavior,
 320                                                    NOTICE,
 321                                                    (objects->numrefs == 1 ? objects->refs : NULL));
 322
 323         /*
 324          * Delete all the objects in the proper order.
 325          */
 326         for (i = 0; i < targetObjects->numrefs; i++)
 327         {
 328                 ObjectAddress *thisobj = targetObjects->refs + i;
 329
 330                 deleteOneObject(thisobj, depRel);
 331         }
 332
 333         /* And clean up */
 334         free_object_addresses(targetObjects);
 335
 336         heap_close(depRel, RowExclusiveLock);
 337 }
 338
 339 /*
 340  * deleteWhatDependsOn: attempt to drop everything that depends on the
 341  * specified object, though not the object itself.      Behavior is always
 342  * CASCADE.
 343  *
 344  * This is currently used only to clean out the contents of a schema
 345  * (namespace): the passed object is a namespace.  We normally want this
 346  * to be done silently, so there's an option to suppress NOTICE messages.
 347  */
 348 void
 349 deleteWhatDependsOn(const ObjectAddress *object,
 350                                         bool showNotices)
 351 {
 352         Relation        depRel;
 353         ObjectAddresses *targetObjects;
 354         int                     i;
 355
 356         /*
 357          * We save some cycles by opening pg_depend just once and passing the
 358          * Relation pointer down to all the recursive deletion steps.
 359          */
 360         depRel = heap_open(DependRelationId, RowExclusiveLock);
 361
 362         /*
 363          * Acquire deletion lock on the target object.  (Ideally the caller has
 364          * done this already, but many places are sloppy about it.)
 365          */
 366         AcquireDeletionLock(object);
 367
 368         /*
 369          * Construct a list of objects to delete (ie, the given object plus
 370          * everything directly or indirectly dependent on it).
 371          */
 372         targetObjects = new_object_addresses();
 373
 374         findDependentObjects(object,
 375                                                  DEPFLAG_ORIGINAL,
 376                                                  NULL,  /* empty stack */
 377                                                  targetObjects,
 378                                                  NULL,  /* no pendingObjects */
 379                                                  depRel);
 380
 381         /*
 382          * Check if deletion is allowed, and report about cascaded deletes.
 383          */
 384         reportDependentObjects(targetObjects,
 385                                                    DROP_CASCADE,
 386                                                    showNotices ? NOTICE : DEBUG2,
 387                                                    object);
 388
 389         /*
 390          * Delete all the objects in the proper order, except we skip the original
 391          * object.
 392          */
 393         for (i = 0; i < targetObjects->numrefs; i++)
 394         {
 395                 ObjectAddress *thisobj = targetObjects->refs + i;
 396                 ObjectAddressExtra *thisextra = targetObjects->extras + i;
 397
 398                 if (thisextra->flags & DEPFLAG_ORIGINAL)
 399                         continue;
 400
 401                 deleteOneObject(thisobj, depRel);
 402         }
 403
 404         /* And clean up */
 405         free_object_addresses(targetObjects);
 406
 407         heap_close(depRel, RowExclusiveLock);
 408 }
 409
 410 /*
 411  * findDependentObjects - find all objects that depend on 'object'
 412  *
 413  * For every object that depends on the starting object, acquire a deletion
 414  * lock on the object, add it to targetObjects (if not already there),
 415  * and recursively find objects that depend on it.      An object's dependencies
 416  * will be placed into targetObjects before the object itself; this means
 417  * that the finished list's order represents a safe deletion order.
 418  *
 419  * The caller must already have a deletion lock on 'object' itself,
 420  * but must not have added it to targetObjects.  (Note: there are corner
 421  * cases where we won't add the object either, and will also release the
 422  * caller-taken lock.  This is a bit ugly, but the API is set up this way
 423  * to allow easy rechecking of an object's liveness after we lock it.  See
 424  * notes within the function.)
 425  *
 426  * When dropping a whole object (subId = 0), we find dependencies for
 427  * its sub-objects too.
 428  *
 429  *      object: the object to add to targetObjects and find dependencies on
 430  *      flags: flags to be ORed into the object's targetObjects entry
 431  *      stack: list of objects being visited in current recursion; topmost item
 432  *                      is the object that we recursed from (NULL for external callers)
 433  *      targetObjects: list of objects that are scheduled to be deleted
 434  *      pendingObjects: list of other objects slated for destruction, but
 435  *                      not necessarily in targetObjects yet (can be NULL if none)
 436  *      depRel: already opened pg_depend relation
 437  */
 438 static void
 439 findDependentObjects(const ObjectAddress *object,
 440                                          int flags,
 441                                          ObjectAddressStack *stack,
 442                                          ObjectAddresses *targetObjects,
 443                                          const ObjectAddresses *pendingObjects,
 444                                          Relation depRel)
 445 {
 446         ScanKeyData key[3];
 447         int                     nkeys;
 448         SysScanDesc scan;
 449         HeapTuple       tup;
 450         ObjectAddress otherObject;
 451         ObjectAddressStack mystack;
 452         ObjectAddressExtra extra;
 453         ObjectAddressStack *stackptr;
 454
 455         /*
 456          * If the target object is already being visited in an outer recursion
 457          * level, just report the current flags back to that level and exit. This
 458          * is needed to avoid infinite recursion in the face of circular
 459          * dependencies.
 460          *
 461          * The stack check alone would result in dependency loops being broken at
 462          * an arbitrary point, ie, the first member object of the loop to be
 463          * visited is the last one to be deleted.  This is obviously unworkable.
 464          * However, the check for internal dependency below guarantees that we
 465          * will not break a loop at an internal dependency: if we enter the loop
 466          * at an "owned" object we will switch and start at the "owning" object
 467          * instead.  We could probably hack something up to avoid breaking at an
 468          * auto dependency, too, if we had to.  However there are no known cases
 469          * where that would be necessary.
 470          */
 471         for (stackptr = stack; stackptr; stackptr = stackptr->next)
 472         {
 473                 if (object->classId == stackptr->object->classId &&
 474                         object->objectId == stackptr->object->objectId)
 475                 {
 476                         if (object->objectSubId == stackptr->object->objectSubId)
 477                         {
 478                                 stackptr->flags |= flags;
 479                                 return;
 480                         }
 481
 482                         /*
 483                          * Could visit column with whole table already on stack; this is
 484                          * the same case noted in object_address_present_add_flags().
 485                          * (It's not clear this can really happen, but we might as well
 486                          * check.)
 487                          */
 488                         if (stackptr->object->objectSubId == 0)
 489                                 return;
 490                 }
 491         }
 492
 493         /*
 494          * It's also possible that the target object has already been completely
 495          * processed and put into targetObjects.  If so, again we just add the
 496          * specified flags to its entry and return.
 497          *
 498          * (Note: in these early-exit cases we could release the caller-taken
 499          * lock, since the object is presumably now locked multiple times; but it
 500          * seems not worth the cycles.)
 501          */
 502         if (object_address_present_add_flags(object, flags, targetObjects))
 503                 return;
 504
 505         /*
 506          * The target object might be internally dependent on some other object
 507          * (its "owner").  If so, and if we aren't recursing from the owning
 508          * object, we have to transform this deletion request into a deletion
 509          * request of the owning object.  (We'll eventually recurse back to this
 510          * object, but the owning object has to be visited first so it will be
 511          * deleted after.)      The way to find out about this is to scan the
 512          * pg_depend entries that show what this object depends on.
 513          */
 514         ScanKeyInit(&key[0],
 515                                 Anum_pg_depend_classid,
 516                                 BTEqualStrategyNumber, F_OIDEQ,
 517                                 ObjectIdGetDatum(object->classId));
 518         ScanKeyInit(&key[1],
 519                                 Anum_pg_depend_objid,
 520                                 BTEqualStrategyNumber, F_OIDEQ,
 521                                 ObjectIdGetDatum(object->objectId));
 522         if (object->objectSubId != 0)
 523         {
 524                 ScanKeyInit(&key[2],
 525                                         Anum_pg_depend_objsubid,
 526                                         BTEqualStrategyNumber, F_INT4EQ,
 527                                         Int32GetDatum(object->objectSubId));
 528                 nkeys = 3;
 529         }
 530         else
 531                 nkeys = 2;
 532
 533         scan = systable_beginscan(depRel, DependDependerIndexId, true,
 534                                                           SnapshotNow, nkeys, key);
 535
 536         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 537         {
 538                 Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup);
 539
 540                 otherObject.classId = foundDep->refclassid;
 541                 otherObject.objectId = foundDep->refobjid;
 542                 otherObject.objectSubId = foundDep->refobjsubid;
 543
 544                 switch (foundDep->deptype)
 545                 {
 546                         case DEPENDENCY_NORMAL:
 547                         case DEPENDENCY_AUTO:
 548                                 /* no problem */
 549                                 break;
 550                         case DEPENDENCY_INTERNAL:
 551
 552                                 /*
 553                                  * This object is part of the internal implementation of
 554                                  * another object.      We have three cases:
 555                                  *
 556                                  * 1. At the outermost recursion level, disallow the DROP. (We
 557                                  * just ereport here, rather than proceeding, since no other
 558                                  * dependencies are likely to be interesting.)  However, if
 559                                  * the other object is listed in pendingObjects, just release
 560                                  * the caller's lock and return; we'll eventually complete the
 561                                  * DROP when we reach that entry in the pending list.
 562                                  */
 563                                 if (stack == NULL)
 564                                 {
 565                                         char       *otherObjDesc;
 566
 567                                         if (pendingObjects &&
 568                                                 object_address_present(&otherObject, pendingObjects))
 569                                         {
 570                                                 systable_endscan(scan);
 571                                                 /* need to release caller's lock; see notes below */
 572                                                 ReleaseDeletionLock(object);
 573                                                 return;
 574                                         }
 575                                         otherObjDesc = getObjectDescription(&otherObject);
 576                                         ereport(ERROR,
 577                                                         (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 578                                                          errmsg("cannot drop %s because %s requires it",
 579                                                                         getObjectDescription(object),
 580                                                                         otherObjDesc),
 581                                                          errhint("You can drop %s instead.",
 582                                                                          otherObjDesc)));
 583                                 }
 584
 585                                 /*
 586                                  * 2. When recursing from the other end of this dependency,
 587                                  * it's okay to continue with the deletion. This holds when
 588                                  * recursing from a whole object that includes the nominal
 589                                  * other end as a component, too.
 590                                  */
 591                                 if (stack->object->classId == otherObject.classId &&
 592                                         stack->object->objectId == otherObject.objectId &&
 593                                         (stack->object->objectSubId == otherObject.objectSubId ||
 594                                          stack->object->objectSubId == 0))
 595                                         break;
 596
 597                                 /*
 598                                  * 3. When recursing from anyplace else, transform this
 599                                  * deletion request into a delete of the other object.
 600                                  *
 601                                  * First, release caller's lock on this object and get
 602                                  * deletion lock on the other object.  (We must release
 603                                  * caller's lock to avoid deadlock against a concurrent
 604                                  * deletion of the other object.)
 605                                  */
 606                                 ReleaseDeletionLock(object);
 607                                 AcquireDeletionLock(&otherObject);
 608
 609                                 /*
 610                                  * The other object might have been deleted while we waited to
 611                                  * lock it; if so, neither it nor the current object are
 612                                  * interesting anymore.  We test this by checking the
 613                                  * pg_depend entry (see notes below).
 614                                  */
 615                                 if (!systable_recheck_tuple(scan, tup))
 616                                 {
 617                                         systable_endscan(scan);
 618                                         ReleaseDeletionLock(&otherObject);
 619                                         return;
 620                                 }
 621
 622                                 /*
 623                                  * Okay, recurse to the other object instead of proceeding. We
 624                                  * treat this exactly as if the original reference had linked
 625                                  * to that object instead of this one; hence, pass through the
 626                                  * same flags and stack.
 627                                  */
 628                                 findDependentObjects(&otherObject,
 629                                                                          flags,
 630                                                                          stack,
 631                                                                          targetObjects,
 632                                                                          pendingObjects,
 633                                                                          depRel);
 634                                 /* And we're done here. */
 635                                 systable_endscan(scan);
 636                                 return;
 637                         case DEPENDENCY_PIN:
 638
 639                                 /*
 640                                  * Should not happen; PIN dependencies should have zeroes in
 641                                  * the depender fields...
 642                                  */
 643                                 elog(ERROR, "incorrect use of PIN dependency with %s",
 644                                          getObjectDescription(object));
 645                                 break;
 646                         default:
 647                                 elog(ERROR, "unrecognized dependency type '%c' for %s",
 648                                          foundDep->deptype, getObjectDescription(object));
 649                                 break;
 650                 }
 651         }
 652
 653         systable_endscan(scan);
 654
 655         /*
 656          * Now recurse to any dependent objects.  We must visit them first since
 657          * they have to be deleted before the current object.
 658          */
 659         mystack.object = object;        /* set up a new stack level */
 660         mystack.flags = flags;
 661         mystack.next = stack;
 662
 663         ScanKeyInit(&key[0],
 664                                 Anum_pg_depend_refclassid,
 665                                 BTEqualStrategyNumber, F_OIDEQ,
 666                                 ObjectIdGetDatum(object->classId));
 667         ScanKeyInit(&key[1],
 668                                 Anum_pg_depend_refobjid,
 669                                 BTEqualStrategyNumber, F_OIDEQ,
 670                                 ObjectIdGetDatum(object->objectId));
 671         if (object->objectSubId != 0)
 672         {
 673                 ScanKeyInit(&key[2],
 674                                         Anum_pg_depend_refobjsubid,
 675                                         BTEqualStrategyNumber, F_INT4EQ,
 676                                         Int32GetDatum(object->objectSubId));
 677                 nkeys = 3;
 678         }
 679         else
 680                 nkeys = 2;
 681
 682         scan = systable_beginscan(depRel, DependReferenceIndexId, true,
 683                                                           SnapshotNow, nkeys, key);
 684
 685         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 686         {
 687                 Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup);
 688                 int                     subflags;
 689
 690                 otherObject.classId = foundDep->classid;
 691                 otherObject.objectId = foundDep->objid;
 692                 otherObject.objectSubId = foundDep->objsubid;
 693
 694                 /*
 695                  * Must lock the dependent object before recursing to it.
 696                  */
 697                 AcquireDeletionLock(&otherObject);
 698
 699                 /*
 700                  * The dependent object might have been deleted while we waited to
 701                  * lock it; if so, we don't need to do anything more with it. We can
 702                  * test this cheaply and independently of the object's type by seeing
 703                  * if the pg_depend tuple we are looking at is still live. (If the
 704                  * object got deleted, the tuple would have been deleted too.)
 705                  */
 706                 if (!systable_recheck_tuple(scan, tup))
 707                 {
 708                         /* release the now-useless lock */
 709                         ReleaseDeletionLock(&otherObject);
 710                         /* and continue scanning for dependencies */
 711                         continue;
 712                 }
 713
 714                 /* Recurse, passing flags indicating the dependency type */
 715                 switch (foundDep->deptype)
 716                 {
 717                         case DEPENDENCY_NORMAL:
 718                                 subflags = DEPFLAG_NORMAL;
 719                                 break;
 720                         case DEPENDENCY_AUTO:
 721                                 subflags = DEPFLAG_AUTO;
 722                                 break;
 723                         case DEPENDENCY_INTERNAL:
 724                                 subflags = DEPFLAG_INTERNAL;
 725                                 break;
 726                         case DEPENDENCY_PIN:
 727
 728                                 /*
 729                                  * For a PIN dependency we just ereport immediately; there
 730                                  * won't be any others to report.
 731                                  */
 732                                 ereport(ERROR,
 733                                                 (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 734                                                  errmsg("cannot drop %s because it is required by the database system",
 735                                                                 getObjectDescription(object))));
 736                                 subflags = 0;   /* keep compiler quiet */
 737                                 break;
 738                         default:
 739                                 elog(ERROR, "unrecognized dependency type '%c' for %s",
 740                                          foundDep->deptype, getObjectDescription(object));
 741                                 subflags = 0;   /* keep compiler quiet */
 742                                 break;
 743                 }
 744
 745                 findDependentObjects(&otherObject,
 746                                                          subflags,
 747                                                          &mystack,
 748                                                          targetObjects,
 749                                                          pendingObjects,
 750                                                          depRel);
 751         }
 752
 753         systable_endscan(scan);
 754
 755         /*
 756          * Finally, we can add the target object to targetObjects.      Be careful to
 757          * include any flags that were passed back down to us from inner recursion
 758          * levels.
 759          */
 760         extra.flags = mystack.flags;
 761         if (stack)
 762                 extra.dependee = *stack->object;
 763         else
 764                 memset(&extra.dependee, 0, sizeof(extra.dependee));
 765         add_exact_object_address_extra(object, &extra, targetObjects);
 766 }
 767
 768 /*
 769  * reportDependentObjects - report about dependencies, and fail if RESTRICT
 770  *
 771  * Tell the user about dependent objects that we are going to delete
 772  * (or would need to delete, but are prevented by RESTRICT mode);
 773  * then error out if there are any and it's not CASCADE mode.
 774  *
 775  *      targetObjects: list of objects that are scheduled to be deleted
 776  *      behavior: RESTRICT or CASCADE
 777  *      msglevel: elog level for non-error report messages
 778  *      origObject: base object of deletion, or NULL if not available
 779  *              (the latter case occurs in DROP OWNED)
 780  */
 781 static void
 782 reportDependentObjects(const ObjectAddresses *targetObjects,
 783                                            DropBehavior behavior,
 784                                            int msglevel,
 785                                            const ObjectAddress *origObject)
 786 {
 787         bool            ok = true;
 788         StringInfoData clientdetail;
 789         StringInfoData logdetail;
 790         int                     numReportedClient = 0;
 791         int                     numNotReportedClient = 0;
 792         int                     i;
 793
 794         /*
 795          * If no error is to be thrown, and the msglevel is too low to be shown to
 796          * either client or server log, there's no need to do any of the work.
 797          *
 798          * Note: this code doesn't know all there is to be known about elog
 799          * levels, but it works for NOTICE and DEBUG2, which are the only values
 800          * msglevel can currently have.  We also assume we are running in a normal
 801          * operating environment.
 802          */
 803         if (behavior == DROP_CASCADE &&
 804                 msglevel < client_min_messages &&
 805                 (msglevel < log_min_messages || log_min_messages == LOG))
 806                 return;
 807
 808         /*
 809          * We limit the number of dependencies reported to the client to
 810          * MAX_REPORTED_DEPS, since client software may not deal well with
 811          * enormous error strings.      The server log always gets a full report.
 812          */
 813 #define MAX_REPORTED_DEPS 100
 814
 815         initStringInfo(&clientdetail);
 816         initStringInfo(&logdetail);
 817
 818         /*
 819          * We process the list back to front (ie, in dependency order not deletion
 820          * order), since this makes for a more understandable display.
 821          */
 822         for (i = targetObjects->numrefs - 1; i >= 0; i--)
 823         {
 824                 const ObjectAddress *obj = &targetObjects->refs[i];
 825                 const ObjectAddressExtra *extra = &targetObjects->extras[i];
 826                 char       *objDesc;
 827
 828                 /* Ignore the original deletion target(s) */
 829                 if (extra->flags & DEPFLAG_ORIGINAL)
 830                         continue;
 831
 832                 objDesc = getObjectDescription(obj);
 833
 834                 /*
 835                  * If, at any stage of the recursive search, we reached the object via
 836                  * an AUTO or INTERNAL dependency, then it's okay to delete it even in
 837                  * RESTRICT mode.
 838                  */
 839                 if (extra->flags & (DEPFLAG_AUTO | DEPFLAG_INTERNAL))
 840                 {
 841                         /*
 842                          * auto-cascades are reported at DEBUG2, not msglevel.  We don't
 843                          * try to combine them with the regular message because the
 844                          * results are too confusing when client_min_messages and
 845                          * log_min_messages are different.
 846                          */
 847                         ereport(DEBUG2,
 848                                         (errmsg("drop auto-cascades to %s",
 849                                                         objDesc)));
 850                 }
 851                 else if (behavior == DROP_RESTRICT)
 852                 {
 853                         char       *otherDesc = getObjectDescription(&extra->dependee);
 854
 855                         if (numReportedClient < MAX_REPORTED_DEPS)
 856                         {
 857                                 /* separate entries with a newline */
 858                                 if (clientdetail.len != 0)
 859                                         appendStringInfoChar(&clientdetail, '\n');
 860                                 appendStringInfo(&clientdetail, _("%s depends on %s"),
 861                                                                  objDesc, otherDesc);
 862                                 numReportedClient++;
 863                         }
 864                         else
 865                                 numNotReportedClient++;
 866                         /* separate entries with a newline */
 867                         if (logdetail.len != 0)
 868                                 appendStringInfoChar(&logdetail, '\n');
 869                         appendStringInfo(&logdetail, _("%s depends on %s"),
 870                                                          objDesc, otherDesc);
 871                         pfree(otherDesc);
 872                         ok = false;
 873                 }
 874                 else
 875                 {
 876                         if (numReportedClient < MAX_REPORTED_DEPS)
 877                         {
 878                                 /* separate entries with a newline */
 879                                 if (clientdetail.len != 0)
 880                                         appendStringInfoChar(&clientdetail, '\n');
 881                                 appendStringInfo(&clientdetail, _("drop cascades to %s"),
 882                                                                  objDesc);
 883                                 numReportedClient++;
 884                         }
 885                         else
 886                                 numNotReportedClient++;
 887                         /* separate entries with a newline */
 888                         if (logdetail.len != 0)
 889                                 appendStringInfoChar(&logdetail, '\n');
 890                         appendStringInfo(&logdetail, _("drop cascades to %s"),
 891                                                          objDesc);
 892                 }
 893
 894                 pfree(objDesc);
 895         }
 896
 897         if (numNotReportedClient > 0)
 898                 appendStringInfo(&clientdetail, ngettext("\nand %d other object "
 899                                                                                                  "(see server log for list)",
 900                                                                                                  "\nand %d other objects "
 901                                                                                                  "(see server log for list)",
 902                                                                                                  numNotReportedClient),
 903                                                  numNotReportedClient);
 904
 905         if (!ok)
 906         {
 907                 if (origObject)
 908                         ereport(ERROR,
 909                                         (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 910                                   errmsg("cannot drop %s because other objects depend on it",
 911                                                  getObjectDescription(origObject)),
 912                                          errdetail("%s", clientdetail.data),
 913                                          errdetail_log("%s", logdetail.data),
 914                                          errhint("Use DROP ... CASCADE to drop the dependent objects too.")));
 915                 else
 916                         ereport(ERROR,
 917                                         (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 918                                          errmsg("cannot drop desired object(s) because other objects depend on them"),
 919                                          errdetail("%s", clientdetail.data),
 920                                          errdetail_log("%s", logdetail.data),
 921                                          errhint("Use DROP ... CASCADE to drop the dependent objects too.")));
 922         }
 923         else if (numReportedClient > 1)
 924         {
 925                 ereport(msglevel,
 926                 /* translator: %d always has a value larger than 1 */
 927                                 (errmsg_plural("drop cascades to %d other object",
 928                                                            "drop cascades to %d other objects",
 929                                                            numReportedClient + numNotReportedClient,
 930                                                            numReportedClient + numNotReportedClient),
 931                                  errdetail("%s", clientdetail.data),
 932                                  errdetail_log("%s", logdetail.data)));
 933         }
 934         else if (numReportedClient == 1)
 935         {
 936                 /* we just use the single item as-is */
 937                 ereport(msglevel,
 938                                 (errmsg_internal("%s", clientdetail.data)));
 939         }
 940
 941         pfree(clientdetail.data);
 942         pfree(logdetail.data);
 943 }
 944
 945 /*
 946  * deleteOneObject: delete a single object for performDeletion.
 947  *
 948  * depRel is the already-open pg_depend relation.
 949  */
 950 static void
 951 deleteOneObject(const ObjectAddress *object, Relation depRel)
 952 {
 953         ScanKeyData key[3];
 954         int                     nkeys;
 955         SysScanDesc scan;
 956         HeapTuple       tup;
 957
 958         /*
 959          * First remove any pg_depend records that link from this object to
 960          * others.      (Any records linking to this object should be gone already.)
 961          *
 962          * When dropping a whole object (subId = 0), remove all pg_depend records
 963          * for its sub-objects too.
 964          */
 965         ScanKeyInit(&key[0],
 966                                 Anum_pg_depend_classid,
 967                                 BTEqualStrategyNumber, F_OIDEQ,
 968                                 ObjectIdGetDatum(object->classId));
 969         ScanKeyInit(&key[1],
 970                                 Anum_pg_depend_objid,
 971                                 BTEqualStrategyNumber, F_OIDEQ,
 972                                 ObjectIdGetDatum(object->objectId));
 973         if (object->objectSubId != 0)
 974         {
 975                 ScanKeyInit(&key[2],
 976                                         Anum_pg_depend_objsubid,
 977                                         BTEqualStrategyNumber, F_INT4EQ,
 978                                         Int32GetDatum(object->objectSubId));
 979                 nkeys = 3;
 980         }
 981         else
 982                 nkeys = 2;
 983
 984         scan = systable_beginscan(depRel, DependDependerIndexId, true,
 985                                                           SnapshotNow, nkeys, key);
 986
 987         while (HeapTupleIsValid(tup = systable_getnext(scan)))
 988         {
 989                 simple_heap_delete(depRel, &tup->t_self);
 990         }
 991
 992         systable_endscan(scan);
 993
 994         /*
 995          * Delete shared dependency references related to this object.  Again, if
 996          * subId = 0, remove records for sub-objects too.
 997          */
 998         deleteSharedDependencyRecordsFor(object->classId, object->objectId,
 999                                                                          object->objectSubId);
1000
1001         /*
1002          * Now delete the object itself, in an object-type-dependent way.
1003          */
1004         doDeletion(object);
1005
1006         /*
1007          * Delete any comments associated with this object.  (This is a convenient
1008          * place to do it instead of having every object type know to do it.)
1009          */
1010         DeleteComments(object->objectId, object->classId, object->objectSubId);
1011
1012         /*
1013          * CommandCounterIncrement here to ensure that preceding changes are all
1014          * visible to the next deletion step.
1015          */
1016         CommandCounterIncrement();
1017
1018         /*
1019          * And we're done!
1020          */
1021 }
1022
1023 /*
1024  * doDeletion: actually delete a single object
1025  */
1026 static void
1027 doDeletion(const ObjectAddress *object)
1028 {
1029         switch (getObjectClass(object))
1030         {
1031                 case OCLASS_CLASS:
1032                         {
1033                                 char            relKind = get_rel_relkind(object->objectId);
1034
1035                                 if (relKind == RELKIND_INDEX)
1036                                 {
1037                                         Assert(object->objectSubId == 0);
1038                                         index_drop(object->objectId);
1039                                 }
1040                                 else
1041                                 {
1042                                         if (object->objectSubId != 0)
1043                                                 RemoveAttributeById(object->objectId,
1044                                                                                         object->objectSubId);
1045                                         else
1046                                                 heap_drop_with_catalog(object->objectId);
1047                                 }
1048                                 break;
1049                         }
1050
1051                 case OCLASS_PROC:
1052                         RemoveFunctionById(object->objectId);
1053                         break;
1054
1055                 case OCLASS_TYPE:
1056                         RemoveTypeById(object->objectId);
1057                         break;
1058
1059                 case OCLASS_CAST:
1060                         DropCastById(object->objectId);
1061                         break;
1062
1063                 case OCLASS_CONSTRAINT:
1064                         RemoveConstraintById(object->objectId);
1065                         break;
1066
1067                 case OCLASS_CONVERSION:
1068                         RemoveConversionById(object->objectId);
1069                         break;
1070
1071                 case OCLASS_DEFAULT:
1072                         RemoveAttrDefaultById(object->objectId);
1073                         break;
1074
1075                 case OCLASS_LANGUAGE:
1076                         DropProceduralLanguageById(object->objectId);
1077                         break;
1078
1079                 case OCLASS_LARGEOBJECT:
1080                         LargeObjectDrop(object->objectId);
1081                         break;
1082
1083                 case OCLASS_OPERATOR:
1084                         RemoveOperatorById(object->objectId);
1085                         break;
1086
1087                 case OCLASS_OPCLASS:
1088                         RemoveOpClassById(object->objectId);
1089                         break;
1090
1091                 case OCLASS_OPFAMILY:
1092                         RemoveOpFamilyById(object->objectId);
1093                         break;
1094
1095                 case OCLASS_AMOP:
1096                         RemoveAmOpEntryById(object->objectId);
1097                         break;
1098
1099                 case OCLASS_AMPROC:
1100                         RemoveAmProcEntryById(object->objectId);
1101                         break;
1102
1103                 case OCLASS_REWRITE:
1104                         RemoveRewriteRuleById(object->objectId);
1105                         break;
1106
1107                 case OCLASS_TRIGGER:
1108                         RemoveTriggerById(object->objectId);
1109                         break;
1110
1111                 case OCLASS_SCHEMA:
1112                         RemoveSchemaById(object->objectId);
1113                         break;
1114
1115                 case OCLASS_TSPARSER:
1116                         RemoveTSParserById(object->objectId);
1117                         break;
1118
1119                 case OCLASS_TSDICT:
1120                         RemoveTSDictionaryById(object->objectId);
1121                         break;
1122
1123                 case OCLASS_TSTEMPLATE:
1124                         RemoveTSTemplateById(object->objectId);
1125                         break;
1126
1127                 case OCLASS_TSCONFIG:
1128                         RemoveTSConfigurationById(object->objectId);
1129                         break;
1130
1131                         /*
1132                          * OCLASS_ROLE, OCLASS_DATABASE, OCLASS_TBLSPACE intentionally not
1133                          * handled here
1134                          */
1135
1136                 case OCLASS_FDW:
1137                         RemoveForeignDataWrapperById(object->objectId);
1138                         break;
1139
1140                 case OCLASS_FOREIGN_SERVER:
1141                         RemoveForeignServerById(object->objectId);
1142                         break;
1143
1144                 case OCLASS_USER_MAPPING:
1145                         RemoveUserMappingById(object->objectId);
1146                         break;
1147
1148                 case OCLASS_DEFACL:
1149                         RemoveDefaultACLById(object->objectId);
1150                         break;
1151
1152                 default:
1153                         elog(ERROR, "unrecognized object class: %u",
1154                                  object->classId);
1155         }
1156 }
1157
1158 /*
1159  * AcquireDeletionLock - acquire a suitable lock for deleting an object
1160  *
1161  * We use LockRelation for relations, LockDatabaseObject for everything
1162  * else.  Note that dependency.c is not concerned with deleting any kind of
1163  * shared-across-databases object, so we have no need for LockSharedObject.
1164  */
1165 static void
1166 AcquireDeletionLock(const ObjectAddress *object)
1167 {
1168         if (object->classId == RelationRelationId)
1169                 LockRelationOid(object->objectId, AccessExclusiveLock);
1170         else
1171                 /* assume we should lock the whole object not a sub-object */
1172                 LockDatabaseObject(object->classId, object->objectId, 0,
1173                                                    AccessExclusiveLock);
1174 }
1175
1176 /*
1177  * ReleaseDeletionLock - release an object deletion lock
1178  */
1179 static void
1180 ReleaseDeletionLock(const ObjectAddress *object)
1181 {
1182         if (object->classId == RelationRelationId)
1183                 UnlockRelationOid(object->objectId, AccessExclusiveLock);
1184         else
1185                 /* assume we should lock the whole object not a sub-object */
1186                 UnlockDatabaseObject(object->classId, object->objectId, 0,
1187                                                          AccessExclusiveLock);
1188 }
1189
1190 /*
1191  * recordDependencyOnExpr - find expression dependencies
1192  *
1193  * This is used to find the dependencies of rules, constraint expressions,
1194  * etc.
1195  *
1196  * Given an expression or query in node-tree form, find all the objects
1197  * it refers to (tables, columns, operators, functions, etc).  Record
1198  * a dependency of the specified type from the given depender object
1199  * to each object mentioned in the expression.
1200  *
1201  * rtable is the rangetable to be used to interpret Vars with varlevelsup=0.
1202  * It can be NIL if no such variables are expected.
1203  */
1204 void
1205 recordDependencyOnExpr(const ObjectAddress *depender,
1206                                            Node *expr, List *rtable,
1207                                            DependencyType behavior)
1208 {
1209         find_expr_references_context context;
1210
1211         context.addrs = new_object_addresses();
1212
1213         /* Set up interpretation for Vars at varlevelsup = 0 */
1214         context.rtables = list_make1(rtable);
1215
1216         /* Scan the expression tree for referenceable objects */
1217         find_expr_references_walker(expr, &context);
1218
1219         /* Remove any duplicates */
1220         eliminate_duplicate_dependencies(context.addrs);
1221
1222         /* And record 'em */
1223         recordMultipleDependencies(depender,
1224                                                            context.addrs->refs, context.addrs->numrefs,
1225                                                            behavior);
1226
1227         free_object_addresses(context.addrs);
1228 }
1229
1230 /*
1231  * recordDependencyOnSingleRelExpr - find expression dependencies
1232  *
1233  * As above, but only one relation is expected to be referenced (with
1234  * varno = 1 and varlevelsup = 0).      Pass the relation OID instead of a
1235  * range table.  An additional frammish is that dependencies on that
1236  * relation (or its component columns) will be marked with 'self_behavior',
1237  * whereas 'behavior' is used for everything else.
1238  */
1239 void
1240 recordDependencyOnSingleRelExpr(const ObjectAddress *depender,
1241                                                                 Node *expr, Oid relId,
1242                                                                 DependencyType behavior,
1243                                                                 DependencyType self_behavior)
1244 {
1245         find_expr_references_context context;
1246         RangeTblEntry rte;
1247
1248         context.addrs = new_object_addresses();
1249
1250         /* We gin up a rather bogus rangetable list to handle Vars */
1251         MemSet(&rte, 0, sizeof(rte));
1252         rte.type = T_RangeTblEntry;
1253         rte.rtekind = RTE_RELATION;
1254         rte.relid = relId;
1255
1256         context.rtables = list_make1(list_make1(&rte));
1257
1258         /* Scan the expression tree for referenceable objects */
1259         find_expr_references_walker(expr, &context);
1260
1261         /* Remove any duplicates */
1262         eliminate_duplicate_dependencies(context.addrs);
1263
1264         /* Separate self-dependencies if necessary */
1265         if (behavior != self_behavior && context.addrs->numrefs > 0)
1266         {
1267                 ObjectAddresses *self_addrs;
1268                 ObjectAddress *outobj;
1269                 int                     oldref,
1270                                         outrefs;
1271
1272                 self_addrs = new_object_addresses();
1273
1274                 outobj = context.addrs->refs;
1275                 outrefs = 0;
1276                 for (oldref = 0; oldref < context.addrs->numrefs; oldref++)
1277                 {
1278                         ObjectAddress *thisobj = context.addrs->refs + oldref;
1279
1280                         if (thisobj->classId == RelationRelationId &&
1281                                 thisobj->objectId == relId)
1282                         {
1283                                 /* Move this ref into self_addrs */
1284                                 add_exact_object_address(thisobj, self_addrs);
1285                         }
1286                         else
1287                         {
1288                                 /* Keep it in context.addrs */
1289                                 *outobj = *thisobj;
1290                                 outobj++;
1291                                 outrefs++;
1292                         }
1293                 }
1294                 context.addrs->numrefs = outrefs;
1295
1296                 /* Record the self-dependencies */
1297                 recordMultipleDependencies(depender,
1298                                                                    self_addrs->refs, self_addrs->numrefs,
1299                                                                    self_behavior);
1300
1301                 free_object_addresses(self_addrs);
1302         }
1303
1304         /* Record the external dependencies */
1305         recordMultipleDependencies(depender,
1306                                                            context.addrs->refs, context.addrs->numrefs,
1307                                                            behavior);
1308
1309         free_object_addresses(context.addrs);
1310 }
1311
1312 /*
1313  * Recursively search an expression tree for object references.
1314  *
1315  * Note: we avoid creating references to columns of tables that participate
1316  * in an SQL JOIN construct, but are not actually used anywhere in the query.
1317  * To do so, we do not scan the joinaliasvars list of a join RTE while
1318  * scanning the query rangetable, but instead scan each individual entry
1319  * of the alias list when we find a reference to it.
1320  *
1321  * Note: in many cases we do not need to create dependencies on the datatypes
1322  * involved in an expression, because we'll have an indirect dependency via
1323  * some other object.  For instance Var nodes depend on a column which depends
1324  * on the datatype, and OpExpr nodes depend on the operator which depends on
1325  * the datatype.  However we do need a type dependency if there is no such
1326  * indirect dependency, as for example in Const and CoerceToDomain nodes.
1327  */
1328 static bool
1329 find_expr_references_walker(Node *node,
1330                                                         find_expr_references_context *context)
1331 {
1332         if (node == NULL)
1333                 return false;
1334         if (IsA(node, Var))
1335         {
1336                 Var                *var = (Var *) node;
1337                 List       *rtable;
1338                 RangeTblEntry *rte;
1339
1340                 /* Find matching rtable entry, or complain if not found */
1341                 if (var->varlevelsup >= list_length(context->rtables))
1342                         elog(ERROR, "invalid varlevelsup %d", var->varlevelsup);
1343                 rtable = (List *) list_nth(context->rtables, var->varlevelsup);
1344                 if (var->varno <= 0 || var->varno > list_length(rtable))
1345                         elog(ERROR, "invalid varno %d", var->varno);
1346                 rte = rt_fetch(var->varno, rtable);
1347
1348                 /*
1349                  * A whole-row Var references no specific columns, so adds no new
1350                  * dependency.
1351                  */
1352                 if (var->varattno == InvalidAttrNumber)
1353                         return false;
1354                 if (rte->rtekind == RTE_RELATION)
1355                 {
1356                         /* If it's a plain relation, reference this column */
1357                         add_object_address(OCLASS_CLASS, rte->relid, var->varattno,
1358                                                            context->addrs);
1359                 }
1360                 else if (rte->rtekind == RTE_JOIN)
1361                 {
1362                         /* Scan join output column to add references to join inputs */
1363                         List       *save_rtables;
1364
1365                         /* We must make the context appropriate for join's level */
1366                         save_rtables = context->rtables;
1367                         context->rtables = list_copy_tail(context->rtables,
1368                                                                                           var->varlevelsup);
1369                         if (var->varattno <= 0 ||
1370                                 var->varattno > list_length(rte->joinaliasvars))
1371                                 elog(ERROR, "invalid varattno %d", var->varattno);
1372                         find_expr_references_walker((Node *) list_nth(rte->joinaliasvars,
1373                                                                                                                   var->varattno - 1),
1374                                                                                 context);
1375                         list_free(context->rtables);
1376                         context->rtables = save_rtables;
1377                 }
1378                 return false;
1379         }
1380         else if (IsA(node, Const))
1381         {
1382                 Const      *con = (Const *) node;
1383                 Oid                     objoid;
1384
1385                 /* A constant must depend on the constant's datatype */
1386                 add_object_address(OCLASS_TYPE, con->consttype, 0,
1387                                                    context->addrs);
1388
1389                 /*
1390                  * If it's a regclass or similar literal referring to an existing
1391                  * object, add a reference to that object.      (Currently, only the
1392                  * regclass and regconfig cases have any likely use, but we may as
1393                  * well handle all the OID-alias datatypes consistently.)
1394                  */
1395                 if (!con->constisnull)
1396                 {
1397                         switch (con->consttype)
1398                         {
1399                                 case REGPROCOID:
1400                                 case REGPROCEDUREOID:
1401                                         objoid = DatumGetObjectId(con->constvalue);
1402                                         if (SearchSysCacheExists1(PROCOID,
1403                                                                                           ObjectIdGetDatum(objoid)))
1404                                                 add_object_address(OCLASS_PROC, objoid, 0,
1405                                                                                    context->addrs);
1406                                         break;
1407                                 case REGOPEROID:
1408                                 case REGOPERATOROID:
1409                                         objoid = DatumGetObjectId(con->constvalue);
1410                                         if (SearchSysCacheExists1(OPEROID,
1411                                                                                           ObjectIdGetDatum(objoid)))
1412                                                 add_object_address(OCLASS_OPERATOR, objoid, 0,
1413                                                                                    context->addrs);
1414                                         break;
1415                                 case REGCLASSOID:
1416                                         objoid = DatumGetObjectId(con->constvalue);
1417                                         if (SearchSysCacheExists1(RELOID,
1418                                                                                           ObjectIdGetDatum(objoid)))
1419                                                 add_object_address(OCLASS_CLASS, objoid, 0,
1420                                                                                    context->addrs);
1421                                         break;
1422                                 case REGTYPEOID:
1423                                         objoid = DatumGetObjectId(con->constvalue);
1424                                         if (SearchSysCacheExists1(TYPEOID,
1425                                                                                           ObjectIdGetDatum(objoid)))
1426                                                 add_object_address(OCLASS_TYPE, objoid, 0,
1427                                                                                    context->addrs);
1428                                         break;
1429                                 case REGCONFIGOID:
1430                                         objoid = DatumGetObjectId(con->constvalue);
1431                                         if (SearchSysCacheExists1(TSCONFIGOID,
1432                                                                                           ObjectIdGetDatum(objoid)))
1433                                                 add_object_address(OCLASS_TSCONFIG, objoid, 0,
1434                                                                                    context->addrs);
1435                                         break;
1436                                 case REGDICTIONARYOID:
1437                                         objoid = DatumGetObjectId(con->constvalue);
1438                                         if (SearchSysCacheExists1(TSDICTOID,
1439                                                                                           ObjectIdGetDatum(objoid)))
1440                                                 add_object_address(OCLASS_TSDICT, objoid, 0,
1441                                                                                    context->addrs);
1442                                         break;
1443                         }
1444                 }
1445                 return false;
1446         }
1447         else if (IsA(node, Param))
1448         {
1449                 Param      *param = (Param *) node;
1450
1451                 /* A parameter must depend on the parameter's datatype */
1452                 add_object_address(OCLASS_TYPE, param->paramtype, 0,
1453                                                    context->addrs);
1454         }
1455         else if (IsA(node, FuncExpr))
1456         {
1457                 FuncExpr   *funcexpr = (FuncExpr *) node;
1458
1459                 add_object_address(OCLASS_PROC, funcexpr->funcid, 0,
1460                                                    context->addrs);
1461                 /* fall through to examine arguments */
1462         }
1463         else if (IsA(node, OpExpr))
1464         {
1465                 OpExpr     *opexpr = (OpExpr *) node;
1466
1467                 add_object_address(OCLASS_OPERATOR, opexpr->opno, 0,
1468                                                    context->addrs);
1469                 /* fall through to examine arguments */
1470         }
1471         else if (IsA(node, DistinctExpr))
1472         {
1473                 DistinctExpr *distinctexpr = (DistinctExpr *) node;
1474
1475                 add_object_address(OCLASS_OPERATOR, distinctexpr->opno, 0,
1476                                                    context->addrs);
1477                 /* fall through to examine arguments */
1478         }
1479         else if (IsA(node, ScalarArrayOpExpr))
1480         {
1481                 ScalarArrayOpExpr *opexpr = (ScalarArrayOpExpr *) node;
1482
1483                 add_object_address(OCLASS_OPERATOR, opexpr->opno, 0,
1484                                                    context->addrs);
1485                 /* fall through to examine arguments */
1486         }
1487         else if (IsA(node, NullIfExpr))
1488         {
1489                 NullIfExpr *nullifexpr = (NullIfExpr *) node;
1490
1491                 add_object_address(OCLASS_OPERATOR, nullifexpr->opno, 0,
1492                                                    context->addrs);
1493                 /* fall through to examine arguments */
1494         }
1495         else if (IsA(node, Aggref))
1496         {
1497                 Aggref     *aggref = (Aggref *) node;
1498
1499                 add_object_address(OCLASS_PROC, aggref->aggfnoid, 0,
1500                                                    context->addrs);
1501                 /* fall through to examine arguments */
1502         }
1503         else if (IsA(node, WindowFunc))
1504         {
1505                 WindowFunc *wfunc = (WindowFunc *) node;
1506
1507                 add_object_address(OCLASS_PROC, wfunc->winfnoid, 0,
1508                                                    context->addrs);
1509                 /* fall through to examine arguments */
1510         }
1511         else if (IsA(node, SubPlan))
1512         {
1513                 /* Extra work needed here if we ever need this case */
1514                 elog(ERROR, "already-planned subqueries not supported");
1515         }
1516         else if (IsA(node, RelabelType))
1517         {
1518                 RelabelType *relab = (RelabelType *) node;
1519
1520                 /* since there is no function dependency, need to depend on type */
1521                 add_object_address(OCLASS_TYPE, relab->resulttype, 0,
1522                                                    context->addrs);
1523         }
1524         else if (IsA(node, CoerceViaIO))
1525         {
1526                 CoerceViaIO *iocoerce = (CoerceViaIO *) node;
1527
1528                 /* since there is no exposed function, need to depend on type */
1529                 add_object_address(OCLASS_TYPE, iocoerce->resulttype, 0,
1530                                                    context->addrs);
1531         }
1532         else if (IsA(node, ArrayCoerceExpr))
1533         {
1534                 ArrayCoerceExpr *acoerce = (ArrayCoerceExpr *) node;
1535
1536                 if (OidIsValid(acoerce->elemfuncid))
1537                         add_object_address(OCLASS_PROC, acoerce->elemfuncid, 0,
1538                                                            context->addrs);
1539                 add_object_address(OCLASS_TYPE, acoerce->resulttype, 0,
1540                                                    context->addrs);
1541                 /* fall through to examine arguments */
1542         }
1543         else if (IsA(node, ConvertRowtypeExpr))
1544         {
1545                 ConvertRowtypeExpr *cvt = (ConvertRowtypeExpr *) node;
1546
1547                 /* since there is no function dependency, need to depend on type */
1548                 add_object_address(OCLASS_TYPE, cvt->resulttype, 0,
1549                                                    context->addrs);
1550         }
1551         else if (IsA(node, RowExpr))
1552         {
1553                 RowExpr    *rowexpr = (RowExpr *) node;
1554
1555                 add_object_address(OCLASS_TYPE, rowexpr->row_typeid, 0,
1556                                                    context->addrs);
1557         }
1558         else if (IsA(node, RowCompareExpr))
1559         {
1560                 RowCompareExpr *rcexpr = (RowCompareExpr *) node;
1561                 ListCell   *l;
1562
1563                 foreach(l, rcexpr->opnos)
1564                 {
1565                         add_object_address(OCLASS_OPERATOR, lfirst_oid(l), 0,
1566                                                            context->addrs);
1567                 }
1568                 foreach(l, rcexpr->opfamilies)
1569                 {
1570                         add_object_address(OCLASS_OPFAMILY, lfirst_oid(l), 0,
1571                                                            context->addrs);
1572                 }
1573                 /* fall through to examine arguments */
1574         }
1575         else if (IsA(node, CoerceToDomain))
1576         {
1577                 CoerceToDomain *cd = (CoerceToDomain *) node;
1578
1579                 add_object_address(OCLASS_TYPE, cd->resulttype, 0,
1580                                                    context->addrs);
1581         }
1582         else if (IsA(node, SortGroupClause))
1583         {
1584                 SortGroupClause *sgc = (SortGroupClause *) node;
1585
1586                 add_object_address(OCLASS_OPERATOR, sgc->eqop, 0,
1587                                                    context->addrs);
1588                 if (OidIsValid(sgc->sortop))
1589                         add_object_address(OCLASS_OPERATOR, sgc->sortop, 0,
1590                                                            context->addrs);
1591                 return false;
1592         }
1593         else if (IsA(node, Query))
1594         {
1595                 /* Recurse into RTE subquery or not-yet-planned sublink subquery */
1596                 Query      *query = (Query *) node;
1597                 ListCell   *rtable;
1598                 bool            result;
1599
1600                 /*
1601                  * Add whole-relation refs for each plain relation mentioned in the
1602                  * subquery's rtable, as well as datatype refs for any datatypes used
1603                  * as a RECORD function's output.  (Note: query_tree_walker takes care
1604                  * of recursing into RTE_FUNCTION RTEs, subqueries, etc, so no need to
1605                  * do that here.  But keep it from looking at join alias lists.)
1606                  */
1607                 foreach(rtable, query->rtable)
1608                 {
1609                         RangeTblEntry *rte = (RangeTblEntry *) lfirst(rtable);
1610                         ListCell   *ct;
1611
1612                         switch (rte->rtekind)
1613                         {
1614                                 case RTE_RELATION:
1615                                         add_object_address(OCLASS_CLASS, rte->relid, 0,
1616                                                                            context->addrs);
1617                                         break;
1618                                 case RTE_FUNCTION:
1619                                         foreach(ct, rte->funccoltypes)
1620                                         {
1621                                                 add_object_address(OCLASS_TYPE, lfirst_oid(ct), 0,
1622                                                                                    context->addrs);
1623                                         }
1624                                         break;
1625                                 default:
1626                                         break;
1627                         }
1628                 }
1629
1630                 /* query_tree_walker ignores ORDER BY etc, but we need those opers */
1631                 find_expr_references_walker((Node *) query->sortClause, context);
1632                 find_expr_references_walker((Node *) query->groupClause, context);
1633                 find_expr_references_walker((Node *) query->windowClause, context);
1634                 find_expr_references_walker((Node *) query->distinctClause, context);
1635
1636                 /* Examine substructure of query */
1637                 context->rtables = lcons(query->rtable, context->rtables);
1638                 result = query_tree_walker(query,
1639                                                                    find_expr_references_walker,
1640                                                                    (void *) context,
1641                                                                    QTW_IGNORE_JOINALIASES);
1642                 context->rtables = list_delete_first(context->rtables);
1643                 return result;
1644         }
1645         else if (IsA(node, SetOperationStmt))
1646         {
1647                 SetOperationStmt *setop = (SetOperationStmt *) node;
1648
1649                 /* we need to look at the groupClauses for operator references */
1650                 find_expr_references_walker((Node *) setop->groupClauses, context);
1651                 /* fall through to examine child nodes */
1652         }
1653
1654         return expression_tree_walker(node, find_expr_references_walker,
1655                                                                   (void *) context);
1656 }
1657
1658 /*
1659  * Given an array of dependency references, eliminate any duplicates.
1660  */
1661 static void
1662 eliminate_duplicate_dependencies(ObjectAddresses *addrs)
1663 {
1664         ObjectAddress *priorobj;
1665         int                     oldref,
1666                                 newrefs;
1667
1668         /*
1669          * We can't sort if the array has "extra" data, because there's no way to
1670          * keep it in sync.  Fortunately that combination of features is not
1671          * needed.
1672          */
1673         Assert(!addrs->extras);
1674
1675         if (addrs->numrefs <= 1)
1676                 return;                                 /* nothing to do */
1677
1678         /* Sort the refs so that duplicates are adjacent */
1679         qsort((void *) addrs->refs, addrs->numrefs, sizeof(ObjectAddress),
1680                   object_address_comparator);
1681
1682         /* Remove dups */
1683         priorobj = addrs->refs;
1684         newrefs = 1;
1685         for (oldref = 1; oldref < addrs->numrefs; oldref++)
1686         {
1687                 ObjectAddress *thisobj = addrs->refs + oldref;
1688
1689                 if (priorobj->classId == thisobj->classId &&
1690                         priorobj->objectId == thisobj->objectId)
1691                 {
1692                         if (priorobj->objectSubId == thisobj->objectSubId)
1693                                 continue;               /* identical, so drop thisobj */
1694
1695                         /*
1696                          * If we have a whole-object reference and a reference to a part
1697                          * of the same object, we don't need the whole-object reference
1698                          * (for example, we don't need to reference both table foo and
1699                          * column foo.bar).  The whole-object reference will always appear
1700                          * first in the sorted list.
1701                          */
1702                         if (priorobj->objectSubId == 0)
1703                         {
1704                                 /* replace whole ref with partial */
1705                                 priorobj->objectSubId = thisobj->objectSubId;
1706                                 continue;
1707                         }
1708                 }
1709                 /* Not identical, so add thisobj to output set */
1710                 priorobj++;
1711                 *priorobj = *thisobj;
1712                 newrefs++;
1713         }
1714
1715         addrs->numrefs = newrefs;
1716 }
1717
1718 /*
1719  * qsort comparator for ObjectAddress items
1720  */
1721 static int
1722 object_address_comparator(const void *a, const void *b)
1723 {
1724         const ObjectAddress *obja = (const ObjectAddress *) a;
1725         const ObjectAddress *objb = (const ObjectAddress *) b;
1726
1727         if (obja->classId < objb->classId)
1728                 return -1;
1729         if (obja->classId > objb->classId)
1730                 return 1;
1731         if (obja->objectId < objb->objectId)
1732                 return -1;
1733         if (obja->objectId > objb->objectId)
1734                 return 1;
1735
1736         /*
1737          * We sort the subId as an unsigned int so that 0 will come first. See
1738          * logic in eliminate_duplicate_dependencies.
1739          */
1740         if ((unsigned int) obja->objectSubId < (unsigned int) objb->objectSubId)
1741                 return -1;
1742         if ((unsigned int) obja->objectSubId > (unsigned int) objb->objectSubId)
1743                 return 1;
1744         return 0;
1745 }
1746
1747 /*
1748  * Routines for handling an expansible array of ObjectAddress items.
1749  *
1750  * new_object_addresses: create a new ObjectAddresses array.
1751  */
1752 ObjectAddresses *
1753 new_object_addresses(void)
1754 {
1755         ObjectAddresses *addrs;
1756
1757         addrs = palloc(sizeof(ObjectAddresses));
1758
1759         addrs->numrefs = 0;
1760         addrs->maxrefs = 32;
1761         addrs->refs = (ObjectAddress *)
1762                 palloc(addrs->maxrefs * sizeof(ObjectAddress));
1763         addrs->extras = NULL;           /* until/unless needed */
1764
1765         return addrs;
1766 }
1767
1768 /*
1769  * Add an entry to an ObjectAddresses array.
1770  *
1771  * It is convenient to specify the class by ObjectClass rather than directly
1772  * by catalog OID.
1773  */
1774 static void
1775 add_object_address(ObjectClass oclass, Oid objectId, int32 subId,
1776                                    ObjectAddresses *addrs)
1777 {
1778         ObjectAddress *item;
1779
1780         /* enlarge array if needed */
1781         if (addrs->numrefs >= addrs->maxrefs)
1782         {
1783                 addrs->maxrefs *= 2;
1784                 addrs->refs = (ObjectAddress *)
1785                         repalloc(addrs->refs, addrs->maxrefs * sizeof(ObjectAddress));
1786                 Assert(!addrs->extras);
1787         }
1788         /* record this item */
1789         item = addrs->refs + addrs->numrefs;
1790         item->classId = object_classes[oclass];
1791         item->objectId = objectId;
1792         item->objectSubId = subId;
1793         addrs->numrefs++;
1794 }
1795
1796 /*
1797  * Add an entry to an ObjectAddresses array.
1798  *
1799  * As above, but specify entry exactly.
1800  */
1801 void
1802 add_exact_object_address(const ObjectAddress *object,
1803                                                  ObjectAddresses *addrs)
1804 {
1805         ObjectAddress *item;
1806
1807         /* enlarge array if needed */
1808         if (addrs->numrefs >= addrs->maxrefs)
1809         {
1810                 addrs->maxrefs *= 2;
1811                 addrs->refs = (ObjectAddress *)
1812                         repalloc(addrs->refs, addrs->maxrefs * sizeof(ObjectAddress));
1813                 Assert(!addrs->extras);
1814         }
1815         /* record this item */
1816         item = addrs->refs + addrs->numrefs;
1817         *item = *object;
1818         addrs->numrefs++;
1819 }
1820
1821 /*
1822  * Add an entry to an ObjectAddresses array.
1823  *
1824  * As above, but specify entry exactly and provide some "extra" data too.
1825  */
1826 static void
1827 add_exact_object_address_extra(const ObjectAddress *object,
1828                                                            const ObjectAddressExtra *extra,
1829                                                            ObjectAddresses *addrs)
1830 {
1831         ObjectAddress *item;
1832         ObjectAddressExtra *itemextra;
1833
1834         /* allocate extra space if first time */
1835         if (!addrs->extras)
1836                 addrs->extras = (ObjectAddressExtra *)
1837                         palloc(addrs->maxrefs * sizeof(ObjectAddressExtra));
1838
1839         /* enlarge array if needed */
1840         if (addrs->numrefs >= addrs->maxrefs)
1841         {
1842                 addrs->maxrefs *= 2;
1843                 addrs->refs = (ObjectAddress *)
1844                         repalloc(addrs->refs, addrs->maxrefs * sizeof(ObjectAddress));
1845                 addrs->extras = (ObjectAddressExtra *)
1846                         repalloc(addrs->extras, addrs->maxrefs * sizeof(ObjectAddressExtra));
1847         }
1848         /* record this item */
1849         item = addrs->refs + addrs->numrefs;
1850         *item = *object;
1851         itemextra = addrs->extras + addrs->numrefs;
1852         *itemextra = *extra;
1853         addrs->numrefs++;
1854 }
1855
1856 /*
1857  * Test whether an object is present in an ObjectAddresses array.
1858  *
1859  * We return "true" if object is a subobject of something in the array, too.
1860  */
1861 bool
1862 object_address_present(const ObjectAddress *object,
1863                                            const ObjectAddresses *addrs)
1864 {
1865         int                     i;
1866
1867         for (i = addrs->numrefs - 1; i >= 0; i--)
1868         {
1869                 const ObjectAddress *thisobj = addrs->refs + i;
1870
1871                 if (object->classId == thisobj->classId &&
1872                         object->objectId == thisobj->objectId)
1873                 {
1874                         if (object->objectSubId == thisobj->objectSubId ||
1875                                 thisobj->objectSubId == 0)
1876                                 return true;
1877                 }
1878         }
1879
1880         return false;
1881 }
1882
1883 /*
1884  * As above, except that if the object is present then also OR the given
1885  * flags into its associated extra data (which must exist).
1886  */
1887 static bool
1888 object_address_present_add_flags(const ObjectAddress *object,
1889                                                                  int flags,
1890                                                                  ObjectAddresses *addrs)
1891 {
1892         int                     i;
1893
1894         for (i = addrs->numrefs - 1; i >= 0; i--)
1895         {
1896                 ObjectAddress *thisobj = addrs->refs + i;
1897
1898                 if (object->classId == thisobj->classId &&
1899                         object->objectId == thisobj->objectId)
1900                 {
1901                         if (object->objectSubId == thisobj->objectSubId)
1902                         {
1903                                 ObjectAddressExtra *thisextra = addrs->extras + i;
1904
1905                                 thisextra->flags |= flags;
1906                                 return true;
1907                         }
1908                         if (thisobj->objectSubId == 0)
1909                         {
1910                                 /*
1911                                  * We get here if we find a need to delete a column after
1912                                  * having already decided to drop its whole table.      Obviously
1913                                  * we no longer need to drop the column.  But don't plaster
1914                                  * its flags on the table.
1915                                  */
1916                                 return true;
1917                         }
1918                 }
1919         }
1920
1921         return false;
1922 }
1923
1924 /*
1925  * Record multiple dependencies from an ObjectAddresses array, after first
1926  * removing any duplicates.
1927  */
1928 void
1929 record_object_address_dependencies(const ObjectAddress *depender,
1930                                                                    ObjectAddresses *referenced,
1931                                                                    DependencyType behavior)
1932 {
1933         eliminate_duplicate_dependencies(referenced);
1934         recordMultipleDependencies(depender,
1935                                                            referenced->refs, referenced->numrefs,
1936                                                            behavior);
1937 }
1938
1939 /*
1940  * Clean up when done with an ObjectAddresses array.
1941  */
1942 void
1943 free_object_addresses(ObjectAddresses *addrs)
1944 {
1945         pfree(addrs->refs);
1946         if (addrs->extras)
1947                 pfree(addrs->extras);
1948         pfree(addrs);
1949 }
1950
1951 /*
1952  * Determine the class of a given object identified by objectAddress.
1953  *
1954  * This function is essentially the reverse mapping for the object_classes[]
1955  * table.  We implement it as a function because the OIDs aren't consecutive.
1956  */
1957 ObjectClass
1958 getObjectClass(const ObjectAddress *object)
1959 {
1960         switch (object->classId)
1961         {
1962                 case RelationRelationId:
1963                         /* caller must check objectSubId */
1964                         return OCLASS_CLASS;
1965
1966                 case ProcedureRelationId:
1967                         Assert(object->objectSubId == 0);
1968                         return OCLASS_PROC;
1969
1970                 case TypeRelationId:
1971                         Assert(object->objectSubId == 0);
1972                         return OCLASS_TYPE;
1973
1974                 case CastRelationId:
1975                         Assert(object->objectSubId == 0);
1976                         return OCLASS_CAST;
1977
1978                 case ConstraintRelationId:
1979                         Assert(object->objectSubId == 0);
1980                         return OCLASS_CONSTRAINT;
1981
1982                 case ConversionRelationId:
1983                         Assert(object->objectSubId == 0);
1984                         return OCLASS_CONVERSION;
1985
1986                 case AttrDefaultRelationId:
1987                         Assert(object->objectSubId == 0);
1988                         return OCLASS_DEFAULT;
1989
1990                 case LanguageRelationId:
1991                         Assert(object->objectSubId == 0);
1992                         return OCLASS_LANGUAGE;
1993
1994                 case LargeObjectRelationId:
1995                         Assert(object->objectSubId == 0);
1996                         return OCLASS_LARGEOBJECT;
1997
1998                 case OperatorRelationId:
1999                         Assert(object->objectSubId == 0);
2000                         return OCLASS_OPERATOR;
2001
2002                 case OperatorClassRelationId:
2003                         Assert(object->objectSubId == 0);
2004                         return OCLASS_OPCLASS;
2005
2006                 case OperatorFamilyRelationId:
2007                         Assert(object->objectSubId == 0);
2008                         return OCLASS_OPFAMILY;
2009
2010                 case AccessMethodOperatorRelationId:
2011                         Assert(object->objectSubId == 0);
2012                         return OCLASS_AMOP;
2013
2014                 case AccessMethodProcedureRelationId:
2015                         Assert(object->objectSubId == 0);
2016                         return OCLASS_AMPROC;
2017
2018                 case RewriteRelationId:
2019                         Assert(object->objectSubId == 0);
2020                         return OCLASS_REWRITE;
2021
2022                 case TriggerRelationId:
2023                         Assert(object->objectSubId == 0);
2024                         return OCLASS_TRIGGER;
2025
2026                 case NamespaceRelationId:
2027                         Assert(object->objectSubId == 0);
2028                         return OCLASS_SCHEMA;
2029
2030                 case TSParserRelationId:
2031                         Assert(object->objectSubId == 0);
2032                         return OCLASS_TSPARSER;
2033
2034                 case TSDictionaryRelationId:
2035                         Assert(object->objectSubId == 0);
2036                         return OCLASS_TSDICT;
2037
2038                 case TSTemplateRelationId:
2039                         Assert(object->objectSubId == 0);
2040                         return OCLASS_TSTEMPLATE;
2041
2042                 case TSConfigRelationId:
2043                         Assert(object->objectSubId == 0);
2044                         return OCLASS_TSCONFIG;
2045
2046                 case AuthIdRelationId:
2047                         Assert(object->objectSubId == 0);
2048                         return OCLASS_ROLE;
2049
2050                 case DatabaseRelationId:
2051                         Assert(object->objectSubId == 0);
2052                         return OCLASS_DATABASE;
2053
2054                 case TableSpaceRelationId:
2055                         Assert(object->objectSubId == 0);
2056                         return OCLASS_TBLSPACE;
2057
2058                 case ForeignDataWrapperRelationId:
2059                         Assert(object->objectSubId == 0);
2060                         return OCLASS_FDW;
2061
2062                 case ForeignServerRelationId:
2063                         Assert(object->objectSubId == 0);
2064                         return OCLASS_FOREIGN_SERVER;
2065
2066                 case UserMappingRelationId:
2067                         Assert(object->objectSubId == 0);
2068                         return OCLASS_USER_MAPPING;
2069
2070                 case DefaultAclRelationId:
2071                         Assert(object->objectSubId == 0);
2072                         return OCLASS_DEFACL;
2073         }
2074
2075         /* shouldn't get here */
2076         elog(ERROR, "unrecognized object class: %u", object->classId);
2077         return OCLASS_CLASS;            /* keep compiler quiet */
2078 }
2079
2080 /*
2081  * getObjectDescription: build an object description for messages
2082  *
2083  * The result is a palloc'd string.
2084  */
2085 char *
2086 getObjectDescription(const ObjectAddress *object)
2087 {
2088         StringInfoData buffer;
2089
2090         initStringInfo(&buffer);
2091
2092         switch (getObjectClass(object))
2093         {
2094                 case OCLASS_CLASS:
2095                         getRelationDescription(&buffer, object->objectId);
2096                         if (object->objectSubId != 0)
2097                                 appendStringInfo(&buffer, _(" column %s"),
2098                                                                  get_relid_attribute_name(object->objectId,
2099                                                                                                            object->objectSubId));
2100                         break;
2101
2102                 case OCLASS_PROC:
2103                         appendStringInfo(&buffer, _("function %s"),
2104                                                          format_procedure(object->objectId));
2105                         break;
2106
2107                 case OCLASS_TYPE:
2108                         appendStringInfo(&buffer, _("type %s"),
2109                                                          format_type_be(object->objectId));
2110                         break;
2111
2112                 case OCLASS_CAST:
2113                         {
2114                                 Relation        castDesc;
2115                                 ScanKeyData skey[1];
2116                                 SysScanDesc rcscan;
2117                                 HeapTuple       tup;
2118                                 Form_pg_cast castForm;
2119
2120                                 castDesc = heap_open(CastRelationId, AccessShareLock);
2121
2122                                 ScanKeyInit(&skey[0],
2123                                                         ObjectIdAttributeNumber,
2124                                                         BTEqualStrategyNumber, F_OIDEQ,
2125                                                         ObjectIdGetDatum(object->objectId));
2126
2127                                 rcscan = systable_beginscan(castDesc, CastOidIndexId, true,
2128                                                                                         SnapshotNow, 1, skey);
2129
2130                                 tup = systable_getnext(rcscan);
2131
2132                                 if (!HeapTupleIsValid(tup))
2133                                         elog(ERROR, "could not find tuple for cast %u",
2134                                                  object->objectId);
2135
2136                                 castForm = (Form_pg_cast) GETSTRUCT(tup);
2137
2138                                 appendStringInfo(&buffer, _("cast from %s to %s"),
2139                                                                  format_type_be(castForm->castsource),
2140                                                                  format_type_be(castForm->casttarget));
2141
2142                                 systable_endscan(rcscan);
2143                                 heap_close(castDesc, AccessShareLock);
2144                                 break;
2145                         }
2146
2147                 case OCLASS_CONSTRAINT:
2148                         {
2149                                 HeapTuple       conTup;
2150                                 Form_pg_constraint con;
2151
2152                                 conTup = SearchSysCache1(CONSTROID,
2153                                                                                  ObjectIdGetDatum(object->objectId));
2154                                 if (!HeapTupleIsValid(conTup))
2155                                         elog(ERROR, "cache lookup failed for constraint %u",
2156                                                  object->objectId);
2157                                 con = (Form_pg_constraint) GETSTRUCT(conTup);
2158
2159                                 if (OidIsValid(con->conrelid))
2160                                 {
2161                                         StringInfoData rel;
2162
2163                                         initStringInfo(&rel);
2164                                         getRelationDescription(&rel, con->conrelid);
2165                                         appendStringInfo(&buffer, _("constraint %s on %s"),
2166                                                                          NameStr(con->conname), rel.data);
2167                                         pfree(rel.data);
2168                                 }
2169                                 else
2170                                 {
2171                                         appendStringInfo(&buffer, _("constraint %s"),
2172                                                                          NameStr(con->conname));
2173                                 }
2174
2175                                 ReleaseSysCache(conTup);
2176                                 break;
2177                         }
2178
2179                 case OCLASS_CONVERSION:
2180                         {
2181                                 HeapTuple       conTup;
2182
2183                                 conTup = SearchSysCache1(CONVOID,
2184                                                                                  ObjectIdGetDatum(object->objectId));
2185                                 if (!HeapTupleIsValid(conTup))
2186                                         elog(ERROR, "cache lookup failed for conversion %u",
2187                                                  object->objectId);
2188                                 appendStringInfo(&buffer, _("conversion %s"),
2189                                  NameStr(((Form_pg_conversion) GETSTRUCT(conTup))->conname));
2190                                 ReleaseSysCache(conTup);
2191                                 break;
2192                         }
2193
2194                 case OCLASS_DEFAULT:
2195                         {
2196                                 Relation        attrdefDesc;
2197                                 ScanKeyData skey[1];
2198                                 SysScanDesc adscan;
2199                                 HeapTuple       tup;
2200                                 Form_pg_attrdef attrdef;
2201                                 ObjectAddress colobject;
2202
2203                                 attrdefDesc = heap_open(AttrDefaultRelationId, AccessShareLock);
2204
2205                                 ScanKeyInit(&skey[0],
2206                                                         ObjectIdAttributeNumber,
2207                                                         BTEqualStrategyNumber, F_OIDEQ,
2208                                                         ObjectIdGetDatum(object->objectId));
2209
2210                                 adscan = systable_beginscan(attrdefDesc, AttrDefaultOidIndexId,
2211                                                                                         true, SnapshotNow, 1, skey);
2212
2213                                 tup = systable_getnext(adscan);
2214
2215                                 if (!HeapTupleIsValid(tup))
2216                                         elog(ERROR, "could not find tuple for attrdef %u",
2217                                                  object->objectId);
2218
2219                                 attrdef = (Form_pg_attrdef) GETSTRUCT(tup);
2220
2221                                 colobject.classId = RelationRelationId;
2222                                 colobject.objectId = attrdef->adrelid;
2223                                 colobject.objectSubId = attrdef->adnum;
2224
2225                                 appendStringInfo(&buffer, _("default for %s"),
2226                                                                  getObjectDescription(&colobject));
2227
2228                                 systable_endscan(adscan);
2229                                 heap_close(attrdefDesc, AccessShareLock);
2230                                 break;
2231                         }
2232
2233                 case OCLASS_LANGUAGE:
2234                         {
2235                                 HeapTuple       langTup;
2236
2237                                 langTup = SearchSysCache1(LANGOID,
2238                                                                                   ObjectIdGetDatum(object->objectId));
2239                                 if (!HeapTupleIsValid(langTup))
2240                                         elog(ERROR, "cache lookup failed for language %u",
2241                                                  object->objectId);
2242                                 appendStringInfo(&buffer, _("language %s"),
2243                                   NameStr(((Form_pg_language) GETSTRUCT(langTup))->lanname));
2244                                 ReleaseSysCache(langTup);
2245                                 break;
2246                         }
2247                 case OCLASS_LARGEOBJECT:
2248                         appendStringInfo(&buffer, _("large object %u"),
2249                                                          object->objectId);
2250                         break;
2251
2252                 case OCLASS_OPERATOR:
2253                         appendStringInfo(&buffer, _("operator %s"),
2254                                                          format_operator(object->objectId));
2255                         break;
2256
2257                 case OCLASS_OPCLASS:
2258                         {
2259                                 HeapTuple       opcTup;
2260                                 Form_pg_opclass opcForm;
2261                                 HeapTuple       amTup;
2262                                 Form_pg_am      amForm;
2263                                 char       *nspname;
2264
2265                                 opcTup = SearchSysCache1(CLAOID,
2266                                                                                  ObjectIdGetDatum(object->objectId));
2267                                 if (!HeapTupleIsValid(opcTup))
2268                                         elog(ERROR, "cache lookup failed for opclass %u",
2269                                                  object->objectId);
2270                                 opcForm = (Form_pg_opclass) GETSTRUCT(opcTup);
2271
2272                                 amTup = SearchSysCache1(AMOID,
2273                                                                                 ObjectIdGetDatum(opcForm->opcmethod));
2274                                 if (!HeapTupleIsValid(amTup))
2275                                         elog(ERROR, "cache lookup failed for access method %u",
2276                                                  opcForm->opcmethod);
2277                                 amForm = (Form_pg_am) GETSTRUCT(amTup);
2278
2279                                 /* Qualify the name if not visible in search path */
2280                                 if (OpclassIsVisible(object->objectId))
2281                                         nspname = NULL;
2282                                 else
2283                                         nspname = get_namespace_name(opcForm->opcnamespace);
2284
2285                                 appendStringInfo(&buffer, _("operator class %s for access method %s"),
2286                                                                  quote_qualified_identifier(nspname,
2287                                                                                                   NameStr(opcForm->opcname)),
2288                                                                  NameStr(amForm->amname));
2289
2290                                 ReleaseSysCache(amTup);
2291                                 ReleaseSysCache(opcTup);
2292                                 break;
2293                         }
2294
2295                 case OCLASS_OPFAMILY:
2296                         getOpFamilyDescription(&buffer, object->objectId);
2297                         break;
2298
2299                 case OCLASS_AMOP:
2300                         {
2301                                 Relation        amopDesc;
2302                                 ScanKeyData skey[1];
2303                                 SysScanDesc amscan;
2304                                 HeapTuple       tup;
2305                                 Form_pg_amop amopForm;
2306                                 StringInfoData opfam;
2307
2308                                 amopDesc = heap_open(AccessMethodOperatorRelationId,
2309                                                                          AccessShareLock);
2310
2311                                 ScanKeyInit(&skey[0],
2312                                                         ObjectIdAttributeNumber,
2313                                                         BTEqualStrategyNumber, F_OIDEQ,
2314                                                         ObjectIdGetDatum(object->objectId));
2315
2316                                 amscan = systable_beginscan(amopDesc, AccessMethodOperatorOidIndexId, true,
2317                                                                                         SnapshotNow, 1, skey);
2318
2319                                 tup = systable_getnext(amscan);
2320
2321                                 if (!HeapTupleIsValid(tup))
2322                                         elog(ERROR, "could not find tuple for amop entry %u",
2323                                                  object->objectId);
2324
2325                                 amopForm = (Form_pg_amop) GETSTRUCT(tup);
2326
2327                                 initStringInfo(&opfam);
2328                                 getOpFamilyDescription(&opfam, amopForm->amopfamily);
2329
2330                                 /*
2331                                  * translator: %d is the operator strategy (a number), the
2332                                  * first %s is the textual form of the operator, and the
2333                                  * second %s is the description of the operator family.
2334                                  */
2335                                 appendStringInfo(&buffer, _("operator %d %s of %s"),
2336                                                                  amopForm->amopstrategy,
2337                                                                  format_operator(amopForm->amopopr),
2338                                                                  opfam.data);
2339                                 pfree(opfam.data);
2340
2341                                 systable_endscan(amscan);
2342                                 heap_close(amopDesc, AccessShareLock);
2343                                 break;
2344                         }
2345
2346                 case OCLASS_AMPROC:
2347                         {
2348                                 Relation        amprocDesc;
2349                                 ScanKeyData skey[1];
2350                                 SysScanDesc amscan;
2351                                 HeapTuple       tup;
2352                                 Form_pg_amproc amprocForm;
2353                                 StringInfoData opfam;
2354
2355                                 amprocDesc = heap_open(AccessMethodProcedureRelationId,
2356                                                                            AccessShareLock);
2357
2358                                 ScanKeyInit(&skey[0],
2359                                                         ObjectIdAttributeNumber,
2360                                                         BTEqualStrategyNumber, F_OIDEQ,
2361                                                         ObjectIdGetDatum(object->objectId));
2362
2363                                 amscan = systable_beginscan(amprocDesc, AccessMethodProcedureOidIndexId, true,
2364                                                                                         SnapshotNow, 1, skey);
2365
2366                                 tup = systable_getnext(amscan);
2367
2368                                 if (!HeapTupleIsValid(tup))
2369                                         elog(ERROR, "could not find tuple for amproc entry %u",
2370                                                  object->objectId);
2371
2372                                 amprocForm = (Form_pg_amproc) GETSTRUCT(tup);
2373
2374                                 initStringInfo(&opfam);
2375                                 getOpFamilyDescription(&opfam, amprocForm->amprocfamily);
2376
2377                                 /*
2378                                  * translator: %d is the function number, the first %s is the
2379                                  * textual form of the function with arguments, and the second
2380                                  * %s is the description of the operator family.
2381                                  */
2382                                 appendStringInfo(&buffer, _("function %d %s of %s"),
2383                                                                  amprocForm->amprocnum,
2384                                                                  format_procedure(amprocForm->amproc),
2385                                                                  opfam.data);
2386                                 pfree(opfam.data);
2387
2388                                 systable_endscan(amscan);
2389                                 heap_close(amprocDesc, AccessShareLock);
2390                                 break;
2391                         }
2392
2393                 case OCLASS_REWRITE:
2394                         {
2395                                 Relation        ruleDesc;
2396                                 ScanKeyData skey[1];
2397                                 SysScanDesc rcscan;
2398                                 HeapTuple       tup;
2399                                 Form_pg_rewrite rule;
2400
2401                                 ruleDesc = heap_open(RewriteRelationId, AccessShareLock);
2402
2403                                 ScanKeyInit(&skey[0],
2404                                                         ObjectIdAttributeNumber,
2405                                                         BTEqualStrategyNumber, F_OIDEQ,
2406                                                         ObjectIdGetDatum(object->objectId));
2407
2408                                 rcscan = systable_beginscan(ruleDesc, RewriteOidIndexId, true,
2409                                                                                         SnapshotNow, 1, skey);
2410
2411                                 tup = systable_getnext(rcscan);
2412
2413                                 if (!HeapTupleIsValid(tup))
2414                                         elog(ERROR, "could not find tuple for rule %u",
2415                                                  object->objectId);
2416
2417                                 rule = (Form_pg_rewrite) GETSTRUCT(tup);
2418
2419                                 appendStringInfo(&buffer, _("rule %s on "),
2420                                                                  NameStr(rule->rulename));
2421                                 getRelationDescription(&buffer, rule->ev_class);
2422
2423                                 systable_endscan(rcscan);
2424                                 heap_close(ruleDesc, AccessShareLock);
2425                                 break;
2426                         }
2427
2428                 case OCLASS_TRIGGER:
2429                         {
2430                                 Relation        trigDesc;
2431                                 ScanKeyData skey[1];
2432                                 SysScanDesc tgscan;
2433                                 HeapTuple       tup;
2434                                 Form_pg_trigger trig;
2435
2436                                 trigDesc = heap_open(TriggerRelationId, AccessShareLock);
2437
2438                                 ScanKeyInit(&skey[0],
2439                                                         ObjectIdAttributeNumber,
2440                                                         BTEqualStrategyNumber, F_OIDEQ,
2441                                                         ObjectIdGetDatum(object->objectId));
2442
2443                                 tgscan = systable_beginscan(trigDesc, TriggerOidIndexId, true,
2444                                                                                         SnapshotNow, 1, skey);
2445
2446                                 tup = systable_getnext(tgscan);
2447
2448                                 if (!HeapTupleIsValid(tup))
2449                                         elog(ERROR, "could not find tuple for trigger %u",
2450                                                  object->objectId);
2451
2452                                 trig = (Form_pg_trigger) GETSTRUCT(tup);
2453
2454                                 appendStringInfo(&buffer, _("trigger %s on "),
2455                                                                  NameStr(trig->tgname));
2456                                 getRelationDescription(&buffer, trig->tgrelid);
2457
2458                                 systable_endscan(tgscan);
2459                                 heap_close(trigDesc, AccessShareLock);
2460                                 break;
2461                         }
2462
2463                 case OCLASS_SCHEMA:
2464                         {
2465                                 char       *nspname;
2466
2467                                 nspname = get_namespace_name(object->objectId);
2468                                 if (!nspname)
2469                                         elog(ERROR, "cache lookup failed for namespace %u",
2470                                                  object->objectId);
2471                                 appendStringInfo(&buffer, _("schema %s"), nspname);
2472                                 break;
2473                         }
2474
2475                 case OCLASS_TSPARSER:
2476                         {
2477                                 HeapTuple       tup;
2478
2479                                 tup = SearchSysCache1(TSPARSEROID,
2480                                                                           ObjectIdGetDatum(object->objectId));
2481                                 if (!HeapTupleIsValid(tup))
2482                                         elog(ERROR, "cache lookup failed for text search parser %u",
2483                                                  object->objectId);
2484                                 appendStringInfo(&buffer, _("text search parser %s"),
2485                                          NameStr(((Form_pg_ts_parser) GETSTRUCT(tup))->prsname));
2486                                 ReleaseSysCache(tup);
2487                                 break;
2488                         }
2489
2490                 case OCLASS_TSDICT:
2491                         {
2492                                 HeapTuple       tup;
2493
2494                                 tup = SearchSysCache1(TSDICTOID,
2495                                                                           ObjectIdGetDatum(object->objectId));
2496                                 if (!HeapTupleIsValid(tup))
2497                                         elog(ERROR, "cache lookup failed for text search dictionary %u",
2498                                                  object->objectId);
2499                                 appendStringInfo(&buffer, _("text search dictionary %s"),
2500                                           NameStr(((Form_pg_ts_dict) GETSTRUCT(tup))->dictname));
2501                                 ReleaseSysCache(tup);
2502                                 break;
2503                         }
2504
2505                 case OCLASS_TSTEMPLATE:
2506                         {
2507                                 HeapTuple       tup;
2508
2509                                 tup = SearchSysCache1(TSTEMPLATEOID,
2510                                                                           ObjectIdGetDatum(object->objectId));
2511                                 if (!HeapTupleIsValid(tup))
2512                                         elog(ERROR, "cache lookup failed for text search template %u",
2513                                                  object->objectId);
2514                                 appendStringInfo(&buffer, _("text search template %s"),
2515                                   NameStr(((Form_pg_ts_template) GETSTRUCT(tup))->tmplname));
2516                                 ReleaseSysCache(tup);
2517                                 break;
2518                         }
2519
2520                 case OCLASS_TSCONFIG:
2521                         {
2522                                 HeapTuple       tup;
2523
2524                                 tup = SearchSysCache1(TSCONFIGOID,
2525                                                                           ObjectIdGetDatum(object->objectId));
2526                                 if (!HeapTupleIsValid(tup))
2527                                         elog(ERROR, "cache lookup failed for text search configuration %u",
2528                                                  object->objectId);
2529                                 appendStringInfo(&buffer, _("text search configuration %s"),
2530                                          NameStr(((Form_pg_ts_config) GETSTRUCT(tup))->cfgname));
2531                                 ReleaseSysCache(tup);
2532                                 break;
2533                         }
2534
2535                 case OCLASS_ROLE:
2536                         {
2537                                 appendStringInfo(&buffer, _("role %s"),
2538                                                                  GetUserNameFromId(object->objectId));
2539                                 break;
2540                         }
2541
2542                 case OCLASS_DATABASE:
2543                         {
2544                                 char       *datname;
2545
2546                                 datname = get_database_name(object->objectId);
2547                                 if (!datname)
2548                                         elog(ERROR, "cache lookup failed for database %u",
2549                                                  object->objectId);
2550                                 appendStringInfo(&buffer, _("database %s"), datname);
2551                                 break;
2552                         }
2553
2554                 case OCLASS_TBLSPACE:
2555                         {
2556                                 char       *tblspace;
2557
2558                                 tblspace = get_tablespace_name(object->objectId);
2559                                 if (!tblspace)
2560                                         elog(ERROR, "cache lookup failed for tablespace %u",
2561                                                  object->objectId);
2562                                 appendStringInfo(&buffer, _("tablespace %s"), tblspace);
2563                                 break;
2564                         }
2565
2566                 case OCLASS_FDW:
2567                         {
2568                                 ForeignDataWrapper *fdw;
2569
2570                                 fdw = GetForeignDataWrapper(object->objectId);
2571                                 appendStringInfo(&buffer, _("foreign-data wrapper %s"), fdw->fdwname);
2572                                 break;
2573                         }
2574
2575                 case OCLASS_FOREIGN_SERVER:
2576                         {
2577                                 ForeignServer *srv;
2578
2579                                 srv = GetForeignServer(object->objectId);
2580                                 appendStringInfo(&buffer, _("server %s"), srv->servername);
2581                                 break;
2582                         }
2583
2584                 case OCLASS_USER_MAPPING:
2585                         {
2586                                 HeapTuple       tup;
2587                                 Oid                     useid;
2588                                 char       *usename;
2589
2590                                 tup = SearchSysCache1(USERMAPPINGOID,
2591                                                                           ObjectIdGetDatum(object->objectId));
2592                                 if (!HeapTupleIsValid(tup))
2593                                         elog(ERROR, "cache lookup failed for user mapping %u",
2594                                                  object->objectId);
2595
2596                                 useid = ((Form_pg_user_mapping) GETSTRUCT(tup))->umuser;
2597
2598                                 ReleaseSysCache(tup);
2599
2600                                 if (OidIsValid(useid))
2601                                         usename = GetUserNameFromId(useid);
2602                                 else
2603                                         usename = "public";
2604
2605                                 appendStringInfo(&buffer, _("user mapping for %s"), usename);
2606                                 break;
2607                         }
2608
2609                 case OCLASS_DEFACL:
2610                         {
2611                                 Relation        defaclrel;
2612                                 ScanKeyData skey[1];
2613                                 SysScanDesc rcscan;
2614                                 HeapTuple       tup;
2615                                 Form_pg_default_acl defacl;
2616
2617                                 defaclrel = heap_open(DefaultAclRelationId, AccessShareLock);
2618
2619                                 ScanKeyInit(&skey[0],
2620                                                         ObjectIdAttributeNumber,
2621                                                         BTEqualStrategyNumber, F_OIDEQ,
2622                                                         ObjectIdGetDatum(object->objectId));
2623
2624                                 rcscan = systable_beginscan(defaclrel, DefaultAclOidIndexId,
2625                                                                                         true, SnapshotNow, 1, skey);
2626
2627                                 tup = systable_getnext(rcscan);
2628
2629                                 if (!HeapTupleIsValid(tup))
2630                                         elog(ERROR, "could not find tuple for default ACL %u",
2631                                                  object->objectId);
2632
2633                                 defacl = (Form_pg_default_acl) GETSTRUCT(tup);
2634
2635                                 switch (defacl->defaclobjtype)
2636                                 {
2637                                         case DEFACLOBJ_RELATION:
2638                                                 appendStringInfo(&buffer,
2639                                                                                  _("default privileges on new relations belonging to role %s"),
2640                                                                           GetUserNameFromId(defacl->defaclrole));
2641                                                 break;
2642                                         case DEFACLOBJ_SEQUENCE:
2643                                                 appendStringInfo(&buffer,
2644                                                                                  _("default privileges on new sequences belonging to role %s"),
2645                                                                           GetUserNameFromId(defacl->defaclrole));
2646                                                 break;
2647                                         case DEFACLOBJ_FUNCTION:
2648                                                 appendStringInfo(&buffer,
2649                                                                                  _("default privileges on new functions belonging to role %s"),
2650                                                                           GetUserNameFromId(defacl->defaclrole));
2651                                                 break;
2652                                         default:
2653                                                 /* shouldn't get here */
2654                                                 appendStringInfo(&buffer,
2655                                                                 _("default privileges belonging to role %s"),
2656                                                                           GetUserNameFromId(defacl->defaclrole));
2657                                                 break;
2658                                 }
2659
2660                                 if (OidIsValid(defacl->defaclnamespace))
2661                                 {
2662                                         appendStringInfo(&buffer,
2663                                                                          _(" in schema %s"),
2664                                                                 get_namespace_name(defacl->defaclnamespace));
2665                                 }
2666
2667                                 systable_endscan(rcscan);
2668                                 heap_close(defaclrel, AccessShareLock);
2669                                 break;
2670                         }
2671
2672                 default:
2673                         appendStringInfo(&buffer, "unrecognized object %u %u %d",
2674                                                          object->classId,
2675                                                          object->objectId,
2676                                                          object->objectSubId);
2677                         break;
2678         }
2679
2680         return buffer.data;
2681 }
2682
2683 /*
2684  * subroutine for getObjectDescription: describe a relation
2685  */
2686 static void
2687 getRelationDescription(StringInfo buffer, Oid relid)
2688 {
2689         HeapTuple       relTup;
2690         Form_pg_class relForm;
2691         char       *nspname;
2692         char       *relname;
2693
2694         relTup = SearchSysCache1(RELOID,
2695                                                          ObjectIdGetDatum(relid));
2696         if (!HeapTupleIsValid(relTup))
2697                 elog(ERROR, "cache lookup failed for relation %u", relid);
2698         relForm = (Form_pg_class) GETSTRUCT(relTup);
2699
2700         /* Qualify the name if not visible in search path */
2701         if (RelationIsVisible(relid))
2702                 nspname = NULL;
2703         else
2704                 nspname = get_namespace_name(relForm->relnamespace);
2705
2706         relname = quote_qualified_identifier(nspname, NameStr(relForm->relname));
2707
2708         switch (relForm->relkind)
2709         {
2710                 case RELKIND_RELATION:
2711                         appendStringInfo(buffer, _("table %s"),
2712                                                          relname);
2713                         break;
2714                 case RELKIND_INDEX:
2715                         appendStringInfo(buffer, _("index %s"),
2716                                                          relname);
2717                         break;
2718                 case RELKIND_SEQUENCE:
2719                         appendStringInfo(buffer, _("sequence %s"),
2720                                                          relname);
2721                         break;
2722                 case RELKIND_UNCATALOGED:
2723                         appendStringInfo(buffer, _("uncataloged table %s"),
2724                                                          relname);
2725                         break;
2726                 case RELKIND_TOASTVALUE:
2727                         appendStringInfo(buffer, _("toast table %s"),
2728                                                          relname);
2729                         break;
2730                 case RELKIND_VIEW:
2731                         appendStringInfo(buffer, _("view %s"),
2732                                                          relname);
2733                         break;
2734                 case RELKIND_COMPOSITE_TYPE:
2735                         appendStringInfo(buffer, _("composite type %s"),
2736                                                          relname);
2737                         break;
2738                 default:
2739                         /* shouldn't get here */
2740                         appendStringInfo(buffer, _("relation %s"),
2741                                                          relname);
2742                         break;
2743         }
2744
2745         ReleaseSysCache(relTup);
2746 }
2747
2748 /*
2749  * subroutine for getObjectDescription: describe an operator family
2750  */
2751 static void
2752 getOpFamilyDescription(StringInfo buffer, Oid opfid)
2753 {
2754         HeapTuple       opfTup;
2755         Form_pg_opfamily opfForm;
2756         HeapTuple       amTup;
2757         Form_pg_am      amForm;
2758         char       *nspname;
2759
2760         opfTup = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfid));
2761         if (!HeapTupleIsValid(opfTup))
2762                 elog(ERROR, "cache lookup failed for opfamily %u", opfid);
2763         opfForm = (Form_pg_opfamily) GETSTRUCT(opfTup);
2764
2765         amTup = SearchSysCache1(AMOID, ObjectIdGetDatum(opfForm->opfmethod));
2766         if (!HeapTupleIsValid(amTup))
2767                 elog(ERROR, "cache lookup failed for access method %u",
2768                          opfForm->opfmethod);
2769         amForm = (Form_pg_am) GETSTRUCT(amTup);
2770
2771         /* Qualify the name if not visible in search path */
2772         if (OpfamilyIsVisible(opfid))
2773                 nspname = NULL;
2774         else
2775                 nspname = get_namespace_name(opfForm->opfnamespace);
2776
2777         appendStringInfo(buffer, _("operator family %s for access method %s"),
2778                                          quote_qualified_identifier(nspname,
2779                                                                                                 NameStr(opfForm->opfname)),
2780                                          NameStr(amForm->amname));
2781
2782         ReleaseSysCache(amTup);
2783         ReleaseSysCache(opfTup);
2784 }