granicus.if.org Git - postgresql/blob - src/backend/commands/indexcmds.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * indexcmds.c
   4  *        POSTGRES define and remove index code.
   5  *
   6  * Portions Copyright (c) 1996-2002, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/commands/indexcmds.c,v 1.94 2002/12/13 19:45:50 tgl Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15
  16 #include "postgres.h"
  17
  18 #include "access/heapam.h"
  19 #include "catalog/catalog.h"
  20 #include "catalog/catname.h"
  21 #include "catalog/dependency.h"
  22 #include "catalog/index.h"
  23 #include "catalog/namespace.h"
  24 #include "catalog/pg_opclass.h"
  25 #include "catalog/pg_proc.h"
  26 #include "commands/defrem.h"
  27 #include "executor/executor.h"
  28 #include "miscadmin.h"
  29 #include "optimizer/clauses.h"
  30 #include "optimizer/planmain.h"
  31 #include "optimizer/prep.h"
  32 #include "parser/parsetree.h"
  33 #include "parser/parse_coerce.h"
  34 #include "parser/parse_func.h"
  35 #include "utils/acl.h"
  36 #include "utils/builtins.h"
  37 #include "utils/lsyscache.h"
  38 #include "utils/syscache.h"
  39
  40
  41 #define IsFuncIndex(ATTR_LIST) (((IndexElem*)lfirst(ATTR_LIST))->funcname != NIL)
  42
  43 /* non-export function prototypes */
  44 static void CheckPredicate(List *predList, List *rangeTable, Oid baseRelOid);
  45 static void FuncIndexArgs(IndexInfo *indexInfo, Oid *classOidP,
  46                           IndexElem *funcIndex,
  47                           Oid relId,
  48                           char *accessMethodName, Oid accessMethodId);
  49 static void NormIndexAttrs(IndexInfo *indexInfo, Oid *classOidP,
  50                            List *attList,
  51                            Oid relId,
  52                            char *accessMethodName, Oid accessMethodId);
  53 static Oid GetAttrOpClass(IndexElem *attribute, Oid attrType,
  54                            char *accessMethodName, Oid accessMethodId);
  55 static Oid      GetDefaultOpClass(Oid attrType, Oid accessMethodId);
  56
  57 /*
  58  * DefineIndex
  59  *              Creates a new index.
  60  *
  61  * 'attributeList' is a list of IndexElem specifying either a functional
  62  *              index or a list of attributes to index on.
  63  * 'predicate' is the qual specified in the where clause.
  64  * 'rangetable' is needed to interpret the predicate.
  65  */
  66 void
  67 DefineIndex(RangeVar *heapRelation,
  68                         char *indexRelationName,
  69                         char *accessMethodName,
  70                         List *attributeList,
  71                         bool unique,
  72                         bool primary,
  73                         bool isconstraint,
  74                         Expr *predicate,
  75                         List *rangetable)
  76 {
  77         Oid                *classObjectId;
  78         Oid                     accessMethodId;
  79         Oid                     relationId;
  80         Oid                     namespaceId;
  81         Relation        rel;
  82         HeapTuple       tuple;
  83         Form_pg_am      accessMethodForm;
  84         IndexInfo  *indexInfo;
  85         int                     numberOfAttributes;
  86         List       *cnfPred = NIL;
  87
  88         /*
  89          * count attributes in index
  90          */
  91         numberOfAttributes = length(attributeList);
  92         if (numberOfAttributes <= 0)
  93                 elog(ERROR, "DefineIndex: must specify at least one attribute");
  94         if (numberOfAttributes > INDEX_MAX_KEYS)
  95                 elog(ERROR, "Cannot use more than %d attributes in an index",
  96                          INDEX_MAX_KEYS);
  97
  98         /*
  99          * Open heap relation, acquire a suitable lock on it, remember its OID
 100          */
 101         rel = heap_openrv(heapRelation, ShareLock);
 102
 103         /* Note: during bootstrap may see uncataloged relation */
 104         if (rel->rd_rel->relkind != RELKIND_RELATION &&
 105                 rel->rd_rel->relkind != RELKIND_UNCATALOGED)
 106                 elog(ERROR, "DefineIndex: relation \"%s\" is not a table",
 107                          heapRelation->relname);
 108
 109         relationId = RelationGetRelid(rel);
 110         namespaceId = RelationGetNamespace(rel);
 111
 112         if (!IsBootstrapProcessingMode() &&
 113                 IsSystemRelation(rel) &&
 114                 !IndexesAreActive(rel))
 115                 elog(ERROR, "Existing indexes are inactive. REINDEX first");
 116
 117         heap_close(rel, NoLock);
 118
 119         /*
 120          * Verify we (still) have CREATE rights in the rel's namespace.
 121          * (Presumably we did when the rel was created, but maybe not
 122          * anymore.) Skip check if bootstrapping, since permissions machinery
 123          * may not be working yet.
 124          */
 125         if (!IsBootstrapProcessingMode())
 126         {
 127                 AclResult       aclresult;
 128
 129                 aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
 130                                                                                   ACL_CREATE);
 131                 if (aclresult != ACLCHECK_OK)
 132                         aclcheck_error(aclresult, get_namespace_name(namespaceId));
 133         }
 134
 135         /*
 136          * look up the access method, verify it can handle the requested
 137          * features
 138          */
 139         tuple = SearchSysCache(AMNAME,
 140                                                    PointerGetDatum(accessMethodName),
 141                                                    0, 0, 0);
 142         if (!HeapTupleIsValid(tuple))
 143                 elog(ERROR, "DefineIndex: access method \"%s\" not found",
 144                          accessMethodName);
 145         accessMethodId = HeapTupleGetOid(tuple);
 146         accessMethodForm = (Form_pg_am) GETSTRUCT(tuple);
 147
 148         if (unique && !accessMethodForm->amcanunique)
 149                 elog(ERROR, "DefineIndex: access method \"%s\" does not support UNIQUE indexes",
 150                          accessMethodName);
 151         if (numberOfAttributes > 1 && !accessMethodForm->amcanmulticol)
 152                 elog(ERROR, "DefineIndex: access method \"%s\" does not support multi-column indexes",
 153                          accessMethodName);
 154
 155         ReleaseSysCache(tuple);
 156
 157         /*
 158          * Convert the partial-index predicate from parsetree form to an
 159          * implicit-AND qual expression, for easier evaluation at runtime.
 160          * While we are at it, we reduce it to a canonical (CNF or DNF) form
 161          * to simplify the task of proving implications.
 162          */
 163         if (predicate)
 164         {
 165                 cnfPred = canonicalize_qual((Expr *) copyObject(predicate), true);
 166                 fix_opfuncids((Node *) cnfPred);
 167                 CheckPredicate(cnfPred, rangetable, relationId);
 168         }
 169
 170         /*
 171          * Prepare arguments for index_create, primarily an IndexInfo
 172          * structure
 173          */
 174         indexInfo = makeNode(IndexInfo);
 175         indexInfo->ii_Predicate = cnfPred;
 176         indexInfo->ii_PredicateState = (List *)
 177                 ExecInitExpr((Expr *) cnfPred, NULL);
 178         indexInfo->ii_FuncOid = InvalidOid;
 179         indexInfo->ii_Unique = unique;
 180
 181         if (IsFuncIndex(attributeList))
 182         {
 183                 IndexElem  *funcIndex = (IndexElem *) lfirst(attributeList);
 184                 int                     nargs;
 185
 186                 /* Parser should have given us only one list item, but check */
 187                 if (numberOfAttributes != 1)
 188                         elog(ERROR, "Functional index can only have one attribute");
 189
 190                 nargs = length(funcIndex->args);
 191                 if (nargs > INDEX_MAX_KEYS)
 192                         elog(ERROR, "Index function can take at most %d arguments",
 193                                  INDEX_MAX_KEYS);
 194
 195                 indexInfo->ii_NumIndexAttrs = 1;
 196                 indexInfo->ii_NumKeyAttrs = nargs;
 197
 198                 classObjectId = (Oid *) palloc(sizeof(Oid));
 199
 200                 FuncIndexArgs(indexInfo, classObjectId, funcIndex,
 201                                           relationId, accessMethodName, accessMethodId);
 202         }
 203         else
 204         {
 205                 indexInfo->ii_NumIndexAttrs = numberOfAttributes;
 206                 indexInfo->ii_NumKeyAttrs = numberOfAttributes;
 207
 208                 classObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
 209
 210                 NormIndexAttrs(indexInfo, classObjectId, attributeList,
 211                                            relationId, accessMethodName, accessMethodId);
 212         }
 213
 214         index_create(relationId, indexRelationName,
 215                                  indexInfo, accessMethodId, classObjectId,
 216                                  primary, isconstraint, allowSystemTableMods);
 217
 218         /*
 219          * We update the relation's pg_class tuple even if it already has
 220          * relhasindex = true.  This is needed to cause a shared-cache-inval
 221          * message to be sent for the pg_class tuple, which will cause other
 222          * backends to flush their relcache entries and in particular their
 223          * cached lists of the indexes for this relation.
 224          */
 225         setRelhasindex(relationId, true, primary, InvalidOid);
 226 }
 227
 228
 229 /*
 230  * CheckPredicate
 231  *              Checks that the given list of partial-index predicates refer
 232  *              (via the given range table) only to the given base relation oid.
 233  *
 234  * This used to also constrain the form of the predicate to forms that
 235  * indxpath.c could do something with.  However, that seems overly
 236  * restrictive.  One useful application of partial indexes is to apply
 237  * a UNIQUE constraint across a subset of a table, and in that scenario
 238  * any evaluatable predicate will work.  So accept any predicate here
 239  * (except ones requiring a plan), and let indxpath.c fend for itself.
 240  */
 241
 242 static void
 243 CheckPredicate(List *predList, List *rangeTable, Oid baseRelOid)
 244 {
 245         if (length(rangeTable) != 1 || getrelid(1, rangeTable) != baseRelOid)
 246                 elog(ERROR,
 247                  "Partial-index predicates may refer only to the base relation");
 248
 249         /*
 250          * We don't currently support generation of an actual query plan for a
 251          * predicate, only simple scalar expressions; hence these
 252          * restrictions.
 253          */
 254         if (contain_subplans((Node *) predList))
 255                 elog(ERROR, "Cannot use subselect in index predicate");
 256         if (contain_agg_clause((Node *) predList))
 257                 elog(ERROR, "Cannot use aggregate in index predicate");
 258
 259         /*
 260          * A predicate using mutable functions is probably wrong, for the same
 261          * reasons that we don't allow a functional index to use one.
 262          */
 263         if (contain_mutable_functions((Node *) predList))
 264                 elog(ERROR, "Functions in index predicate must be marked isImmutable");
 265 }
 266
 267
 268 static void
 269 FuncIndexArgs(IndexInfo *indexInfo,
 270                           Oid *classOidP,
 271                           IndexElem *funcIndex,
 272                           Oid relId,
 273                           char *accessMethodName,
 274                           Oid accessMethodId)
 275 {
 276         Oid                     argTypes[FUNC_MAX_ARGS];
 277         List       *arglist;
 278         int                     nargs = 0;
 279         int                     i;
 280         FuncDetailCode fdresult;
 281         Oid                     funcid;
 282         Oid                     rettype;
 283         bool            retset;
 284         Oid                *true_typeids;
 285
 286         /*
 287          * process the function arguments, which are a list of T_String
 288          * (someday ought to allow more general expressions?)
 289          *
 290          * Note caller already checked that list is not too long.
 291          */
 292         MemSet(argTypes, 0, sizeof(argTypes));
 293
 294         foreach(arglist, funcIndex->args)
 295         {
 296                 char       *arg = strVal(lfirst(arglist));
 297                 HeapTuple       tuple;
 298                 Form_pg_attribute att;
 299
 300                 tuple = SearchSysCacheAttName(relId, arg);
 301                 if (!HeapTupleIsValid(tuple))
 302                         elog(ERROR, "DefineIndex: attribute \"%s\" not found", arg);
 303                 att = (Form_pg_attribute) GETSTRUCT(tuple);
 304                 indexInfo->ii_KeyAttrNumbers[nargs] = att->attnum;
 305                 argTypes[nargs] = att->atttypid;
 306                 ReleaseSysCache(tuple);
 307                 nargs++;
 308         }
 309
 310         /*
 311          * Lookup the function procedure to get its OID and result type.
 312          *
 313          * We rely on parse_func.c to find the correct function in the possible
 314          * presence of binary-compatible types.  However, parse_func may do
 315          * too much: it will accept a function that requires run-time coercion
 316          * of input types, and the executor is not currently set up to support
 317          * that.  So, check to make sure that the selected function has
 318          * exact-match or binary-compatible input types.
 319          */
 320         fdresult = func_get_detail(funcIndex->funcname, funcIndex->args,
 321                                                            nargs, argTypes,
 322                                                            &funcid, &rettype, &retset,
 323                                                            &true_typeids);
 324         if (fdresult != FUNCDETAIL_NORMAL)
 325         {
 326                 if (fdresult == FUNCDETAIL_AGGREGATE)
 327                         elog(ERROR, "DefineIndex: functional index may not use an aggregate function");
 328                 else if (fdresult == FUNCDETAIL_COERCION)
 329                         elog(ERROR, "DefineIndex: functional index must use a real function, not a type coercion"
 330                                  "\n\tTry specifying the index opclass you want to use, instead");
 331                 else
 332                         func_error("DefineIndex", funcIndex->funcname, nargs, argTypes,
 333                                            NULL);
 334         }
 335
 336         if (retset)
 337                 elog(ERROR, "DefineIndex: cannot index on a function returning a set");
 338
 339         for (i = 0; i < nargs; i++)
 340         {
 341                 if (!IsBinaryCoercible(argTypes[i], true_typeids[i]))
 342                         func_error("DefineIndex", funcIndex->funcname, nargs, true_typeids,
 343                                            "Index function must be binary-compatible with table datatype");
 344         }
 345
 346         /*
 347          * Require that the function be marked immutable.  Using a mutable
 348          * function for a functional index is highly questionable, since if
 349          * you aren't going to get the same result for the same data every
 350          * time, it's not clear what the index entries mean at all.
 351          */
 352         if (func_volatile(funcid) != PROVOLATILE_IMMUTABLE)
 353                 elog(ERROR, "DefineIndex: index function must be marked isImmutable");
 354
 355         /* Process opclass, using func return type as default type */
 356
 357         classOidP[0] = GetAttrOpClass(funcIndex, rettype,
 358                                                                   accessMethodName, accessMethodId);
 359
 360         /* OK, return results */
 361
 362         indexInfo->ii_FuncOid = funcid;
 363         /* Need to do the fmgr function lookup now, too */
 364         fmgr_info(funcid, &indexInfo->ii_FuncInfo);
 365 }
 366
 367 static void
 368 NormIndexAttrs(IndexInfo *indexInfo,
 369                            Oid *classOidP,
 370                            List *attList,       /* list of IndexElem's */
 371                            Oid relId,
 372                            char *accessMethodName,
 373                            Oid accessMethodId)
 374 {
 375         List       *rest;
 376         int                     attn = 0;
 377
 378         /*
 379          * process attributeList
 380          */
 381         foreach(rest, attList)
 382         {
 383                 IndexElem  *attribute = (IndexElem *) lfirst(rest);
 384                 HeapTuple       atttuple;
 385                 Form_pg_attribute attform;
 386
 387                 if (attribute->name == NULL)
 388                         elog(ERROR, "missing attribute for define index");
 389
 390                 atttuple = SearchSysCacheAttName(relId, attribute->name);
 391                 if (!HeapTupleIsValid(atttuple))
 392                         elog(ERROR, "DefineIndex: attribute \"%s\" not found",
 393                                  attribute->name);
 394                 attform = (Form_pg_attribute) GETSTRUCT(atttuple);
 395
 396                 indexInfo->ii_KeyAttrNumbers[attn] = attform->attnum;
 397
 398                 classOidP[attn] = GetAttrOpClass(attribute, attform->atttypid,
 399                                                                            accessMethodName, accessMethodId);
 400
 401                 ReleaseSysCache(atttuple);
 402                 attn++;
 403         }
 404 }
 405
 406 static Oid
 407 GetAttrOpClass(IndexElem *attribute, Oid attrType,
 408                            char *accessMethodName, Oid accessMethodId)
 409 {
 410         char       *schemaname;
 411         char       *opcname;
 412         HeapTuple       tuple;
 413         Oid                     opClassId,
 414                                 opInputType;
 415
 416         if (attribute->opclass == NIL)
 417         {
 418                 /* no operator class specified, so find the default */
 419                 opClassId = GetDefaultOpClass(attrType, accessMethodId);
 420                 if (!OidIsValid(opClassId))
 421                         elog(ERROR, "data type %s has no default operator class for access method \"%s\""
 422                                  "\n\tYou must specify an operator class for the index or define a"
 423                                  "\n\tdefault operator class for the data type",
 424                                  format_type_be(attrType), accessMethodName);
 425                 return opClassId;
 426         }
 427
 428         /*
 429          * Specific opclass name given, so look up the opclass.
 430          */
 431
 432         /* deconstruct the name list */
 433         DeconstructQualifiedName(attribute->opclass, &schemaname, &opcname);
 434
 435         if (schemaname)
 436         {
 437                 /* Look in specific schema only */
 438                 Oid                     namespaceId;
 439
 440                 namespaceId = LookupExplicitNamespace(schemaname);
 441                 tuple = SearchSysCache(CLAAMNAMENSP,
 442                                                            ObjectIdGetDatum(accessMethodId),
 443                                                            PointerGetDatum(opcname),
 444                                                            ObjectIdGetDatum(namespaceId),
 445                                                            0);
 446         }
 447         else
 448         {
 449                 /* Unqualified opclass name, so search the search path */
 450                 opClassId = OpclassnameGetOpcid(accessMethodId, opcname);
 451                 if (!OidIsValid(opClassId))
 452                         elog(ERROR, "DefineIndex: operator class \"%s\" not supported by access method \"%s\"",
 453                                  opcname, accessMethodName);
 454                 tuple = SearchSysCache(CLAOID,
 455                                                            ObjectIdGetDatum(opClassId),
 456                                                            0, 0, 0);
 457         }
 458
 459         if (!HeapTupleIsValid(tuple))
 460                 elog(ERROR, "DefineIndex: operator class \"%s\" not supported by access method \"%s\"",
 461                          NameListToString(attribute->opclass), accessMethodName);
 462
 463         /*
 464          * Verify that the index operator class accepts this datatype.  Note
 465          * we will accept binary compatibility.
 466          */
 467         opClassId = HeapTupleGetOid(tuple);
 468         opInputType = ((Form_pg_opclass) GETSTRUCT(tuple))->opcintype;
 469
 470         if (!IsBinaryCoercible(attrType, opInputType))
 471                 elog(ERROR, "operator class \"%s\" does not accept data type %s",
 472                  NameListToString(attribute->opclass), format_type_be(attrType));
 473
 474         ReleaseSysCache(tuple);
 475
 476         return opClassId;
 477 }
 478
 479 static Oid
 480 GetDefaultOpClass(Oid attrType, Oid accessMethodId)
 481 {
 482         OpclassCandidateList opclass;
 483         int                     nexact = 0;
 484         int                     ncompatible = 0;
 485         Oid                     exactOid = InvalidOid;
 486         Oid                     compatibleOid = InvalidOid;
 487
 488         /* If it's a domain, look at the base type instead */
 489         attrType = getBaseType(attrType);
 490
 491         /*
 492          * We scan through all the opclasses available for the access method,
 493          * looking for one that is marked default and matches the target type
 494          * (either exactly or binary-compatibly, but prefer an exact match).
 495          *
 496          * We could find more than one binary-compatible match, in which case we
 497          * require the user to specify which one he wants.      If we find more
 498          * than one exact match, then someone put bogus entries in pg_opclass.
 499          *
 500          * The initial search is done by namespace.c so that we only consider
 501          * opclasses visible in the current namespace search path.
 502          */
 503         for (opclass = OpclassGetCandidates(accessMethodId);
 504                  opclass != NULL;
 505                  opclass = opclass->next)
 506         {
 507                 if (opclass->opcdefault)
 508                 {
 509                         if (opclass->opcintype == attrType)
 510                         {
 511                                 nexact++;
 512                                 exactOid = opclass->oid;
 513                         }
 514                         else if (IsBinaryCoercible(attrType, opclass->opcintype))
 515                         {
 516                                 ncompatible++;
 517                                 compatibleOid = opclass->oid;
 518                         }
 519                 }
 520         }
 521
 522         if (nexact == 1)
 523                 return exactOid;
 524         if (nexact != 0)
 525                 elog(ERROR, "pg_opclass contains multiple default opclasses for data type %s",
 526                          format_type_be(attrType));
 527         if (ncompatible == 1)
 528                 return compatibleOid;
 529
 530         return InvalidOid;
 531 }
 532
 533 /*
 534  * RemoveIndex
 535  *              Deletes an index.
 536  */
 537 void
 538 RemoveIndex(RangeVar *relation, DropBehavior behavior)
 539 {
 540         Oid                     indOid;
 541         char            relkind;
 542         ObjectAddress object;
 543
 544         indOid = RangeVarGetRelid(relation, false);
 545         relkind = get_rel_relkind(indOid);
 546         if (relkind != RELKIND_INDEX)
 547                 elog(ERROR, "relation \"%s\" is of type \"%c\"",
 548                          relation->relname, relkind);
 549
 550         object.classId = RelOid_pg_class;
 551         object.objectId = indOid;
 552         object.objectSubId = 0;
 553
 554         performDeletion(&object, behavior);
 555 }
 556
 557 /*
 558  * ReindexIndex
 559  *              Recreate an index.
 560  */
 561 void
 562 ReindexIndex(RangeVar *indexRelation, bool force /* currently unused */ )
 563 {
 564         Oid                     indOid;
 565         HeapTuple       tuple;
 566         bool            overwrite;
 567
 568         /* Choose in-place-or-not mode */
 569         overwrite = IsIgnoringSystemIndexes();
 570
 571         indOid = RangeVarGetRelid(indexRelation, false);
 572         tuple = SearchSysCache(RELOID,
 573                                                    ObjectIdGetDatum(indOid),
 574                                                    0, 0, 0);
 575         if (!HeapTupleIsValid(tuple))
 576                 elog(ERROR, "index \"%s\" does not exist", indexRelation->relname);
 577
 578         if (((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_INDEX)
 579                 elog(ERROR, "relation \"%s\" is of type \"%c\"",
 580                          indexRelation->relname,
 581                          ((Form_pg_class) GETSTRUCT(tuple))->relkind);
 582
 583         if (IsSystemClass((Form_pg_class) GETSTRUCT(tuple)) &&
 584                 !IsToastClass((Form_pg_class) GETSTRUCT(tuple)))
 585         {
 586                 if (!allowSystemTableMods)
 587                         elog(ERROR, "\"%s\" is a system index. call REINDEX under standalone postgres with -O -P options",
 588                                  indexRelation->relname);
 589                 if (!IsIgnoringSystemIndexes())
 590                         elog(ERROR, "\"%s\" is a system index. call REINDEX under standalone postgres with -P -O options",
 591                                  indexRelation->relname);
 592         }
 593
 594         ReleaseSysCache(tuple);
 595
 596         /*
 597          * In-place REINDEX within a transaction block is dangerous, because
 598          * if the transaction is later rolled back we have no way to undo
 599          * truncation of the index's physical file.  Disallow it.
 600          */
 601         if (overwrite)
 602                 PreventTransactionChain((void *) indexRelation, "REINDEX");
 603
 604         if (!reindex_index(indOid, force, overwrite))
 605                 elog(WARNING, "index \"%s\" wasn't reindexed", indexRelation->relname);
 606 }
 607
 608 /*
 609  * ReindexTable
 610  *              Recreate indexes of a table.
 611  */
 612 void
 613 ReindexTable(RangeVar *relation, bool force)
 614 {
 615         Oid                     heapOid;
 616         char            relkind;
 617
 618         heapOid = RangeVarGetRelid(relation, false);
 619         relkind = get_rel_relkind(heapOid);
 620
 621         if (relkind != RELKIND_RELATION && relkind != RELKIND_TOASTVALUE)
 622                 elog(ERROR, "relation \"%s\" is of type \"%c\"",
 623                          relation->relname, relkind);
 624
 625         /*
 626          * In-place REINDEX within a transaction block is dangerous, because
 627          * if the transaction is later rolled back we have no way to undo
 628          * truncation of the index's physical file.  Disallow it.
 629          *
 630          * XXX we assume that in-place reindex will only be done if
 631          * IsIgnoringSystemIndexes() is true.
 632          */
 633         if (IsIgnoringSystemIndexes())
 634                 PreventTransactionChain((void *) relation, "REINDEX");
 635
 636         if (!reindex_relation(heapOid, force))
 637                 elog(WARNING, "table \"%s\" wasn't reindexed", relation->relname);
 638 }
 639
 640 /*
 641  * ReindexDatabase
 642  *              Recreate indexes of a database.
 643  */
 644 void
 645 ReindexDatabase(const char *dbname, bool force, bool all)
 646 {
 647         Relation        relationRelation;
 648         HeapScanDesc scan;
 649         HeapTuple       tuple;
 650         MemoryContext private_context;
 651         MemoryContext old;
 652         int                     relcnt,
 653                                 relalc,
 654                                 i,
 655                                 oncealc = 200;
 656         Oid                *relids = (Oid *) NULL;
 657
 658         AssertArg(dbname);
 659
 660         if (strcmp(dbname, DatabaseName) != 0)
 661                 elog(ERROR, "REINDEX DATABASE: Can be executed only on the currently open database.");
 662
 663         if (!(superuser() || is_dbadmin(MyDatabaseId)))
 664                 elog(ERROR, "REINDEX DATABASE: Permission denied.");
 665
 666         if (!allowSystemTableMods)
 667                 elog(ERROR, "must be called under standalone postgres with -O -P options");
 668         if (!IsIgnoringSystemIndexes())
 669                 elog(ERROR, "must be called under standalone postgres with -P -O options");
 670
 671         /*
 672          * We cannot run inside a user transaction block; if we were inside a
 673          * transaction, then our commit- and start-transaction-command calls
 674          * would not have the intended effect!
 675          */
 676         PreventTransactionChain((void *) dbname, "REINDEX");
 677
 678         /*
 679          * Create a memory context that will survive forced transaction
 680          * commits we do below.  Since it is a child of QueryContext, it will
 681          * go away eventually even if we suffer an error; there's no need for
 682          * special abort cleanup logic.
 683          */
 684         private_context = AllocSetContextCreate(QueryContext,
 685                                                                                         "ReindexDatabase",
 686                                                                                         ALLOCSET_DEFAULT_MINSIZE,
 687                                                                                         ALLOCSET_DEFAULT_INITSIZE,
 688                                                                                         ALLOCSET_DEFAULT_MAXSIZE);
 689
 690         /*
 691          * Scan pg_class to build a list of the relations we need to reindex.
 692          */
 693         relationRelation = heap_openr(RelationRelationName, AccessShareLock);
 694         scan = heap_beginscan(relationRelation, SnapshotNow, 0, NULL);
 695         relcnt = relalc = 0;
 696         while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 697         {
 698                 char            relkind;
 699
 700                 if (!all)
 701                 {
 702                         if (!(IsSystemClass((Form_pg_class) GETSTRUCT(tuple)) &&
 703                                   !IsToastClass((Form_pg_class) GETSTRUCT(tuple))))
 704                                 continue;
 705                 }
 706                 relkind = ((Form_pg_class) GETSTRUCT(tuple))->relkind;
 707                 if (relkind == RELKIND_RELATION || relkind == RELKIND_TOASTVALUE)
 708                 {
 709                         old = MemoryContextSwitchTo(private_context);
 710                         if (relcnt == 0)
 711                         {
 712                                 relalc = oncealc;
 713                                 relids = palloc(sizeof(Oid) * relalc);
 714                         }
 715                         else if (relcnt >= relalc)
 716                         {
 717                                 relalc *= 2;
 718                                 relids = repalloc(relids, sizeof(Oid) * relalc);
 719                         }
 720                         MemoryContextSwitchTo(old);
 721                         relids[relcnt] = HeapTupleGetOid(tuple);
 722                         relcnt++;
 723                 }
 724         }
 725         heap_endscan(scan);
 726         heap_close(relationRelation, AccessShareLock);
 727
 728         /* Now reindex each rel in a separate transaction */
 729         CommitTransactionCommand(true);
 730         for (i = 0; i < relcnt; i++)
 731         {
 732                 StartTransactionCommand(true);
 733                 SetQuerySnapshot();             /* might be needed for functional index */
 734                 if (reindex_relation(relids[i], force))
 735                         elog(NOTICE, "relation %u was reindexed", relids[i]);
 736                 CommitTransactionCommand(true);
 737         }
 738         /* Tell xact.c not to chain the upcoming commit */
 739         StartTransactionCommand(true);
 740
 741         MemoryContextDelete(private_context);
 742 }