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