LCOV - code coverage report
Current view: top level - src/backend/commands - indexcmds.c (source / functions) Hit Total Coverage
Test: PostgreSQL 17devel Lines: 1223 1334 91.7 %
Date: 2024-04-13 10:11:43 Functions: 26 26 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * indexcmds.c
       4             :  *    POSTGRES define and remove index code.
       5             :  *
       6             :  * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *    src/backend/commands/indexcmds.c
      12             :  *
      13             :  *-------------------------------------------------------------------------
      14             :  */
      15             : 
      16             : #include "postgres.h"
      17             : 
      18             : #include "access/amapi.h"
      19             : #include "access/gist.h"
      20             : #include "access/heapam.h"
      21             : #include "access/htup_details.h"
      22             : #include "access/reloptions.h"
      23             : #include "access/sysattr.h"
      24             : #include "access/tableam.h"
      25             : #include "access/xact.h"
      26             : #include "catalog/catalog.h"
      27             : #include "catalog/index.h"
      28             : #include "catalog/indexing.h"
      29             : #include "catalog/namespace.h"
      30             : #include "catalog/pg_am.h"
      31             : #include "catalog/pg_authid.h"
      32             : #include "catalog/pg_constraint.h"
      33             : #include "catalog/pg_database.h"
      34             : #include "catalog/pg_inherits.h"
      35             : #include "catalog/pg_namespace.h"
      36             : #include "catalog/pg_opclass.h"
      37             : #include "catalog/pg_opfamily.h"
      38             : #include "catalog/pg_tablespace.h"
      39             : #include "catalog/pg_type.h"
      40             : #include "commands/comment.h"
      41             : #include "commands/dbcommands.h"
      42             : #include "commands/defrem.h"
      43             : #include "commands/event_trigger.h"
      44             : #include "commands/progress.h"
      45             : #include "commands/tablecmds.h"
      46             : #include "commands/tablespace.h"
      47             : #include "mb/pg_wchar.h"
      48             : #include "miscadmin.h"
      49             : #include "nodes/makefuncs.h"
      50             : #include "nodes/nodeFuncs.h"
      51             : #include "optimizer/optimizer.h"
      52             : #include "parser/parse_coerce.h"
      53             : #include "parser/parse_oper.h"
      54             : #include "partitioning/partdesc.h"
      55             : #include "pgstat.h"
      56             : #include "rewrite/rewriteManip.h"
      57             : #include "storage/lmgr.h"
      58             : #include "storage/proc.h"
      59             : #include "storage/procarray.h"
      60             : #include "storage/sinvaladt.h"
      61             : #include "utils/acl.h"
      62             : #include "utils/builtins.h"
      63             : #include "utils/fmgroids.h"
      64             : #include "utils/guc.h"
      65             : #include "utils/inval.h"
      66             : #include "utils/lsyscache.h"
      67             : #include "utils/memutils.h"
      68             : #include "utils/partcache.h"
      69             : #include "utils/pg_rusage.h"
      70             : #include "utils/regproc.h"
      71             : #include "utils/snapmgr.h"
      72             : #include "utils/syscache.h"
      73             : 
      74             : 
      75             : /* non-export function prototypes */
      76             : static bool CompareOpclassOptions(const Datum *opts1, const Datum *opts2, int natts);
      77             : static void CheckPredicate(Expr *predicate);
      78             : static void ComputeIndexAttrs(IndexInfo *indexInfo,
      79             :                               Oid *typeOids,
      80             :                               Oid *collationOids,
      81             :                               Oid *opclassOids,
      82             :                               Datum *opclassOptions,
      83             :                               int16 *colOptions,
      84             :                               const List *attList,
      85             :                               const List *exclusionOpNames,
      86             :                               Oid relId,
      87             :                               const char *accessMethodName,
      88             :                               Oid accessMethodId,
      89             :                               bool amcanorder,
      90             :                               bool isconstraint,
      91             :                               bool iswithoutoverlaps,
      92             :                               Oid ddl_userid,
      93             :                               int ddl_sec_context,
      94             :                               int *ddl_save_nestlevel);
      95             : static char *ChooseIndexName(const char *tabname, Oid namespaceId,
      96             :                              const List *colnames, const List *exclusionOpNames,
      97             :                              bool primary, bool isconstraint);
      98             : static char *ChooseIndexNameAddition(const List *colnames);
      99             : static List *ChooseIndexColumnNames(const List *indexElems);
     100             : static void ReindexIndex(const ReindexStmt *stmt, const ReindexParams *params,
     101             :                          bool isTopLevel);
     102             : static void RangeVarCallbackForReindexIndex(const RangeVar *relation,
     103             :                                             Oid relId, Oid oldRelId, void *arg);
     104             : static Oid  ReindexTable(const ReindexStmt *stmt, const ReindexParams *params,
     105             :                          bool isTopLevel);
     106             : static void ReindexMultipleTables(const ReindexStmt *stmt,
     107             :                                   const ReindexParams *params);
     108             : static void reindex_error_callback(void *arg);
     109             : static void ReindexPartitions(const ReindexStmt *stmt, Oid relid,
     110             :                               const ReindexParams *params, bool isTopLevel);
     111             : static void ReindexMultipleInternal(const ReindexStmt *stmt, const List *relids,
     112             :                                     const ReindexParams *params);
     113             : static bool ReindexRelationConcurrently(const ReindexStmt *stmt,
     114             :                                         Oid relationOid,
     115             :                                         const ReindexParams *params);
     116             : static void update_relispartition(Oid relationId, bool newval);
     117             : static inline void set_indexsafe_procflags(void);
     118             : 
     119             : /*
     120             :  * callback argument type for RangeVarCallbackForReindexIndex()
     121             :  */
     122             : struct ReindexIndexCallbackState
     123             : {
     124             :     ReindexParams params;       /* options from statement */
     125             :     Oid         locked_table_oid;   /* tracks previously locked table */
     126             : };
     127             : 
     128             : /*
     129             :  * callback arguments for reindex_error_callback()
     130             :  */
     131             : typedef struct ReindexErrorInfo
     132             : {
     133             :     char       *relname;
     134             :     char       *relnamespace;
     135             :     char        relkind;
     136             : } ReindexErrorInfo;
     137             : 
     138             : /*
     139             :  * CheckIndexCompatible
     140             :  *      Determine whether an existing index definition is compatible with a
     141             :  *      prospective index definition, such that the existing index storage
     142             :  *      could become the storage of the new index, avoiding a rebuild.
     143             :  *
     144             :  * 'oldId': the OID of the existing index
     145             :  * 'accessMethodName': name of the AM to use.
     146             :  * 'attributeList': a list of IndexElem specifying columns and expressions
     147             :  *      to index on.
     148             :  * 'exclusionOpNames': list of names of exclusion-constraint operators,
     149             :  *      or NIL if not an exclusion constraint.
     150             :  * 'isWithoutOverlaps': true iff this index has a WITHOUT OVERLAPS clause.
     151             :  *
     152             :  * This is tailored to the needs of ALTER TABLE ALTER TYPE, which recreates
     153             :  * any indexes that depended on a changing column from their pg_get_indexdef
     154             :  * or pg_get_constraintdef definitions.  We omit some of the sanity checks of
     155             :  * DefineIndex.  We assume that the old and new indexes have the same number
     156             :  * of columns and that if one has an expression column or predicate, both do.
     157             :  * Errors arising from the attribute list still apply.
     158             :  *
     159             :  * Most column type changes that can skip a table rewrite do not invalidate
     160             :  * indexes.  We acknowledge this when all operator classes, collations and
     161             :  * exclusion operators match.  Though we could further permit intra-opfamily
     162             :  * changes for btree and hash indexes, that adds subtle complexity with no
     163             :  * concrete benefit for core types. Note, that INCLUDE columns aren't
     164             :  * checked by this function, for them it's enough that table rewrite is
     165             :  * skipped.
     166             :  *
     167             :  * When a comparison or exclusion operator has a polymorphic input type, the
     168             :  * actual input types must also match.  This defends against the possibility
     169             :  * that operators could vary behavior in response to get_fn_expr_argtype().
     170             :  * At present, this hazard is theoretical: check_exclusion_constraint() and
     171             :  * all core index access methods decline to set fn_expr for such calls.
     172             :  *
     173             :  * We do not yet implement a test to verify compatibility of expression
     174             :  * columns or predicates, so assume any such index is incompatible.
     175             :  */
     176             : bool
     177         102 : CheckIndexCompatible(Oid oldId,
     178             :                      const char *accessMethodName,
     179             :                      const List *attributeList,
     180             :                      const List *exclusionOpNames,
     181             :                      bool isWithoutOverlaps)
     182             : {
     183             :     bool        isconstraint;
     184             :     Oid        *typeIds;
     185             :     Oid        *collationIds;
     186             :     Oid        *opclassIds;
     187             :     Datum      *opclassOptions;
     188             :     Oid         accessMethodId;
     189             :     Oid         relationId;
     190             :     HeapTuple   tuple;
     191             :     Form_pg_index indexForm;
     192             :     Form_pg_am  accessMethodForm;
     193             :     IndexAmRoutine *amRoutine;
     194             :     bool        amcanorder;
     195             :     bool        amsummarizing;
     196             :     int16      *coloptions;
     197             :     IndexInfo  *indexInfo;
     198             :     int         numberOfAttributes;
     199             :     int         old_natts;
     200         102 :     bool        ret = true;
     201             :     oidvector  *old_indclass;
     202             :     oidvector  *old_indcollation;
     203             :     Relation    irel;
     204             :     int         i;
     205             :     Datum       d;
     206             : 
     207             :     /* Caller should already have the relation locked in some way. */
     208         102 :     relationId = IndexGetRelation(oldId, false);
     209             : 
     210             :     /*
     211             :      * We can pretend isconstraint = false unconditionally.  It only serves to
     212             :      * decide the text of an error message that should never happen for us.
     213             :      */
     214         102 :     isconstraint = false;
     215             : 
     216         102 :     numberOfAttributes = list_length(attributeList);
     217             :     Assert(numberOfAttributes > 0);
     218             :     Assert(numberOfAttributes <= INDEX_MAX_KEYS);
     219             : 
     220             :     /* look up the access method */
     221         102 :     tuple = SearchSysCache1(AMNAME, PointerGetDatum(accessMethodName));
     222         102 :     if (!HeapTupleIsValid(tuple))
     223           0 :         ereport(ERROR,
     224             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
     225             :                  errmsg("access method \"%s\" does not exist",
     226             :                         accessMethodName)));
     227         102 :     accessMethodForm = (Form_pg_am) GETSTRUCT(tuple);
     228         102 :     accessMethodId = accessMethodForm->oid;
     229         102 :     amRoutine = GetIndexAmRoutine(accessMethodForm->amhandler);
     230         102 :     ReleaseSysCache(tuple);
     231             : 
     232         102 :     amcanorder = amRoutine->amcanorder;
     233         102 :     amsummarizing = amRoutine->amsummarizing;
     234             : 
     235             :     /*
     236             :      * Compute the operator classes, collations, and exclusion operators for
     237             :      * the new index, so we can test whether it's compatible with the existing
     238             :      * one.  Note that ComputeIndexAttrs might fail here, but that's OK:
     239             :      * DefineIndex would have failed later.  Our attributeList contains only
     240             :      * key attributes, thus we're filling ii_NumIndexAttrs and
     241             :      * ii_NumIndexKeyAttrs with same value.
     242             :      */
     243         102 :     indexInfo = makeIndexInfo(numberOfAttributes, numberOfAttributes,
     244             :                               accessMethodId, NIL, NIL, false, false,
     245             :                               false, false, amsummarizing);
     246         102 :     typeIds = palloc_array(Oid, numberOfAttributes);
     247         102 :     collationIds = palloc_array(Oid, numberOfAttributes);
     248         102 :     opclassIds = palloc_array(Oid, numberOfAttributes);
     249         102 :     opclassOptions = palloc_array(Datum, numberOfAttributes);
     250         102 :     coloptions = palloc_array(int16, numberOfAttributes);
     251         102 :     ComputeIndexAttrs(indexInfo,
     252             :                       typeIds, collationIds, opclassIds, opclassOptions,
     253             :                       coloptions, attributeList,
     254             :                       exclusionOpNames, relationId,
     255             :                       accessMethodName, accessMethodId,
     256             :                       amcanorder, isconstraint, isWithoutOverlaps, InvalidOid,
     257             :                       0, NULL);
     258             : 
     259             :     /* Get the soon-obsolete pg_index tuple. */
     260         102 :     tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(oldId));
     261         102 :     if (!HeapTupleIsValid(tuple))
     262           0 :         elog(ERROR, "cache lookup failed for index %u", oldId);
     263         102 :     indexForm = (Form_pg_index) GETSTRUCT(tuple);
     264             : 
     265             :     /*
     266             :      * We don't assess expressions or predicates; assume incompatibility.
     267             :      * Also, if the index is invalid for any reason, treat it as incompatible.
     268             :      */
     269         204 :     if (!(heap_attisnull(tuple, Anum_pg_index_indpred, NULL) &&
     270         102 :           heap_attisnull(tuple, Anum_pg_index_indexprs, NULL) &&
     271         102 :           indexForm->indisvalid))
     272             :     {
     273           0 :         ReleaseSysCache(tuple);
     274           0 :         return false;
     275             :     }
     276             : 
     277             :     /* Any change in operator class or collation breaks compatibility. */
     278         102 :     old_natts = indexForm->indnkeyatts;
     279             :     Assert(old_natts == numberOfAttributes);
     280             : 
     281         102 :     d = SysCacheGetAttrNotNull(INDEXRELID, tuple, Anum_pg_index_indcollation);
     282         102 :     old_indcollation = (oidvector *) DatumGetPointer(d);
     283             : 
     284         102 :     d = SysCacheGetAttrNotNull(INDEXRELID, tuple, Anum_pg_index_indclass);
     285         102 :     old_indclass = (oidvector *) DatumGetPointer(d);
     286             : 
     287         204 :     ret = (memcmp(old_indclass->values, opclassIds, old_natts * sizeof(Oid)) == 0 &&
     288         102 :            memcmp(old_indcollation->values, collationIds, old_natts * sizeof(Oid)) == 0);
     289             : 
     290         102 :     ReleaseSysCache(tuple);
     291             : 
     292         102 :     if (!ret)
     293           0 :         return false;
     294             : 
     295             :     /* For polymorphic opcintype, column type changes break compatibility. */
     296         102 :     irel = index_open(oldId, AccessShareLock);  /* caller probably has a lock */
     297         210 :     for (i = 0; i < old_natts; i++)
     298             :     {
     299         108 :         if (IsPolymorphicType(get_opclass_input_type(opclassIds[i])) &&
     300           0 :             TupleDescAttr(irel->rd_att, i)->atttypid != typeIds[i])
     301             :         {
     302           0 :             ret = false;
     303           0 :             break;
     304             :         }
     305             :     }
     306             : 
     307             :     /* Any change in opclass options break compatibility. */
     308         102 :     if (ret)
     309             :     {
     310         102 :         Datum      *oldOpclassOptions = palloc_array(Datum, old_natts);
     311             : 
     312         210 :         for (i = 0; i < old_natts; i++)
     313         108 :             oldOpclassOptions[i] = get_attoptions(oldId, i + 1);
     314             : 
     315         102 :         ret = CompareOpclassOptions(oldOpclassOptions, opclassOptions, old_natts);
     316             : 
     317         102 :         pfree(oldOpclassOptions);
     318             :     }
     319             : 
     320             :     /* Any change in exclusion operator selections breaks compatibility. */
     321         102 :     if (ret && indexInfo->ii_ExclusionOps != NULL)
     322             :     {
     323             :         Oid        *old_operators,
     324             :                    *old_procs;
     325             :         uint16     *old_strats;
     326             : 
     327           0 :         RelationGetExclusionInfo(irel, &old_operators, &old_procs, &old_strats);
     328           0 :         ret = memcmp(old_operators, indexInfo->ii_ExclusionOps,
     329             :                      old_natts * sizeof(Oid)) == 0;
     330             : 
     331             :         /* Require an exact input type match for polymorphic operators. */
     332           0 :         if (ret)
     333             :         {
     334           0 :             for (i = 0; i < old_natts && ret; i++)
     335             :             {
     336             :                 Oid         left,
     337             :                             right;
     338             : 
     339           0 :                 op_input_types(indexInfo->ii_ExclusionOps[i], &left, &right);
     340           0 :                 if ((IsPolymorphicType(left) || IsPolymorphicType(right)) &&
     341           0 :                     TupleDescAttr(irel->rd_att, i)->atttypid != typeIds[i])
     342             :                 {
     343           0 :                     ret = false;
     344           0 :                     break;
     345             :                 }
     346             :             }
     347             :         }
     348             :     }
     349             : 
     350         102 :     index_close(irel, NoLock);
     351         102 :     return ret;
     352             : }
     353             : 
     354             : /*
     355             :  * CompareOpclassOptions
     356             :  *
     357             :  * Compare per-column opclass options which are represented by arrays of text[]
     358             :  * datums.  Both elements of arrays and array themselves can be NULL.
     359             :  */
     360             : static bool
     361         102 : CompareOpclassOptions(const Datum *opts1, const Datum *opts2, int natts)
     362             : {
     363             :     int         i;
     364             : 
     365         102 :     if (!opts1 && !opts2)
     366           0 :         return true;
     367             : 
     368         210 :     for (i = 0; i < natts; i++)
     369             :     {
     370         108 :         Datum       opt1 = opts1 ? opts1[i] : (Datum) 0;
     371         108 :         Datum       opt2 = opts2 ? opts2[i] : (Datum) 0;
     372             : 
     373         108 :         if (opt1 == (Datum) 0)
     374             :         {
     375         108 :             if (opt2 == (Datum) 0)
     376         108 :                 continue;
     377             :             else
     378           0 :                 return false;
     379             :         }
     380           0 :         else if (opt2 == (Datum) 0)
     381           0 :             return false;
     382             : 
     383             :         /* Compare non-NULL text[] datums. */
     384           0 :         if (!DatumGetBool(DirectFunctionCall2(array_eq, opt1, opt2)))
     385           0 :             return false;
     386             :     }
     387             : 
     388         102 :     return true;
     389             : }
     390             : 
     391             : /*
     392             :  * WaitForOlderSnapshots
     393             :  *
     394             :  * Wait for transactions that might have an older snapshot than the given xmin
     395             :  * limit, because it might not contain tuples deleted just before it has
     396             :  * been taken. Obtain a list of VXIDs of such transactions, and wait for them
     397             :  * individually. This is used when building an index concurrently.
     398             :  *
     399             :  * We can exclude any running transactions that have xmin > the xmin given;
     400             :  * their oldest snapshot must be newer than our xmin limit.
     401             :  * We can also exclude any transactions that have xmin = zero, since they
     402             :  * evidently have no live snapshot at all (and any one they might be in
     403             :  * process of taking is certainly newer than ours).  Transactions in other
     404             :  * DBs can be ignored too, since they'll never even be able to see the
     405             :  * index being worked on.
     406             :  *
     407             :  * We can also exclude autovacuum processes and processes running manual
     408             :  * lazy VACUUMs, because they won't be fazed by missing index entries
     409             :  * either.  (Manual ANALYZEs, however, can't be excluded because they
     410             :  * might be within transactions that are going to do arbitrary operations
     411             :  * later.)  Processes running CREATE INDEX CONCURRENTLY or REINDEX CONCURRENTLY
     412             :  * on indexes that are neither expressional nor partial are also safe to
     413             :  * ignore, since we know that those processes won't examine any data
     414             :  * outside the table they're indexing.
     415             :  *
     416             :  * Also, GetCurrentVirtualXIDs never reports our own vxid, so we need not
     417             :  * check for that.
     418             :  *
     419             :  * If a process goes idle-in-transaction with xmin zero, we do not need to
     420             :  * wait for it anymore, per the above argument.  We do not have the
     421             :  * infrastructure right now to stop waiting if that happens, but we can at
     422             :  * least avoid the folly of waiting when it is idle at the time we would
     423             :  * begin to wait.  We do this by repeatedly rechecking the output of
     424             :  * GetCurrentVirtualXIDs.  If, during any iteration, a particular vxid
     425             :  * doesn't show up in the output, we know we can forget about it.
     426             :  */
     427             : void
     428         620 : WaitForOlderSnapshots(TransactionId limitXmin, bool progress)
     429             : {
     430             :     int         n_old_snapshots;
     431             :     int         i;
     432             :     VirtualTransactionId *old_snapshots;
     433             : 
     434         620 :     old_snapshots = GetCurrentVirtualXIDs(limitXmin, true, false,
     435             :                                           PROC_IS_AUTOVACUUM | PROC_IN_VACUUM
     436             :                                           | PROC_IN_SAFE_IC,
     437             :                                           &n_old_snapshots);
     438         620 :     if (progress)
     439         606 :         pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, n_old_snapshots);
     440             : 
     441         900 :     for (i = 0; i < n_old_snapshots; i++)
     442             :     {
     443         280 :         if (!VirtualTransactionIdIsValid(old_snapshots[i]))
     444          56 :             continue;           /* found uninteresting in previous cycle */
     445             : 
     446         224 :         if (i > 0)
     447             :         {
     448             :             /* see if anything's changed ... */
     449             :             VirtualTransactionId *newer_snapshots;
     450             :             int         n_newer_snapshots;
     451             :             int         j;
     452             :             int         k;
     453             : 
     454          68 :             newer_snapshots = GetCurrentVirtualXIDs(limitXmin,
     455             :                                                     true, false,
     456             :                                                     PROC_IS_AUTOVACUUM | PROC_IN_VACUUM
     457             :                                                     | PROC_IN_SAFE_IC,
     458             :                                                     &n_newer_snapshots);
     459         238 :             for (j = i; j < n_old_snapshots; j++)
     460             :             {
     461         170 :                 if (!VirtualTransactionIdIsValid(old_snapshots[j]))
     462          18 :                     continue;   /* found uninteresting in previous cycle */
     463         524 :                 for (k = 0; k < n_newer_snapshots; k++)
     464             :                 {
     465         442 :                     if (VirtualTransactionIdEquals(old_snapshots[j],
     466             :                                                    newer_snapshots[k]))
     467          70 :                         break;
     468             :                 }
     469         152 :                 if (k >= n_newer_snapshots) /* not there anymore */
     470          82 :                     SetInvalidVirtualTransactionId(old_snapshots[j]);
     471             :             }
     472          68 :             pfree(newer_snapshots);
     473             :         }
     474             : 
     475         224 :         if (VirtualTransactionIdIsValid(old_snapshots[i]))
     476             :         {
     477             :             /* If requested, publish who we're going to wait for. */
     478         198 :             if (progress)
     479             :             {
     480         198 :                 PGPROC     *holder = ProcNumberGetProc(old_snapshots[i].procNumber);
     481             : 
     482         198 :                 if (holder)
     483         198 :                     pgstat_progress_update_param(PROGRESS_WAITFOR_CURRENT_PID,
     484         198 :                                                  holder->pid);
     485             :             }
     486         198 :             VirtualXactLock(old_snapshots[i], true);
     487             :         }
     488             : 
     489         224 :         if (progress)
     490         224 :             pgstat_progress_update_param(PROGRESS_WAITFOR_DONE, i + 1);
     491             :     }
     492         620 : }
     493             : 
     494             : 
     495             : /*
     496             :  * DefineIndex
     497             :  *      Creates a new index.
     498             :  *
     499             :  * This function manages the current userid according to the needs of pg_dump.
     500             :  * Recreating old-database catalog entries in new-database is fine, regardless
     501             :  * of which users would have permission to recreate those entries now.  That's
     502             :  * just preservation of state.  Running opaque expressions, like calling a
     503             :  * function named in a catalog entry or evaluating a pg_node_tree in a catalog
     504             :  * entry, as anyone other than the object owner, is not fine.  To adhere to
     505             :  * those principles and to remain fail-safe, use the table owner userid for
     506             :  * most ACL checks.  Use the original userid for ACL checks reached without
     507             :  * traversing opaque expressions.  (pg_dump can predict such ACL checks from
     508             :  * catalogs.)  Overall, this is a mess.  Future DDL development should
     509             :  * consider offering one DDL command for catalog setup and a separate DDL
     510             :  * command for steps that run opaque expressions.
     511             :  *
     512             :  * 'tableId': the OID of the table relation on which the index is to be
     513             :  *      created
     514             :  * 'stmt': IndexStmt describing the properties of the new index.
     515             :  * 'indexRelationId': normally InvalidOid, but during bootstrap can be
     516             :  *      nonzero to specify a preselected OID for the index.
     517             :  * 'parentIndexId': the OID of the parent index; InvalidOid if not the child
     518             :  *      of a partitioned index.
     519             :  * 'parentConstraintId': the OID of the parent constraint; InvalidOid if not
     520             :  *      the child of a constraint (only used when recursing)
     521             :  * 'total_parts': total number of direct and indirect partitions of relation;
     522             :  *      pass -1 if not known or rel is not partitioned.
     523             :  * 'is_alter_table': this is due to an ALTER rather than a CREATE operation.
     524             :  * 'check_rights': check for CREATE rights in namespace and tablespace.  (This
     525             :  *      should be true except when ALTER is deleting/recreating an index.)
     526             :  * 'check_not_in_use': check for table not already in use in current session.
     527             :  *      This should be true unless caller is holding the table open, in which
     528             :  *      case the caller had better have checked it earlier.
     529             :  * 'skip_build': make the catalog entries but don't create the index files
     530             :  * 'quiet': suppress the NOTICE chatter ordinarily provided for constraints.
     531             :  *
     532             :  * Returns the object address of the created index.
     533             :  */
     534             : ObjectAddress
     535       27198 : DefineIndex(Oid tableId,
     536             :             IndexStmt *stmt,
     537             :             Oid indexRelationId,
     538             :             Oid parentIndexId,
     539             :             Oid parentConstraintId,
     540             :             int total_parts,
     541             :             bool is_alter_table,
     542             :             bool check_rights,
     543             :             bool check_not_in_use,
     544             :             bool skip_build,
     545             :             bool quiet)
     546             : {
     547             :     bool        concurrent;
     548             :     char       *indexRelationName;
     549             :     char       *accessMethodName;
     550             :     Oid        *typeIds;
     551             :     Oid        *collationIds;
     552             :     Oid        *opclassIds;
     553             :     Datum      *opclassOptions;
     554             :     Oid         accessMethodId;
     555             :     Oid         namespaceId;
     556             :     Oid         tablespaceId;
     557       27198 :     Oid         createdConstraintId = InvalidOid;
     558             :     List       *indexColNames;
     559             :     List       *allIndexParams;
     560             :     Relation    rel;
     561             :     HeapTuple   tuple;
     562             :     Form_pg_am  accessMethodForm;
     563             :     IndexAmRoutine *amRoutine;
     564             :     bool        amcanorder;
     565             :     bool        amissummarizing;
     566             :     amoptions_function amoptions;
     567             :     bool        exclusion;
     568             :     bool        partitioned;
     569             :     bool        safe_index;
     570             :     Datum       reloptions;
     571             :     int16      *coloptions;
     572             :     IndexInfo  *indexInfo;
     573             :     bits16      flags;
     574             :     bits16      constr_flags;
     575             :     int         numberOfAttributes;
     576             :     int         numberOfKeyAttributes;
     577             :     TransactionId limitXmin;
     578             :     ObjectAddress address;
     579             :     LockRelId   heaprelid;
     580             :     LOCKTAG     heaplocktag;
     581             :     LOCKMODE    lockmode;
     582             :     Snapshot    snapshot;
     583             :     Oid         root_save_userid;
     584             :     int         root_save_sec_context;
     585             :     int         root_save_nestlevel;
     586             : 
     587       27198 :     root_save_nestlevel = NewGUCNestLevel();
     588             : 
     589       27198 :     RestrictSearchPath();
     590             : 
     591             :     /*
     592             :      * Some callers need us to run with an empty default_tablespace; this is a
     593             :      * necessary hack to be able to reproduce catalog state accurately when
     594             :      * recreating indexes after table-rewriting ALTER TABLE.
     595             :      */
     596       27198 :     if (stmt->reset_default_tblspc)
     597         512 :         (void) set_config_option("default_tablespace", "",
     598             :                                  PGC_USERSET, PGC_S_SESSION,
     599             :                                  GUC_ACTION_SAVE, true, 0, false);
     600             : 
     601             :     /*
     602             :      * Force non-concurrent build on temporary relations, even if CONCURRENTLY
     603             :      * was requested.  Other backends can't access a temporary relation, so
     604             :      * there's no harm in grabbing a stronger lock, and a non-concurrent DROP
     605             :      * is more efficient.  Do this before any use of the concurrent option is
     606             :      * done.
     607             :      */
     608       27198 :     if (stmt->concurrent && get_rel_persistence(tableId) != RELPERSISTENCE_TEMP)
     609         168 :         concurrent = true;
     610             :     else
     611       27030 :         concurrent = false;
     612             : 
     613             :     /*
     614             :      * Start progress report.  If we're building a partition, this was already
     615             :      * done.
     616             :      */
     617       27198 :     if (!OidIsValid(parentIndexId))
     618             :     {
     619       24420 :         pgstat_progress_start_command(PROGRESS_COMMAND_CREATE_INDEX, tableId);
     620       24420 :         pgstat_progress_update_param(PROGRESS_CREATEIDX_COMMAND,
     621             :                                      concurrent ?
     622             :                                      PROGRESS_CREATEIDX_COMMAND_CREATE_CONCURRENTLY :
     623             :                                      PROGRESS_CREATEIDX_COMMAND_CREATE);
     624             :     }
     625             : 
     626             :     /*
     627             :      * No index OID to report yet
     628             :      */
     629       27198 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_INDEX_OID,
     630             :                                  InvalidOid);
     631             : 
     632             :     /*
     633             :      * count key attributes in index
     634             :      */
     635       27198 :     numberOfKeyAttributes = list_length(stmt->indexParams);
     636             : 
     637             :     /*
     638             :      * Calculate the new list of index columns including both key columns and
     639             :      * INCLUDE columns.  Later we can determine which of these are key
     640             :      * columns, and which are just part of the INCLUDE list by checking the
     641             :      * list position.  A list item in a position less than ii_NumIndexKeyAttrs
     642             :      * is part of the key columns, and anything equal to and over is part of
     643             :      * the INCLUDE columns.
     644             :      */
     645       27198 :     allIndexParams = list_concat_copy(stmt->indexParams,
     646       27198 :                                       stmt->indexIncludingParams);
     647       27198 :     numberOfAttributes = list_length(allIndexParams);
     648             : 
     649       27198 :     if (numberOfKeyAttributes <= 0)
     650           0 :         ereport(ERROR,
     651             :                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
     652             :                  errmsg("must specify at least one column")));
     653       27198 :     if (numberOfAttributes > INDEX_MAX_KEYS)
     654           0 :         ereport(ERROR,
     655             :                 (errcode(ERRCODE_TOO_MANY_COLUMNS),
     656             :                  errmsg("cannot use more than %d columns in an index",
     657             :                         INDEX_MAX_KEYS)));
     658             : 
     659             :     /*
     660             :      * Only SELECT ... FOR UPDATE/SHARE are allowed while doing a standard
     661             :      * index build; but for concurrent builds we allow INSERT/UPDATE/DELETE
     662             :      * (but not VACUUM).
     663             :      *
     664             :      * NB: Caller is responsible for making sure that tableId refers to the
     665             :      * relation on which the index should be built; except in bootstrap mode,
     666             :      * this will typically require the caller to have already locked the
     667             :      * relation.  To avoid lock upgrade hazards, that lock should be at least
     668             :      * as strong as the one we take here.
     669             :      *
     670             :      * NB: If the lock strength here ever changes, code that is run by
     671             :      * parallel workers under the control of certain particular ambuild
     672             :      * functions will need to be updated, too.
     673             :      */
     674       27198 :     lockmode = concurrent ? ShareUpdateExclusiveLock : ShareLock;
     675       27198 :     rel = table_open(tableId, lockmode);
     676             : 
     677             :     /*
     678             :      * Switch to the table owner's userid, so that any index functions are run
     679             :      * as that user.  Also lock down security-restricted operations.  We
     680             :      * already arranged to make GUC variable changes local to this command.
     681             :      */
     682       27198 :     GetUserIdAndSecContext(&root_save_userid, &root_save_sec_context);
     683       27198 :     SetUserIdAndSecContext(rel->rd_rel->relowner,
     684             :                            root_save_sec_context | SECURITY_RESTRICTED_OPERATION);
     685             : 
     686       27198 :     namespaceId = RelationGetNamespace(rel);
     687             : 
     688             :     /*
     689             :      * It has exclusion constraint behavior if it's an EXCLUDE constraint or a
     690             :      * temporal PRIMARY KEY/UNIQUE constraint
     691             :      */
     692       27198 :     exclusion = stmt->excludeOpNames || stmt->iswithoutoverlaps;
     693             : 
     694             :     /* Ensure that it makes sense to index this kind of relation */
     695       27198 :     switch (rel->rd_rel->relkind)
     696             :     {
     697       27192 :         case RELKIND_RELATION:
     698             :         case RELKIND_MATVIEW:
     699             :         case RELKIND_PARTITIONED_TABLE:
     700             :             /* OK */
     701       27192 :             break;
     702           6 :         default:
     703           6 :             ereport(ERROR,
     704             :                     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
     705             :                      errmsg("cannot create index on relation \"%s\"",
     706             :                             RelationGetRelationName(rel)),
     707             :                      errdetail_relkind_not_supported(rel->rd_rel->relkind)));
     708             :             break;
     709             :     }
     710             : 
     711             :     /*
     712             :      * Establish behavior for partitioned tables, and verify sanity of
     713             :      * parameters.
     714             :      *
     715             :      * We do not build an actual index in this case; we only create a few
     716             :      * catalog entries.  The actual indexes are built by recursing for each
     717             :      * partition.
     718             :      */
     719       27192 :     partitioned = rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE;
     720       27192 :     if (partitioned)
     721             :     {
     722             :         /*
     723             :          * Note: we check 'stmt->concurrent' rather than 'concurrent', so that
     724             :          * the error is thrown also for temporary tables.  Seems better to be
     725             :          * consistent, even though we could do it on temporary table because
     726             :          * we're not actually doing it concurrently.
     727             :          */
     728        2026 :         if (stmt->concurrent)
     729           6 :             ereport(ERROR,
     730             :                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     731             :                      errmsg("cannot create index on partitioned table \"%s\" concurrently",
     732             :                             RelationGetRelationName(rel))));
     733             :     }
     734             : 
     735             :     /*
     736             :      * Don't try to CREATE INDEX on temp tables of other backends.
     737             :      */
     738       27186 :     if (RELATION_IS_OTHER_TEMP(rel))
     739           0 :         ereport(ERROR,
     740             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     741             :                  errmsg("cannot create indexes on temporary tables of other sessions")));
     742             : 
     743             :     /*
     744             :      * Unless our caller vouches for having checked this already, insist that
     745             :      * the table not be in use by our own session, either.  Otherwise we might
     746             :      * fail to make entries in the new index (for instance, if an INSERT or
     747             :      * UPDATE is in progress and has already made its list of target indexes).
     748             :      */
     749       27186 :     if (check_not_in_use)
     750       13780 :         CheckTableNotInUse(rel, "CREATE INDEX");
     751             : 
     752             :     /*
     753             :      * Verify we (still) have CREATE rights in the rel's namespace.
     754             :      * (Presumably we did when the rel was created, but maybe not anymore.)
     755             :      * Skip check if caller doesn't want it.  Also skip check if
     756             :      * bootstrapping, since permissions machinery may not be working yet.
     757             :      */
     758       27180 :     if (check_rights && !IsBootstrapProcessingMode())
     759             :     {
     760             :         AclResult   aclresult;
     761             : 
     762       14848 :         aclresult = object_aclcheck(NamespaceRelationId, namespaceId, root_save_userid,
     763             :                                     ACL_CREATE);
     764       14848 :         if (aclresult != ACLCHECK_OK)
     765           0 :             aclcheck_error(aclresult, OBJECT_SCHEMA,
     766           0 :                            get_namespace_name(namespaceId));
     767             :     }
     768             : 
     769             :     /*
     770             :      * Select tablespace to use.  If not specified, use default tablespace
     771             :      * (which may in turn default to database's default).
     772             :      */
     773       27180 :     if (stmt->tableSpace)
     774             :     {
     775         200 :         tablespaceId = get_tablespace_oid(stmt->tableSpace, false);
     776         200 :         if (partitioned && tablespaceId == MyDatabaseTableSpace)
     777           6 :             ereport(ERROR,
     778             :                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     779             :                      errmsg("cannot specify default tablespace for partitioned relations")));
     780             :     }
     781             :     else
     782             :     {
     783       26980 :         tablespaceId = GetDefaultTablespace(rel->rd_rel->relpersistence,
     784             :                                             partitioned);
     785             :         /* note InvalidOid is OK in this case */
     786             :     }
     787             : 
     788             :     /* Check tablespace permissions */
     789       27168 :     if (check_rights &&
     790         104 :         OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
     791             :     {
     792             :         AclResult   aclresult;
     793             : 
     794         104 :         aclresult = object_aclcheck(TableSpaceRelationId, tablespaceId, root_save_userid,
     795             :                                     ACL_CREATE);
     796         104 :         if (aclresult != ACLCHECK_OK)
     797           0 :             aclcheck_error(aclresult, OBJECT_TABLESPACE,
     798           0 :                            get_tablespace_name(tablespaceId));
     799             :     }
     800             : 
     801             :     /*
     802             :      * Force shared indexes into the pg_global tablespace.  This is a bit of a
     803             :      * hack but seems simpler than marking them in the BKI commands.  On the
     804             :      * other hand, if it's not shared, don't allow it to be placed there.
     805             :      */
     806       27168 :     if (rel->rd_rel->relisshared)
     807        1680 :         tablespaceId = GLOBALTABLESPACE_OID;
     808       25488 :     else if (tablespaceId == GLOBALTABLESPACE_OID)
     809           0 :         ereport(ERROR,
     810             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     811             :                  errmsg("only shared relations can be placed in pg_global tablespace")));
     812             : 
     813             :     /*
     814             :      * Choose the index column names.
     815             :      */
     816       27168 :     indexColNames = ChooseIndexColumnNames(allIndexParams);
     817             : 
     818             :     /*
     819             :      * Select name for index if caller didn't specify
     820             :      */
     821       27168 :     indexRelationName = stmt->idxname;
     822       27168 :     if (indexRelationName == NULL)
     823       11074 :         indexRelationName = ChooseIndexName(RelationGetRelationName(rel),
     824             :                                             namespaceId,
     825             :                                             indexColNames,
     826       11074 :                                             stmt->excludeOpNames,
     827       11074 :                                             stmt->primary,
     828       11074 :                                             stmt->isconstraint);
     829             : 
     830             :     /*
     831             :      * look up the access method, verify it can handle the requested features
     832             :      */
     833       27168 :     accessMethodName = stmt->accessMethod;
     834       27168 :     tuple = SearchSysCache1(AMNAME, PointerGetDatum(accessMethodName));
     835       27168 :     if (!HeapTupleIsValid(tuple))
     836             :     {
     837             :         /*
     838             :          * Hack to provide more-or-less-transparent updating of old RTREE
     839             :          * indexes to GiST: if RTREE is requested and not found, use GIST.
     840             :          */
     841           6 :         if (strcmp(accessMethodName, "rtree") == 0)
     842             :         {
     843           6 :             ereport(NOTICE,
     844             :                     (errmsg("substituting access method \"gist\" for obsolete method \"rtree\"")));
     845           6 :             accessMethodName = "gist";
     846           6 :             tuple = SearchSysCache1(AMNAME, PointerGetDatum(accessMethodName));
     847             :         }
     848             : 
     849           6 :         if (!HeapTupleIsValid(tuple))
     850           0 :             ereport(ERROR,
     851             :                     (errcode(ERRCODE_UNDEFINED_OBJECT),
     852             :                      errmsg("access method \"%s\" does not exist",
     853             :                             accessMethodName)));
     854             :     }
     855       27168 :     accessMethodForm = (Form_pg_am) GETSTRUCT(tuple);
     856       27168 :     accessMethodId = accessMethodForm->oid;
     857       27168 :     amRoutine = GetIndexAmRoutine(accessMethodForm->amhandler);
     858             : 
     859       27168 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_ACCESS_METHOD_OID,
     860             :                                  accessMethodId);
     861             : 
     862       27168 :     if (stmt->unique && !stmt->iswithoutoverlaps && !amRoutine->amcanunique)
     863           0 :         ereport(ERROR,
     864             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     865             :                  errmsg("access method \"%s\" does not support unique indexes",
     866             :                         accessMethodName)));
     867       27168 :     if (stmt->indexIncludingParams != NIL && !amRoutine->amcaninclude)
     868          18 :         ereport(ERROR,
     869             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     870             :                  errmsg("access method \"%s\" does not support included columns",
     871             :                         accessMethodName)));
     872       27150 :     if (numberOfKeyAttributes > 1 && !amRoutine->amcanmulticol)
     873           0 :         ereport(ERROR,
     874             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     875             :                  errmsg("access method \"%s\" does not support multicolumn indexes",
     876             :                         accessMethodName)));
     877       27150 :     if (exclusion && amRoutine->amgettuple == NULL)
     878           0 :         ereport(ERROR,
     879             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     880             :                  errmsg("access method \"%s\" does not support exclusion constraints",
     881             :                         accessMethodName)));
     882             : 
     883       27150 :     amcanorder = amRoutine->amcanorder;
     884       27150 :     amoptions = amRoutine->amoptions;
     885       27150 :     amissummarizing = amRoutine->amsummarizing;
     886             : 
     887       27150 :     pfree(amRoutine);
     888       27150 :     ReleaseSysCache(tuple);
     889             : 
     890             :     /*
     891             :      * Validate predicate, if given
     892             :      */
     893       27150 :     if (stmt->whereClause)
     894         394 :         CheckPredicate((Expr *) stmt->whereClause);
     895             : 
     896             :     /*
     897             :      * Parse AM-specific options, convert to text array form, validate.
     898             :      */
     899       27150 :     reloptions = transformRelOptions((Datum) 0, stmt->options,
     900             :                                      NULL, NULL, false, false);
     901             : 
     902       27144 :     (void) index_reloptions(amoptions, reloptions, true);
     903             : 
     904             :     /*
     905             :      * Prepare arguments for index_create, primarily an IndexInfo structure.
     906             :      * Note that predicates must be in implicit-AND format.  In a concurrent
     907             :      * build, mark it not-ready-for-inserts.
     908             :      */
     909       27080 :     indexInfo = makeIndexInfo(numberOfAttributes,
     910             :                               numberOfKeyAttributes,
     911             :                               accessMethodId,
     912             :                               NIL,  /* expressions, NIL for now */
     913       27080 :                               make_ands_implicit((Expr *) stmt->whereClause),
     914       27080 :                               stmt->unique,
     915       27080 :                               stmt->nulls_not_distinct,
     916       27080 :                               !concurrent,
     917             :                               concurrent,
     918       27080 :                               amissummarizing);
     919             : 
     920       27080 :     typeIds = palloc_array(Oid, numberOfAttributes);
     921       27080 :     collationIds = palloc_array(Oid, numberOfAttributes);
     922       27080 :     opclassIds = palloc_array(Oid, numberOfAttributes);
     923       27080 :     opclassOptions = palloc_array(Datum, numberOfAttributes);
     924       27080 :     coloptions = palloc_array(int16, numberOfAttributes);
     925       27080 :     ComputeIndexAttrs(indexInfo,
     926             :                       typeIds, collationIds, opclassIds, opclassOptions,
     927             :                       coloptions, allIndexParams,
     928       27080 :                       stmt->excludeOpNames, tableId,
     929             :                       accessMethodName, accessMethodId,
     930       27080 :                       amcanorder, stmt->isconstraint, stmt->iswithoutoverlaps,
     931             :                       root_save_userid, root_save_sec_context,
     932             :                       &root_save_nestlevel);
     933             : 
     934             :     /*
     935             :      * Extra checks when creating a PRIMARY KEY index.
     936             :      */
     937       26850 :     if (stmt->primary)
     938        8338 :         index_check_primary_key(rel, indexInfo, is_alter_table, stmt);
     939             : 
     940             :     /*
     941             :      * If this table is partitioned and we're creating a unique index, primary
     942             :      * key, or exclusion constraint, make sure that the partition key is a
     943             :      * subset of the index's columns.  Otherwise it would be possible to
     944             :      * violate uniqueness by putting values that ought to be unique in
     945             :      * different partitions.
     946             :      *
     947             :      * We could lift this limitation if we had global indexes, but those have
     948             :      * their own problems, so this is a useful feature combination.
     949             :      */
     950       26802 :     if (partitioned && (stmt->unique || exclusion))
     951             :     {
     952        1150 :         PartitionKey key = RelationGetPartitionKey(rel);
     953             :         const char *constraint_type;
     954             :         int         i;
     955             : 
     956        1150 :         if (stmt->primary)
     957         848 :             constraint_type = "PRIMARY KEY";
     958         302 :         else if (stmt->unique)
     959         202 :             constraint_type = "UNIQUE";
     960         100 :         else if (stmt->excludeOpNames)
     961         100 :             constraint_type = "EXCLUDE";
     962             :         else
     963             :         {
     964           0 :             elog(ERROR, "unknown constraint type");
     965             :             constraint_type = NULL; /* keep compiler quiet */
     966             :         }
     967             : 
     968             :         /*
     969             :          * Verify that all the columns in the partition key appear in the
     970             :          * unique key definition, with the same notion of equality.
     971             :          */
     972        2298 :         for (i = 0; i < key->partnatts; i++)
     973             :         {
     974        1252 :             bool        found = false;
     975             :             int         eq_strategy;
     976             :             Oid         ptkey_eqop;
     977             :             int         j;
     978             : 
     979             :             /*
     980             :              * Identify the equality operator associated with this partkey
     981             :              * column.  For list and range partitioning, partkeys use btree
     982             :              * operator classes; hash partitioning uses hash operator classes.
     983             :              * (Keep this in sync with ComputePartitionAttrs!)
     984             :              */
     985        1252 :             if (key->strategy == PARTITION_STRATEGY_HASH)
     986          46 :                 eq_strategy = HTEqualStrategyNumber;
     987             :             else
     988        1206 :                 eq_strategy = BTEqualStrategyNumber;
     989             : 
     990        1252 :             ptkey_eqop = get_opfamily_member(key->partopfamily[i],
     991        1252 :                                              key->partopcintype[i],
     992        1252 :                                              key->partopcintype[i],
     993             :                                              eq_strategy);
     994        1252 :             if (!OidIsValid(ptkey_eqop))
     995           0 :                 elog(ERROR, "missing operator %d(%u,%u) in partition opfamily %u",
     996             :                      eq_strategy, key->partopcintype[i], key->partopcintype[i],
     997             :                      key->partopfamily[i]);
     998             : 
     999             :             /*
    1000             :              * We'll need to be able to identify the equality operators
    1001             :              * associated with index columns, too.  We know what to do with
    1002             :              * btree opclasses; if there are ever any other index types that
    1003             :              * support unique indexes, this logic will need extension. But if
    1004             :              * we have an exclusion constraint (or a temporal PK), it already
    1005             :              * knows the operators, so we don't have to infer them.
    1006             :              */
    1007        1252 :             if (stmt->unique && !stmt->iswithoutoverlaps && accessMethodId != BTREE_AM_OID)
    1008           0 :                 ereport(ERROR,
    1009             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1010             :                          errmsg("cannot match partition key to an index using access method \"%s\"",
    1011             :                                 accessMethodName)));
    1012             : 
    1013             :             /*
    1014             :              * It may be possible to support UNIQUE constraints when partition
    1015             :              * keys are expressions, but is it worth it?  Give up for now.
    1016             :              */
    1017        1252 :             if (key->partattrs[i] == 0)
    1018          12 :                 ereport(ERROR,
    1019             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1020             :                          errmsg("unsupported %s constraint with partition key definition",
    1021             :                                 constraint_type),
    1022             :                          errdetail("%s constraints cannot be used when partition keys include expressions.",
    1023             :                                    constraint_type)));
    1024             : 
    1025             :             /* Search the index column(s) for a match */
    1026        1410 :             for (j = 0; j < indexInfo->ii_NumIndexKeyAttrs; j++)
    1027             :             {
    1028        1332 :                 if (key->partattrs[i] == indexInfo->ii_IndexAttrNumbers[j])
    1029             :                 {
    1030             :                     /*
    1031             :                      * Matched the column, now what about the collation and
    1032             :                      * equality op?
    1033             :                      */
    1034             :                     Oid         idx_opfamily;
    1035             :                     Oid         idx_opcintype;
    1036             : 
    1037        1162 :                     if (key->partcollation[i] != collationIds[j])
    1038           0 :                         continue;
    1039             : 
    1040        1162 :                     if (get_opclass_opfamily_and_input_type(opclassIds[j],
    1041             :                                                             &idx_opfamily,
    1042             :                                                             &idx_opcintype))
    1043             :                     {
    1044        1162 :                         Oid         idx_eqop = InvalidOid;
    1045             : 
    1046        1162 :                         if (stmt->unique && !stmt->iswithoutoverlaps)
    1047        1026 :                             idx_eqop = get_opfamily_member(idx_opfamily,
    1048             :                                                            idx_opcintype,
    1049             :                                                            idx_opcintype,
    1050             :                                                            BTEqualStrategyNumber);
    1051         136 :                         else if (exclusion)
    1052         136 :                             idx_eqop = indexInfo->ii_ExclusionOps[j];
    1053             :                         Assert(idx_eqop);
    1054             : 
    1055        1162 :                         if (ptkey_eqop == idx_eqop)
    1056             :                         {
    1057        1148 :                             found = true;
    1058        1148 :                             break;
    1059             :                         }
    1060          14 :                         else if (exclusion)
    1061             :                         {
    1062             :                             /*
    1063             :                              * We found a match, but it's not an equality
    1064             :                              * operator. Instead of failing below with an
    1065             :                              * error message about a missing column, fail now
    1066             :                              * and explain that the operator is wrong.
    1067             :                              */
    1068          14 :                             Form_pg_attribute att = TupleDescAttr(RelationGetDescr(rel), key->partattrs[i] - 1);
    1069             : 
    1070          14 :                             ereport(ERROR,
    1071             :                                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1072             :                                      errmsg("cannot match partition key to index on column \"%s\" using non-equal operator \"%s\"",
    1073             :                                             NameStr(att->attname),
    1074             :                                             get_opname(indexInfo->ii_ExclusionOps[j]))));
    1075             :                         }
    1076             :                     }
    1077             :                 }
    1078             :             }
    1079             : 
    1080        1226 :             if (!found)
    1081             :             {
    1082             :                 Form_pg_attribute att;
    1083             : 
    1084          78 :                 att = TupleDescAttr(RelationGetDescr(rel),
    1085             :                                     key->partattrs[i] - 1);
    1086          78 :                 ereport(ERROR,
    1087             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1088             :                          errmsg("unique constraint on partitioned table must include all partitioning columns"),
    1089             :                          errdetail("%s constraint on table \"%s\" lacks column \"%s\" which is part of the partition key.",
    1090             :                                    constraint_type, RelationGetRelationName(rel),
    1091             :                                    NameStr(att->attname))));
    1092             :             }
    1093             :         }
    1094             :     }
    1095             : 
    1096             : 
    1097             :     /*
    1098             :      * We disallow indexes on system columns.  They would not necessarily get
    1099             :      * updated correctly, and they don't seem useful anyway.
    1100             :      */
    1101       63892 :     for (int i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
    1102             :     {
    1103       37194 :         AttrNumber  attno = indexInfo->ii_IndexAttrNumbers[i];
    1104             : 
    1105       37194 :         if (attno < 0)
    1106           0 :             ereport(ERROR,
    1107             :                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1108             :                      errmsg("index creation on system columns is not supported")));
    1109             :     }
    1110             : 
    1111             :     /*
    1112             :      * Also check for system columns used in expressions or predicates.
    1113             :      */
    1114       26698 :     if (indexInfo->ii_Expressions || indexInfo->ii_Predicate)
    1115             :     {
    1116        1106 :         Bitmapset  *indexattrs = NULL;
    1117             : 
    1118        1106 :         pull_varattnos((Node *) indexInfo->ii_Expressions, 1, &indexattrs);
    1119        1106 :         pull_varattnos((Node *) indexInfo->ii_Predicate, 1, &indexattrs);
    1120             : 
    1121        7730 :         for (int i = FirstLowInvalidHeapAttributeNumber + 1; i < 0; i++)
    1122             :         {
    1123        6636 :             if (bms_is_member(i - FirstLowInvalidHeapAttributeNumber,
    1124             :                               indexattrs))
    1125          12 :                 ereport(ERROR,
    1126             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1127             :                          errmsg("index creation on system columns is not supported")));
    1128             :         }
    1129             :     }
    1130             : 
    1131             :     /* Is index safe for others to ignore?  See set_indexsafe_procflags() */
    1132       52566 :     safe_index = indexInfo->ii_Expressions == NIL &&
    1133       25880 :         indexInfo->ii_Predicate == NIL;
    1134             : 
    1135             :     /*
    1136             :      * Report index creation if appropriate (delay this till after most of the
    1137             :      * error checks)
    1138             :      */
    1139       26686 :     if (stmt->isconstraint && !quiet)
    1140             :     {
    1141             :         const char *constraint_type;
    1142             : 
    1143        9128 :         if (stmt->primary)
    1144        8134 :             constraint_type = "PRIMARY KEY";
    1145         994 :         else if (stmt->unique)
    1146         818 :             constraint_type = "UNIQUE";
    1147         176 :         else if (stmt->excludeOpNames)
    1148         176 :             constraint_type = "EXCLUDE";
    1149             :         else
    1150             :         {
    1151           0 :             elog(ERROR, "unknown constraint type");
    1152             :             constraint_type = NULL; /* keep compiler quiet */
    1153             :         }
    1154             : 
    1155        9128 :         ereport(DEBUG1,
    1156             :                 (errmsg_internal("%s %s will create implicit index \"%s\" for table \"%s\"",
    1157             :                                  is_alter_table ? "ALTER TABLE / ADD" : "CREATE TABLE /",
    1158             :                                  constraint_type,
    1159             :                                  indexRelationName, RelationGetRelationName(rel))));
    1160             :     }
    1161             : 
    1162             :     /*
    1163             :      * A valid stmt->oldNumber implies that we already have a built form of
    1164             :      * the index.  The caller should also decline any index build.
    1165             :      */
    1166             :     Assert(!RelFileNumberIsValid(stmt->oldNumber) || (skip_build && !concurrent));
    1167             : 
    1168             :     /*
    1169             :      * Make the catalog entries for the index, including constraints. This
    1170             :      * step also actually builds the index, except if caller requested not to
    1171             :      * or in concurrent mode, in which case it'll be done later, or doing a
    1172             :      * partitioned index (because those don't have storage).
    1173             :      */
    1174       26686 :     flags = constr_flags = 0;
    1175       26686 :     if (stmt->isconstraint)
    1176        9362 :         flags |= INDEX_CREATE_ADD_CONSTRAINT;
    1177       26686 :     if (skip_build || concurrent || partitioned)
    1178       12414 :         flags |= INDEX_CREATE_SKIP_BUILD;
    1179       26686 :     if (stmt->if_not_exists)
    1180          18 :         flags |= INDEX_CREATE_IF_NOT_EXISTS;
    1181       26686 :     if (concurrent)
    1182         162 :         flags |= INDEX_CREATE_CONCURRENT;
    1183       26686 :     if (partitioned)
    1184        1904 :         flags |= INDEX_CREATE_PARTITIONED;
    1185       26686 :     if (stmt->primary)
    1186        8260 :         flags |= INDEX_CREATE_IS_PRIMARY;
    1187             : 
    1188             :     /*
    1189             :      * If the table is partitioned, and recursion was declined but partitions
    1190             :      * exist, mark the index as invalid.
    1191             :      */
    1192       26686 :     if (partitioned && stmt->relation && !stmt->relation->inh)
    1193             :     {
    1194         228 :         PartitionDesc pd = RelationGetPartitionDesc(rel, true);
    1195             : 
    1196         228 :         if (pd->nparts != 0)
    1197         214 :             flags |= INDEX_CREATE_INVALID;
    1198             :     }
    1199             : 
    1200       26686 :     if (stmt->deferrable)
    1201         138 :         constr_flags |= INDEX_CONSTR_CREATE_DEFERRABLE;
    1202       26686 :     if (stmt->initdeferred)
    1203          38 :         constr_flags |= INDEX_CONSTR_CREATE_INIT_DEFERRED;
    1204       26686 :     if (stmt->iswithoutoverlaps)
    1205         404 :         constr_flags |= INDEX_CONSTR_CREATE_WITHOUT_OVERLAPS;
    1206             : 
    1207             :     indexRelationId =
    1208       26686 :         index_create(rel, indexRelationName, indexRelationId, parentIndexId,
    1209             :                      parentConstraintId,
    1210             :                      stmt->oldNumber, indexInfo, indexColNames,
    1211             :                      accessMethodId, tablespaceId,
    1212             :                      collationIds, opclassIds, opclassOptions,
    1213             :                      coloptions, NULL, reloptions,
    1214             :                      flags, constr_flags,
    1215       26686 :                      allowSystemTableMods, !check_rights,
    1216       26686 :                      &createdConstraintId);
    1217             : 
    1218       26502 :     ObjectAddressSet(address, RelationRelationId, indexRelationId);
    1219             : 
    1220       26502 :     if (!OidIsValid(indexRelationId))
    1221             :     {
    1222             :         /*
    1223             :          * Roll back any GUC changes executed by index functions.  Also revert
    1224             :          * to original default_tablespace if we changed it above.
    1225             :          */
    1226          18 :         AtEOXact_GUC(false, root_save_nestlevel);
    1227             : 
    1228             :         /* Restore userid and security context */
    1229          18 :         SetUserIdAndSecContext(root_save_userid, root_save_sec_context);
    1230             : 
    1231          18 :         table_close(rel, NoLock);
    1232             : 
    1233             :         /* If this is the top-level index, we're done */
    1234          18 :         if (!OidIsValid(parentIndexId))
    1235          18 :             pgstat_progress_end_command();
    1236             : 
    1237          18 :         return address;
    1238             :     }
    1239             : 
    1240             :     /*
    1241             :      * Roll back any GUC changes executed by index functions, and keep
    1242             :      * subsequent changes local to this command.  This is essential if some
    1243             :      * index function changed a behavior-affecting GUC, e.g. search_path.
    1244             :      */
    1245       26484 :     AtEOXact_GUC(false, root_save_nestlevel);
    1246       26484 :     root_save_nestlevel = NewGUCNestLevel();
    1247             : 
    1248             :     /* Add any requested comment */
    1249       26484 :     if (stmt->idxcomment != NULL)
    1250          90 :         CreateComments(indexRelationId, RelationRelationId, 0,
    1251          90 :                        stmt->idxcomment);
    1252             : 
    1253       26484 :     if (partitioned)
    1254             :     {
    1255             :         PartitionDesc partdesc;
    1256             : 
    1257             :         /*
    1258             :          * Unless caller specified to skip this step (via ONLY), process each
    1259             :          * partition to make sure they all contain a corresponding index.
    1260             :          *
    1261             :          * If we're called internally (no stmt->relation), recurse always.
    1262             :          */
    1263        1904 :         partdesc = RelationGetPartitionDesc(rel, true);
    1264        1904 :         if ((!stmt->relation || stmt->relation->inh) && partdesc->nparts > 0)
    1265             :         {
    1266         562 :             int         nparts = partdesc->nparts;
    1267         562 :             Oid        *part_oids = palloc_array(Oid, nparts);
    1268         562 :             bool        invalidate_parent = false;
    1269             :             Relation    parentIndex;
    1270             :             TupleDesc   parentDesc;
    1271             : 
    1272             :             /*
    1273             :              * Report the total number of partitions at the start of the
    1274             :              * command; don't update it when being called recursively.
    1275             :              */
    1276         562 :             if (!OidIsValid(parentIndexId))
    1277             :             {
    1278             :                 /*
    1279             :                  * When called by ProcessUtilitySlow, the number of partitions
    1280             :                  * is passed in as an optimization; but other callers pass -1
    1281             :                  * since they don't have the value handy.  This should count
    1282             :                  * partitions the same way, ie one less than the number of
    1283             :                  * relations find_all_inheritors reports.
    1284             :                  *
    1285             :                  * We assume we needn't ask find_all_inheritors to take locks,
    1286             :                  * because that should have happened already for all callers.
    1287             :                  * Even if it did not, this is safe as long as we don't try to
    1288             :                  * touch the partitions here; the worst consequence would be a
    1289             :                  * bogus progress-reporting total.
    1290             :                  */
    1291         450 :                 if (total_parts < 0)
    1292             :                 {
    1293         122 :                     List       *children = find_all_inheritors(tableId, NoLock, NULL);
    1294             : 
    1295         122 :                     total_parts = list_length(children) - 1;
    1296         122 :                     list_free(children);
    1297             :                 }
    1298             : 
    1299         450 :                 pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_TOTAL,
    1300             :                                              total_parts);
    1301             :             }
    1302             : 
    1303             :             /* Make a local copy of partdesc->oids[], just for safety */
    1304         562 :             memcpy(part_oids, partdesc->oids, sizeof(Oid) * nparts);
    1305             : 
    1306             :             /*
    1307             :              * We'll need an IndexInfo describing the parent index.  The one
    1308             :              * built above is almost good enough, but not quite, because (for
    1309             :              * example) its predicate expression if any hasn't been through
    1310             :              * expression preprocessing.  The most reliable way to get an
    1311             :              * IndexInfo that will match those for child indexes is to build
    1312             :              * it the same way, using BuildIndexInfo().
    1313             :              */
    1314         562 :             parentIndex = index_open(indexRelationId, lockmode);
    1315         562 :             indexInfo = BuildIndexInfo(parentIndex);
    1316             : 
    1317         562 :             parentDesc = RelationGetDescr(rel);
    1318             : 
    1319             :             /*
    1320             :              * For each partition, scan all existing indexes; if one matches
    1321             :              * our index definition and is not already attached to some other
    1322             :              * parent index, attach it to the one we just created.
    1323             :              *
    1324             :              * If none matches, build a new index by calling ourselves
    1325             :              * recursively with the same options (except for the index name).
    1326             :              */
    1327        1560 :             for (int i = 0; i < nparts; i++)
    1328             :             {
    1329        1022 :                 Oid         childRelid = part_oids[i];
    1330             :                 Relation    childrel;
    1331             :                 Oid         child_save_userid;
    1332             :                 int         child_save_sec_context;
    1333             :                 int         child_save_nestlevel;
    1334             :                 List       *childidxs;
    1335             :                 ListCell   *cell;
    1336             :                 AttrMap    *attmap;
    1337        1022 :                 bool        found = false;
    1338             : 
    1339        1022 :                 childrel = table_open(childRelid, lockmode);
    1340             : 
    1341        1022 :                 GetUserIdAndSecContext(&child_save_userid,
    1342             :                                        &child_save_sec_context);
    1343        1022 :                 SetUserIdAndSecContext(childrel->rd_rel->relowner,
    1344             :                                        child_save_sec_context | SECURITY_RESTRICTED_OPERATION);
    1345        1022 :                 child_save_nestlevel = NewGUCNestLevel();
    1346        1022 :                 RestrictSearchPath();
    1347             : 
    1348             :                 /*
    1349             :                  * Don't try to create indexes on foreign tables, though. Skip
    1350             :                  * those if a regular index, or fail if trying to create a
    1351             :                  * constraint index.
    1352             :                  */
    1353        1022 :                 if (childrel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
    1354             :                 {
    1355          18 :                     if (stmt->unique || stmt->primary)
    1356          12 :                         ereport(ERROR,
    1357             :                                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
    1358             :                                  errmsg("cannot create unique index on partitioned table \"%s\"",
    1359             :                                         RelationGetRelationName(rel)),
    1360             :                                  errdetail("Table \"%s\" contains partitions that are foreign tables.",
    1361             :                                            RelationGetRelationName(rel))));
    1362             : 
    1363           6 :                     AtEOXact_GUC(false, child_save_nestlevel);
    1364           6 :                     SetUserIdAndSecContext(child_save_userid,
    1365             :                                            child_save_sec_context);
    1366           6 :                     table_close(childrel, lockmode);
    1367           6 :                     continue;
    1368             :                 }
    1369             : 
    1370        1004 :                 childidxs = RelationGetIndexList(childrel);
    1371             :                 attmap =
    1372        1004 :                     build_attrmap_by_name(RelationGetDescr(childrel),
    1373             :                                           parentDesc,
    1374             :                                           false);
    1375             : 
    1376        1334 :                 foreach(cell, childidxs)
    1377             :                 {
    1378         390 :                     Oid         cldidxid = lfirst_oid(cell);
    1379             :                     Relation    cldidx;
    1380             :                     IndexInfo  *cldIdxInfo;
    1381             : 
    1382             :                     /* this index is already partition of another one */
    1383         390 :                     if (has_superclass(cldidxid))
    1384         306 :                         continue;
    1385             : 
    1386          84 :                     cldidx = index_open(cldidxid, lockmode);
    1387          84 :                     cldIdxInfo = BuildIndexInfo(cldidx);
    1388          84 :                     if (CompareIndexInfo(cldIdxInfo, indexInfo,
    1389          84 :                                          cldidx->rd_indcollation,
    1390          84 :                                          parentIndex->rd_indcollation,
    1391          84 :                                          cldidx->rd_opfamily,
    1392          84 :                                          parentIndex->rd_opfamily,
    1393             :                                          attmap))
    1394             :                     {
    1395          60 :                         Oid         cldConstrOid = InvalidOid;
    1396             : 
    1397             :                         /*
    1398             :                          * Found a match.
    1399             :                          *
    1400             :                          * If this index is being created in the parent
    1401             :                          * because of a constraint, then the child needs to
    1402             :                          * have a constraint also, so look for one.  If there
    1403             :                          * is no such constraint, this index is no good, so
    1404             :                          * keep looking.
    1405             :                          */
    1406          60 :                         if (createdConstraintId != InvalidOid)
    1407             :                         {
    1408             :                             cldConstrOid =
    1409          12 :                                 get_relation_idx_constraint_oid(childRelid,
    1410             :                                                                 cldidxid);
    1411          12 :                             if (cldConstrOid == InvalidOid)
    1412             :                             {
    1413           0 :                                 index_close(cldidx, lockmode);
    1414           0 :                                 continue;
    1415             :                             }
    1416             :                         }
    1417             : 
    1418             :                         /* Attach index to parent and we're done. */
    1419          60 :                         IndexSetParentIndex(cldidx, indexRelationId);
    1420          60 :                         if (createdConstraintId != InvalidOid)
    1421          12 :                             ConstraintSetParentConstraint(cldConstrOid,
    1422             :                                                           createdConstraintId,
    1423             :                                                           childRelid);
    1424             : 
    1425          60 :                         if (!cldidx->rd_index->indisvalid)
    1426          18 :                             invalidate_parent = true;
    1427             : 
    1428          60 :                         found = true;
    1429             : 
    1430             :                         /*
    1431             :                          * Report this partition as processed.  Note that if
    1432             :                          * the partition has children itself, we'd ideally
    1433             :                          * count the children and update the progress report
    1434             :                          * for all of them; but that seems unduly expensive.
    1435             :                          * Instead, the progress report will act like all such
    1436             :                          * indirect children were processed in zero time at
    1437             :                          * the end of the command.
    1438             :                          */
    1439          60 :                         pgstat_progress_incr_param(PROGRESS_CREATEIDX_PARTITIONS_DONE, 1);
    1440             : 
    1441             :                         /* keep lock till commit */
    1442          60 :                         index_close(cldidx, NoLock);
    1443          60 :                         break;
    1444             :                     }
    1445             : 
    1446          24 :                     index_close(cldidx, lockmode);
    1447             :                 }
    1448             : 
    1449        1004 :                 list_free(childidxs);
    1450        1004 :                 AtEOXact_GUC(false, child_save_nestlevel);
    1451        1004 :                 SetUserIdAndSecContext(child_save_userid,
    1452             :                                        child_save_sec_context);
    1453        1004 :                 table_close(childrel, NoLock);
    1454             : 
    1455             :                 /*
    1456             :                  * If no matching index was found, create our own.
    1457             :                  */
    1458        1004 :                 if (!found)
    1459             :                 {
    1460         944 :                     IndexStmt  *childStmt = copyObject(stmt);
    1461             :                     bool        found_whole_row;
    1462             :                     ListCell   *lc;
    1463             :                     ObjectAddress childAddr;
    1464             : 
    1465             :                     /*
    1466             :                      * We can't use the same index name for the child index,
    1467             :                      * so clear idxname to let the recursive invocation choose
    1468             :                      * a new name.  Likewise, the existing target relation
    1469             :                      * field is wrong, and if indexOid or oldNumber are set,
    1470             :                      * they mustn't be applied to the child either.
    1471             :                      */
    1472         944 :                     childStmt->idxname = NULL;
    1473         944 :                     childStmt->relation = NULL;
    1474         944 :                     childStmt->indexOid = InvalidOid;
    1475         944 :                     childStmt->oldNumber = InvalidRelFileNumber;
    1476         944 :                     childStmt->oldCreateSubid = InvalidSubTransactionId;
    1477         944 :                     childStmt->oldFirstRelfilelocatorSubid = InvalidSubTransactionId;
    1478             : 
    1479             :                     /*
    1480             :                      * Adjust any Vars (both in expressions and in the index's
    1481             :                      * WHERE clause) to match the partition's column numbering
    1482             :                      * in case it's different from the parent's.
    1483             :                      */
    1484        2088 :                     foreach(lc, childStmt->indexParams)
    1485             :                     {
    1486        1144 :                         IndexElem  *ielem = lfirst(lc);
    1487             : 
    1488             :                         /*
    1489             :                          * If the index parameter is an expression, we must
    1490             :                          * translate it to contain child Vars.
    1491             :                          */
    1492        1144 :                         if (ielem->expr)
    1493             :                         {
    1494          84 :                             ielem->expr =
    1495          84 :                                 map_variable_attnos((Node *) ielem->expr,
    1496             :                                                     1, 0, attmap,
    1497             :                                                     InvalidOid,
    1498             :                                                     &found_whole_row);
    1499          84 :                             if (found_whole_row)
    1500           0 :                                 elog(ERROR, "cannot convert whole-row table reference");
    1501             :                         }
    1502             :                     }
    1503         944 :                     childStmt->whereClause =
    1504         944 :                         map_variable_attnos(stmt->whereClause, 1, 0,
    1505             :                                             attmap,
    1506             :                                             InvalidOid, &found_whole_row);
    1507         944 :                     if (found_whole_row)
    1508           0 :                         elog(ERROR, "cannot convert whole-row table reference");
    1509             : 
    1510             :                     /*
    1511             :                      * Recurse as the starting user ID.  Callee will use that
    1512             :                      * for permission checks, then switch again.
    1513             :                      */
    1514             :                     Assert(GetUserId() == child_save_userid);
    1515         944 :                     SetUserIdAndSecContext(root_save_userid,
    1516             :                                            root_save_sec_context);
    1517             :                     childAddr =
    1518         944 :                         DefineIndex(childRelid, childStmt,
    1519             :                                     InvalidOid, /* no predefined OID */
    1520             :                                     indexRelationId,    /* this is our child */
    1521             :                                     createdConstraintId,
    1522             :                                     -1,
    1523             :                                     is_alter_table, check_rights,
    1524             :                                     check_not_in_use,
    1525             :                                     skip_build, quiet);
    1526         932 :                     SetUserIdAndSecContext(child_save_userid,
    1527             :                                            child_save_sec_context);
    1528             : 
    1529             :                     /*
    1530             :                      * Check if the index just created is valid or not, as it
    1531             :                      * could be possible that it has been switched as invalid
    1532             :                      * when recursing across multiple partition levels.
    1533             :                      */
    1534         932 :                     if (!get_index_isvalid(childAddr.objectId))
    1535           6 :                         invalidate_parent = true;
    1536             :                 }
    1537             : 
    1538         992 :                 free_attrmap(attmap);
    1539             :             }
    1540             : 
    1541         538 :             index_close(parentIndex, lockmode);
    1542             : 
    1543             :             /*
    1544             :              * The pg_index row we inserted for this index was marked
    1545             :              * indisvalid=true.  But if we attached an existing index that is
    1546             :              * invalid, this is incorrect, so update our row to invalid too.
    1547             :              */
    1548         538 :             if (invalidate_parent)
    1549             :             {
    1550          24 :                 Relation    pg_index = table_open(IndexRelationId, RowExclusiveLock);
    1551             :                 HeapTuple   tup,
    1552             :                             newtup;
    1553             : 
    1554          24 :                 tup = SearchSysCache1(INDEXRELID,
    1555             :                                       ObjectIdGetDatum(indexRelationId));
    1556          24 :                 if (!HeapTupleIsValid(tup))
    1557           0 :                     elog(ERROR, "cache lookup failed for index %u",
    1558             :                          indexRelationId);
    1559          24 :                 newtup = heap_copytuple(tup);
    1560          24 :                 ((Form_pg_index) GETSTRUCT(newtup))->indisvalid = false;
    1561          24 :                 CatalogTupleUpdate(pg_index, &tup->t_self, newtup);
    1562          24 :                 ReleaseSysCache(tup);
    1563          24 :                 table_close(pg_index, RowExclusiveLock);
    1564          24 :                 heap_freetuple(newtup);
    1565             : 
    1566             :                 /*
    1567             :                  * CCI here to make this update visible, in case this recurses
    1568             :                  * across multiple partition levels.
    1569             :                  */
    1570          24 :                 CommandCounterIncrement();
    1571             :             }
    1572             :         }
    1573             : 
    1574             :         /*
    1575             :          * Indexes on partitioned tables are not themselves built, so we're
    1576             :          * done here.
    1577             :          */
    1578        1880 :         AtEOXact_GUC(false, root_save_nestlevel);
    1579        1880 :         SetUserIdAndSecContext(root_save_userid, root_save_sec_context);
    1580        1880 :         table_close(rel, NoLock);
    1581        1880 :         if (!OidIsValid(parentIndexId))
    1582        1610 :             pgstat_progress_end_command();
    1583             :         else
    1584             :         {
    1585             :             /* Update progress for an intermediate partitioned index itself */
    1586         270 :             pgstat_progress_incr_param(PROGRESS_CREATEIDX_PARTITIONS_DONE, 1);
    1587             :         }
    1588             : 
    1589        1880 :         return address;
    1590             :     }
    1591             : 
    1592       24580 :     AtEOXact_GUC(false, root_save_nestlevel);
    1593       24580 :     SetUserIdAndSecContext(root_save_userid, root_save_sec_context);
    1594             : 
    1595       24580 :     if (!concurrent)
    1596             :     {
    1597             :         /* Close the heap and we're done, in the non-concurrent case */
    1598       24430 :         table_close(rel, NoLock);
    1599             : 
    1600             :         /*
    1601             :          * If this is the top-level index, the command is done overall;
    1602             :          * otherwise, increment progress to report one child index is done.
    1603             :          */
    1604       24430 :         if (!OidIsValid(parentIndexId))
    1605       21964 :             pgstat_progress_end_command();
    1606             :         else
    1607        2466 :             pgstat_progress_incr_param(PROGRESS_CREATEIDX_PARTITIONS_DONE, 1);
    1608             : 
    1609       24430 :         return address;
    1610             :     }
    1611             : 
    1612             :     /* save lockrelid and locktag for below, then close rel */
    1613         150 :     heaprelid = rel->rd_lockInfo.lockRelId;
    1614         150 :     SET_LOCKTAG_RELATION(heaplocktag, heaprelid.dbId, heaprelid.relId);
    1615         150 :     table_close(rel, NoLock);
    1616             : 
    1617             :     /*
    1618             :      * For a concurrent build, it's important to make the catalog entries
    1619             :      * visible to other transactions before we start to build the index. That
    1620             :      * will prevent them from making incompatible HOT updates.  The new index
    1621             :      * will be marked not indisready and not indisvalid, so that no one else
    1622             :      * tries to either insert into it or use it for queries.
    1623             :      *
    1624             :      * We must commit our current transaction so that the index becomes
    1625             :      * visible; then start another.  Note that all the data structures we just
    1626             :      * built are lost in the commit.  The only data we keep past here are the
    1627             :      * relation IDs.
    1628             :      *
    1629             :      * Before committing, get a session-level lock on the table, to ensure
    1630             :      * that neither it nor the index can be dropped before we finish. This
    1631             :      * cannot block, even if someone else is waiting for access, because we
    1632             :      * already have the same lock within our transaction.
    1633             :      *
    1634             :      * Note: we don't currently bother with a session lock on the index,
    1635             :      * because there are no operations that could change its state while we
    1636             :      * hold lock on the parent table.  This might need to change later.
    1637             :      */
    1638         150 :     LockRelationIdForSession(&heaprelid, ShareUpdateExclusiveLock);
    1639             : 
    1640         150 :     PopActiveSnapshot();
    1641         150 :     CommitTransactionCommand();
    1642         150 :     StartTransactionCommand();
    1643             : 
    1644             :     /* Tell concurrent index builds to ignore us, if index qualifies */
    1645         150 :     if (safe_index)
    1646         110 :         set_indexsafe_procflags();
    1647             : 
    1648             :     /*
    1649             :      * The index is now visible, so we can report the OID.  While on it,
    1650             :      * include the report for the beginning of phase 2.
    1651             :      */
    1652             :     {
    1653         150 :         const int   progress_cols[] = {
    1654             :             PROGRESS_CREATEIDX_INDEX_OID,
    1655             :             PROGRESS_CREATEIDX_PHASE
    1656             :         };
    1657         150 :         const int64 progress_vals[] = {
    1658             :             indexRelationId,
    1659             :             PROGRESS_CREATEIDX_PHASE_WAIT_1
    1660             :         };
    1661             : 
    1662         150 :         pgstat_progress_update_multi_param(2, progress_cols, progress_vals);
    1663             :     }
    1664             : 
    1665             :     /*
    1666             :      * Phase 2 of concurrent index build (see comments for validate_index()
    1667             :      * for an overview of how this works)
    1668             :      *
    1669             :      * Now we must wait until no running transaction could have the table open
    1670             :      * with the old list of indexes.  Use ShareLock to consider running
    1671             :      * transactions that hold locks that permit writing to the table.  Note we
    1672             :      * do not need to worry about xacts that open the table for writing after
    1673             :      * this point; they will see the new index when they open it.
    1674             :      *
    1675             :      * Note: the reason we use actual lock acquisition here, rather than just
    1676             :      * checking the ProcArray and sleeping, is that deadlock is possible if
    1677             :      * one of the transactions in question is blocked trying to acquire an
    1678             :      * exclusive lock on our table.  The lock code will detect deadlock and
    1679             :      * error out properly.
    1680             :      */
    1681         150 :     WaitForLockers(heaplocktag, ShareLock, true);
    1682             : 
    1683             :     /*
    1684             :      * At this moment we are sure that there are no transactions with the
    1685             :      * table open for write that don't have this new index in their list of
    1686             :      * indexes.  We have waited out all the existing transactions and any new
    1687             :      * transaction will have the new index in its list, but the index is still
    1688             :      * marked as "not-ready-for-inserts".  The index is consulted while
    1689             :      * deciding HOT-safety though.  This arrangement ensures that no new HOT
    1690             :      * chains can be created where the new tuple and the old tuple in the
    1691             :      * chain have different index keys.
    1692             :      *
    1693             :      * We now take a new snapshot, and build the index using all tuples that
    1694             :      * are visible in this snapshot.  We can be sure that any HOT updates to
    1695             :      * these tuples will be compatible with the index, since any updates made
    1696             :      * by transactions that didn't know about the index are now committed or
    1697             :      * rolled back.  Thus, each visible tuple is either the end of its
    1698             :      * HOT-chain or the extension of the chain is HOT-safe for this index.
    1699             :      */
    1700             : 
    1701             :     /* Set ActiveSnapshot since functions in the indexes may need it */
    1702         150 :     PushActiveSnapshot(GetTransactionSnapshot());
    1703             : 
    1704             :     /* Perform concurrent build of index */
    1705         150 :     index_concurrently_build(tableId, indexRelationId);
    1706             : 
    1707             :     /* we can do away with our snapshot */
    1708         132 :     PopActiveSnapshot();
    1709             : 
    1710             :     /*
    1711             :      * Commit this transaction to make the indisready update visible.
    1712             :      */
    1713         132 :     CommitTransactionCommand();
    1714         132 :     StartTransactionCommand();
    1715             : 
    1716             :     /* Tell concurrent index builds to ignore us, if index qualifies */
    1717         132 :     if (safe_index)
    1718          98 :         set_indexsafe_procflags();
    1719             : 
    1720             :     /*
    1721             :      * Phase 3 of concurrent index build
    1722             :      *
    1723             :      * We once again wait until no transaction can have the table open with
    1724             :      * the index marked as read-only for updates.
    1725             :      */
    1726         132 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    1727             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_2);
    1728         132 :     WaitForLockers(heaplocktag, ShareLock, true);
    1729             : 
    1730             :     /*
    1731             :      * Now take the "reference snapshot" that will be used by validate_index()
    1732             :      * to filter candidate tuples.  Beware!  There might still be snapshots in
    1733             :      * use that treat some transaction as in-progress that our reference
    1734             :      * snapshot treats as committed.  If such a recently-committed transaction
    1735             :      * deleted tuples in the table, we will not include them in the index; yet
    1736             :      * those transactions which see the deleting one as still-in-progress will
    1737             :      * expect such tuples to be there once we mark the index as valid.
    1738             :      *
    1739             :      * We solve this by waiting for all endangered transactions to exit before
    1740             :      * we mark the index as valid.
    1741             :      *
    1742             :      * We also set ActiveSnapshot to this snap, since functions in indexes may
    1743             :      * need a snapshot.
    1744             :      */
    1745         132 :     snapshot = RegisterSnapshot(GetTransactionSnapshot());
    1746         132 :     PushActiveSnapshot(snapshot);
    1747             : 
    1748             :     /*
    1749             :      * Scan the index and the heap, insert any missing index entries.
    1750             :      */
    1751         132 :     validate_index(tableId, indexRelationId, snapshot);
    1752             : 
    1753             :     /*
    1754             :      * Drop the reference snapshot.  We must do this before waiting out other
    1755             :      * snapshot holders, else we will deadlock against other processes also
    1756             :      * doing CREATE INDEX CONCURRENTLY, which would see our snapshot as one
    1757             :      * they must wait for.  But first, save the snapshot's xmin to use as
    1758             :      * limitXmin for GetCurrentVirtualXIDs().
    1759             :      */
    1760         132 :     limitXmin = snapshot->xmin;
    1761             : 
    1762         132 :     PopActiveSnapshot();
    1763         132 :     UnregisterSnapshot(snapshot);
    1764             : 
    1765             :     /*
    1766             :      * The snapshot subsystem could still contain registered snapshots that
    1767             :      * are holding back our process's advertised xmin; in particular, if
    1768             :      * default_transaction_isolation = serializable, there is a transaction
    1769             :      * snapshot that is still active.  The CatalogSnapshot is likewise a
    1770             :      * hazard.  To ensure no deadlocks, we must commit and start yet another
    1771             :      * transaction, and do our wait before any snapshot has been taken in it.
    1772             :      */
    1773         132 :     CommitTransactionCommand();
    1774         132 :     StartTransactionCommand();
    1775             : 
    1776             :     /* Tell concurrent index builds to ignore us, if index qualifies */
    1777         132 :     if (safe_index)
    1778          98 :         set_indexsafe_procflags();
    1779             : 
    1780             :     /* We should now definitely not be advertising any xmin. */
    1781             :     Assert(MyProc->xmin == InvalidTransactionId);
    1782             : 
    1783             :     /*
    1784             :      * The index is now valid in the sense that it contains all currently
    1785             :      * interesting tuples.  But since it might not contain tuples deleted just
    1786             :      * before the reference snap was taken, we have to wait out any
    1787             :      * transactions that might have older snapshots.
    1788             :      */
    1789         132 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    1790             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_3);
    1791         132 :     WaitForOlderSnapshots(limitXmin, true);
    1792             : 
    1793             :     /*
    1794             :      * Index can now be marked valid -- update its pg_index entry
    1795             :      */
    1796         132 :     index_set_state_flags(indexRelationId, INDEX_CREATE_SET_VALID);
    1797             : 
    1798             :     /*
    1799             :      * The pg_index update will cause backends (including this one) to update
    1800             :      * relcache entries for the index itself, but we should also send a
    1801             :      * relcache inval on the parent table to force replanning of cached plans.
    1802             :      * Otherwise existing sessions might fail to use the new index where it
    1803             :      * would be useful.  (Note that our earlier commits did not create reasons
    1804             :      * to replan; so relcache flush on the index itself was sufficient.)
    1805             :      */
    1806         132 :     CacheInvalidateRelcacheByRelid(heaprelid.relId);
    1807             : 
    1808             :     /*
    1809             :      * Last thing to do is release the session-level lock on the parent table.
    1810             :      */
    1811         132 :     UnlockRelationIdForSession(&heaprelid, ShareUpdateExclusiveLock);
    1812             : 
    1813         132 :     pgstat_progress_end_command();
    1814             : 
    1815         132 :     return address;
    1816             : }
    1817             : 
    1818             : 
    1819             : /*
    1820             :  * CheckPredicate
    1821             :  *      Checks that the given partial-index predicate is valid.
    1822             :  *
    1823             :  * This used to also constrain the form of the predicate to forms that
    1824             :  * indxpath.c could do something with.  However, that seems overly
    1825             :  * restrictive.  One useful application of partial indexes is to apply
    1826             :  * a UNIQUE constraint across a subset of a table, and in that scenario
    1827             :  * any evaluable predicate will work.  So accept any predicate here
    1828             :  * (except ones requiring a plan), and let indxpath.c fend for itself.
    1829             :  */
    1830             : static void
    1831         394 : CheckPredicate(Expr *predicate)
    1832             : {
    1833             :     /*
    1834             :      * transformExpr() should have already rejected subqueries, aggregates,
    1835             :      * and window functions, based on the EXPR_KIND_ for a predicate.
    1836             :      */
    1837             : 
    1838             :     /*
    1839             :      * A predicate using mutable functions is probably wrong, for the same
    1840             :      * reasons that we don't allow an index expression to use one.
    1841             :      */
    1842         394 :     if (contain_mutable_functions_after_planning(predicate))
    1843           0 :         ereport(ERROR,
    1844             :                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    1845             :                  errmsg("functions in index predicate must be marked IMMUTABLE")));
    1846         394 : }
    1847             : 
    1848             : /*
    1849             :  * Compute per-index-column information, including indexed column numbers
    1850             :  * or index expressions, opclasses and their options. Note, all output vectors
    1851             :  * should be allocated for all columns, including "including" ones.
    1852             :  *
    1853             :  * If the caller switched to the table owner, ddl_userid is the role for ACL
    1854             :  * checks reached without traversing opaque expressions.  Otherwise, it's
    1855             :  * InvalidOid, and other ddl_* arguments are undefined.
    1856             :  */
    1857             : static void
    1858       27182 : ComputeIndexAttrs(IndexInfo *indexInfo,
    1859             :                   Oid *typeOids,
    1860             :                   Oid *collationOids,
    1861             :                   Oid *opclassOids,
    1862             :                   Datum *opclassOptions,
    1863             :                   int16 *colOptions,
    1864             :                   const List *attList,  /* list of IndexElem's */
    1865             :                   const List *exclusionOpNames,
    1866             :                   Oid relId,
    1867             :                   const char *accessMethodName,
    1868             :                   Oid accessMethodId,
    1869             :                   bool amcanorder,
    1870             :                   bool isconstraint,
    1871             :                   bool iswithoutoverlaps,
    1872             :                   Oid ddl_userid,
    1873             :                   int ddl_sec_context,
    1874             :                   int *ddl_save_nestlevel)
    1875             : {
    1876             :     ListCell   *nextExclOp;
    1877             :     ListCell   *lc;
    1878             :     int         attn;
    1879       27182 :     int         nkeycols = indexInfo->ii_NumIndexKeyAttrs;
    1880             :     Oid         save_userid;
    1881             :     int         save_sec_context;
    1882             : 
    1883             :     /* Allocate space for exclusion operator info, if needed */
    1884       27182 :     if (exclusionOpNames)
    1885             :     {
    1886             :         Assert(list_length(exclusionOpNames) == nkeycols);
    1887         288 :         indexInfo->ii_ExclusionOps = palloc_array(Oid, nkeycols);
    1888         288 :         indexInfo->ii_ExclusionProcs = palloc_array(Oid, nkeycols);
    1889         288 :         indexInfo->ii_ExclusionStrats = palloc_array(uint16, nkeycols);
    1890         288 :         nextExclOp = list_head(exclusionOpNames);
    1891             :     }
    1892             :     else
    1893       26894 :         nextExclOp = NULL;
    1894             : 
    1895             :     /* exclusionOpNames can be non-NIL if we are creating a partition */
    1896       27182 :     if (iswithoutoverlaps && exclusionOpNames == NIL)
    1897             :     {
    1898         368 :         indexInfo->ii_ExclusionOps = palloc_array(Oid, nkeycols);
    1899         368 :         indexInfo->ii_ExclusionProcs = palloc_array(Oid, nkeycols);
    1900         368 :         indexInfo->ii_ExclusionStrats = palloc_array(uint16, nkeycols);
    1901             :     }
    1902             : 
    1903       27182 :     if (OidIsValid(ddl_userid))
    1904       27080 :         GetUserIdAndSecContext(&save_userid, &save_sec_context);
    1905             : 
    1906             :     /*
    1907             :      * process attributeList
    1908             :      */
    1909       27182 :     attn = 0;
    1910       64668 :     foreach(lc, attList)
    1911             :     {
    1912       37716 :         IndexElem  *attribute = (IndexElem *) lfirst(lc);
    1913             :         Oid         atttype;
    1914             :         Oid         attcollation;
    1915             : 
    1916             :         /*
    1917             :          * Process the column-or-expression to be indexed.
    1918             :          */
    1919       37716 :         if (attribute->name != NULL)
    1920             :         {
    1921             :             /* Simple index attribute */
    1922             :             HeapTuple   atttuple;
    1923             :             Form_pg_attribute attform;
    1924             : 
    1925             :             Assert(attribute->expr == NULL);
    1926       36696 :             atttuple = SearchSysCacheAttName(relId, attribute->name);
    1927       36696 :             if (!HeapTupleIsValid(atttuple))
    1928             :             {
    1929             :                 /* difference in error message spellings is historical */
    1930          30 :                 if (isconstraint)
    1931          18 :                     ereport(ERROR,
    1932             :                             (errcode(ERRCODE_UNDEFINED_COLUMN),
    1933             :                              errmsg("column \"%s\" named in key does not exist",
    1934             :                                     attribute->name)));
    1935             :                 else
    1936          12 :                     ereport(ERROR,
    1937             :                             (errcode(ERRCODE_UNDEFINED_COLUMN),
    1938             :                              errmsg("column \"%s\" does not exist",
    1939             :                                     attribute->name)));
    1940             :             }
    1941       36666 :             attform = (Form_pg_attribute) GETSTRUCT(atttuple);
    1942       36666 :             indexInfo->ii_IndexAttrNumbers[attn] = attform->attnum;
    1943       36666 :             atttype = attform->atttypid;
    1944       36666 :             attcollation = attform->attcollation;
    1945       36666 :             ReleaseSysCache(atttuple);
    1946             :         }
    1947             :         else
    1948             :         {
    1949             :             /* Index expression */
    1950        1020 :             Node       *expr = attribute->expr;
    1951             : 
    1952             :             Assert(expr != NULL);
    1953             : 
    1954        1020 :             if (attn >= nkeycols)
    1955           0 :                 ereport(ERROR,
    1956             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1957             :                          errmsg("expressions are not supported in included columns")));
    1958        1020 :             atttype = exprType(expr);
    1959        1020 :             attcollation = exprCollation(expr);
    1960             : 
    1961             :             /*
    1962             :              * Strip any top-level COLLATE clause.  This ensures that we treat
    1963             :              * "x COLLATE y" and "(x COLLATE y)" alike.
    1964             :              */
    1965        1050 :             while (IsA(expr, CollateExpr))
    1966          30 :                 expr = (Node *) ((CollateExpr *) expr)->arg;
    1967             : 
    1968        1020 :             if (IsA(expr, Var) &&
    1969          12 :                 ((Var *) expr)->varattno != InvalidAttrNumber)
    1970             :             {
    1971             :                 /*
    1972             :                  * User wrote "(column)" or "(column COLLATE something)".
    1973             :                  * Treat it like simple attribute anyway.
    1974             :                  */
    1975          12 :                 indexInfo->ii_IndexAttrNumbers[attn] = ((Var *) expr)->varattno;
    1976             :             }
    1977             :             else
    1978             :             {
    1979        1008 :                 indexInfo->ii_IndexAttrNumbers[attn] = 0;    /* marks expression */
    1980        1008 :                 indexInfo->ii_Expressions = lappend(indexInfo->ii_Expressions,
    1981             :                                                     expr);
    1982             : 
    1983             :                 /*
    1984             :                  * transformExpr() should have already rejected subqueries,
    1985             :                  * aggregates, and window functions, based on the EXPR_KIND_
    1986             :                  * for an index expression.
    1987             :                  */
    1988             : 
    1989             :                 /*
    1990             :                  * An expression using mutable functions is probably wrong,
    1991             :                  * since if you aren't going to get the same result for the
    1992             :                  * same data every time, it's not clear what the index entries
    1993             :                  * mean at all.
    1994             :                  */
    1995        1008 :                 if (contain_mutable_functions_after_planning((Expr *) expr))
    1996         168 :                     ereport(ERROR,
    1997             :                             (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    1998             :                              errmsg("functions in index expression must be marked IMMUTABLE")));
    1999             :             }
    2000             :         }
    2001             : 
    2002       37518 :         typeOids[attn] = atttype;
    2003             : 
    2004             :         /*
    2005             :          * Included columns have no collation, no opclass and no ordering
    2006             :          * options.
    2007             :          */
    2008       37518 :         if (attn >= nkeycols)
    2009             :         {
    2010         638 :             if (attribute->collation)
    2011           0 :                 ereport(ERROR,
    2012             :                         (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2013             :                          errmsg("including column does not support a collation")));
    2014         638 :             if (attribute->opclass)
    2015           0 :                 ereport(ERROR,
    2016             :                         (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2017             :                          errmsg("including column does not support an operator class")));
    2018         638 :             if (attribute->ordering != SORTBY_DEFAULT)
    2019           0 :                 ereport(ERROR,
    2020             :                         (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2021             :                          errmsg("including column does not support ASC/DESC options")));
    2022         638 :             if (attribute->nulls_ordering != SORTBY_NULLS_DEFAULT)
    2023           0 :                 ereport(ERROR,
    2024             :                         (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2025             :                          errmsg("including column does not support NULLS FIRST/LAST options")));
    2026             : 
    2027         638 :             opclassOids[attn] = InvalidOid;
    2028         638 :             opclassOptions[attn] = (Datum) 0;
    2029         638 :             colOptions[attn] = 0;
    2030         638 :             collationOids[attn] = InvalidOid;
    2031         638 :             attn++;
    2032             : 
    2033         638 :             continue;
    2034             :         }
    2035             : 
    2036             :         /*
    2037             :          * Apply collation override if any.  Use of ddl_userid is necessary
    2038             :          * due to ACL checks therein, and it's safe because collations don't
    2039             :          * contain opaque expressions (or non-opaque expressions).
    2040             :          */
    2041       36880 :         if (attribute->collation)
    2042             :         {
    2043         112 :             if (OidIsValid(ddl_userid))
    2044             :             {
    2045         112 :                 AtEOXact_GUC(false, *ddl_save_nestlevel);
    2046         112 :                 SetUserIdAndSecContext(ddl_userid, ddl_sec_context);
    2047             :             }
    2048         112 :             attcollation = get_collation_oid(attribute->collation, false);
    2049         110 :             if (OidIsValid(ddl_userid))
    2050             :             {
    2051         110 :                 SetUserIdAndSecContext(save_userid, save_sec_context);
    2052         110 :                 *ddl_save_nestlevel = NewGUCNestLevel();
    2053             :             }
    2054             :         }
    2055             : 
    2056             :         /*
    2057             :          * Check we have a collation iff it's a collatable type.  The only
    2058             :          * expected failures here are (1) COLLATE applied to a noncollatable
    2059             :          * type, or (2) index expression had an unresolved collation.  But we
    2060             :          * might as well code this to be a complete consistency check.
    2061             :          */
    2062       36878 :         if (type_is_collatable(atttype))
    2063             :         {
    2064        5772 :             if (!OidIsValid(attcollation))
    2065           0 :                 ereport(ERROR,
    2066             :                         (errcode(ERRCODE_INDETERMINATE_COLLATION),
    2067             :                          errmsg("could not determine which collation to use for index expression"),
    2068             :                          errhint("Use the COLLATE clause to set the collation explicitly.")));
    2069             :         }
    2070             :         else
    2071             :         {
    2072       31106 :             if (OidIsValid(attcollation))
    2073          12 :                 ereport(ERROR,
    2074             :                         (errcode(ERRCODE_DATATYPE_MISMATCH),
    2075             :                          errmsg("collations are not supported by type %s",
    2076             :                                 format_type_be(atttype))));
    2077             :         }
    2078             : 
    2079       36866 :         collationOids[attn] = attcollation;
    2080             : 
    2081             :         /*
    2082             :          * Identify the opclass to use.  Use of ddl_userid is necessary due to
    2083             :          * ACL checks therein.  This is safe despite opclasses containing
    2084             :          * opaque expressions (specifically, functions), because only
    2085             :          * superusers can define opclasses.
    2086             :          */
    2087       36866 :         if (OidIsValid(ddl_userid))
    2088             :         {
    2089       36758 :             AtEOXact_GUC(false, *ddl_save_nestlevel);
    2090       36758 :             SetUserIdAndSecContext(ddl_userid, ddl_sec_context);
    2091             :         }
    2092       36866 :         opclassOids[attn] = ResolveOpClass(attribute->opclass,
    2093             :                                            atttype,
    2094             :                                            accessMethodName,
    2095             :                                            accessMethodId);
    2096       36848 :         if (OidIsValid(ddl_userid))
    2097             :         {
    2098       36740 :             SetUserIdAndSecContext(save_userid, save_sec_context);
    2099       36740 :             *ddl_save_nestlevel = NewGUCNestLevel();
    2100             :         }
    2101             : 
    2102             :         /*
    2103             :          * Identify the exclusion operator, if any.
    2104             :          */
    2105       36848 :         if (nextExclOp)
    2106             :         {
    2107         444 :             List       *opname = (List *) lfirst(nextExclOp);
    2108             :             Oid         opid;
    2109             :             Oid         opfamily;
    2110             :             int         strat;
    2111             : 
    2112             :             /*
    2113             :              * Find the operator --- it must accept the column datatype
    2114             :              * without runtime coercion (but binary compatibility is OK).
    2115             :              * Operators contain opaque expressions (specifically, functions).
    2116             :              * compatible_oper_opid() boils down to oper() and
    2117             :              * IsBinaryCoercible().  PostgreSQL would have security problems
    2118             :              * elsewhere if oper() started calling opaque expressions.
    2119             :              */
    2120         444 :             if (OidIsValid(ddl_userid))
    2121             :             {
    2122         444 :                 AtEOXact_GUC(false, *ddl_save_nestlevel);
    2123         444 :                 SetUserIdAndSecContext(ddl_userid, ddl_sec_context);
    2124             :             }
    2125         444 :             opid = compatible_oper_opid(opname, atttype, atttype, false);
    2126         444 :             if (OidIsValid(ddl_userid))
    2127             :             {
    2128         444 :                 SetUserIdAndSecContext(save_userid, save_sec_context);
    2129         444 :                 *ddl_save_nestlevel = NewGUCNestLevel();
    2130             :             }
    2131             : 
    2132             :             /*
    2133             :              * Only allow commutative operators to be used in exclusion
    2134             :              * constraints. If X conflicts with Y, but Y does not conflict
    2135             :              * with X, bad things will happen.
    2136             :              */
    2137         444 :             if (get_commutator(opid) != opid)
    2138           0 :                 ereport(ERROR,
    2139             :                         (errcode(ERRCODE_WRONG_OBJECT_TYPE),
    2140             :                          errmsg("operator %s is not commutative",
    2141             :                                 format_operator(opid)),
    2142             :                          errdetail("Only commutative operators can be used in exclusion constraints.")));
    2143             : 
    2144             :             /*
    2145             :              * Operator must be a member of the right opfamily, too
    2146             :              */
    2147         444 :             opfamily = get_opclass_family(opclassOids[attn]);
    2148         444 :             strat = get_op_opfamily_strategy(opid, opfamily);
    2149         444 :             if (strat == 0)
    2150             :             {
    2151             :                 HeapTuple   opftuple;
    2152             :                 Form_pg_opfamily opfform;
    2153             : 
    2154             :                 /*
    2155             :                  * attribute->opclass might not explicitly name the opfamily,
    2156             :                  * so fetch the name of the selected opfamily for use in the
    2157             :                  * error message.
    2158             :                  */
    2159           0 :                 opftuple = SearchSysCache1(OPFAMILYOID,
    2160             :                                            ObjectIdGetDatum(opfamily));
    2161           0 :                 if (!HeapTupleIsValid(opftuple))
    2162           0 :                     elog(ERROR, "cache lookup failed for opfamily %u",
    2163             :                          opfamily);
    2164           0 :                 opfform = (Form_pg_opfamily) GETSTRUCT(opftuple);
    2165             : 
    2166           0 :                 ereport(ERROR,
    2167             :                         (errcode(ERRCODE_WRONG_OBJECT_TYPE),
    2168             :                          errmsg("operator %s is not a member of operator family \"%s\"",
    2169             :                                 format_operator(opid),
    2170             :                                 NameStr(opfform->opfname)),
    2171             :                          errdetail("The exclusion operator must be related to the index operator class for the constraint.")));
    2172             :             }
    2173             : 
    2174         444 :             indexInfo->ii_ExclusionOps[attn] = opid;
    2175         444 :             indexInfo->ii_ExclusionProcs[attn] = get_opcode(opid);
    2176         444 :             indexInfo->ii_ExclusionStrats[attn] = strat;
    2177         444 :             nextExclOp = lnext(exclusionOpNames, nextExclOp);
    2178             :         }
    2179       36404 :         else if (iswithoutoverlaps)
    2180             :         {
    2181             :             StrategyNumber strat;
    2182             :             Oid         opid;
    2183             : 
    2184         752 :             if (attn == nkeycols - 1)
    2185         356 :                 strat = RTOverlapStrategyNumber;
    2186             :             else
    2187         396 :                 strat = RTEqualStrategyNumber;
    2188         752 :             GetOperatorFromWellKnownStrategy(opclassOids[attn], InvalidOid,
    2189             :                                              &opid, &strat);
    2190         752 :             indexInfo->ii_ExclusionOps[attn] = opid;
    2191         752 :             indexInfo->ii_ExclusionProcs[attn] = get_opcode(opid);
    2192         752 :             indexInfo->ii_ExclusionStrats[attn] = strat;
    2193             :         }
    2194             : 
    2195             :         /*
    2196             :          * Set up the per-column options (indoption field).  For now, this is
    2197             :          * zero for any un-ordered index, while ordered indexes have DESC and
    2198             :          * NULLS FIRST/LAST options.
    2199             :          */
    2200       36848 :         colOptions[attn] = 0;
    2201       36848 :         if (amcanorder)
    2202             :         {
    2203             :             /* default ordering is ASC */
    2204       33234 :             if (attribute->ordering == SORTBY_DESC)
    2205          42 :                 colOptions[attn] |= INDOPTION_DESC;
    2206             :             /* default null ordering is LAST for ASC, FIRST for DESC */
    2207       33234 :             if (attribute->nulls_ordering == SORTBY_NULLS_DEFAULT)
    2208             :             {
    2209       33204 :                 if (attribute->ordering == SORTBY_DESC)
    2210          30 :                     colOptions[attn] |= INDOPTION_NULLS_FIRST;
    2211             :             }
    2212          30 :             else if (attribute->nulls_ordering == SORTBY_NULLS_FIRST)
    2213          12 :                 colOptions[attn] |= INDOPTION_NULLS_FIRST;
    2214             :         }
    2215             :         else
    2216             :         {
    2217             :             /* index AM does not support ordering */
    2218        3614 :             if (attribute->ordering != SORTBY_DEFAULT)
    2219           0 :                 ereport(ERROR,
    2220             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    2221             :                          errmsg("access method \"%s\" does not support ASC/DESC options",
    2222             :                                 accessMethodName)));
    2223        3614 :             if (attribute->nulls_ordering != SORTBY_NULLS_DEFAULT)
    2224           0 :                 ereport(ERROR,
    2225             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    2226             :                          errmsg("access method \"%s\" does not support NULLS FIRST/LAST options",
    2227             :                                 accessMethodName)));
    2228             :         }
    2229             : 
    2230             :         /* Set up the per-column opclass options (attoptions field). */
    2231       36848 :         if (attribute->opclassopts)
    2232             :         {
    2233             :             Assert(attn < nkeycols);
    2234             : 
    2235         138 :             opclassOptions[attn] =
    2236         138 :                 transformRelOptions((Datum) 0, attribute->opclassopts,
    2237             :                                     NULL, NULL, false, false);
    2238             :         }
    2239             :         else
    2240       36710 :             opclassOptions[attn] = (Datum) 0;
    2241             : 
    2242       36848 :         attn++;
    2243             :     }
    2244       26952 : }
    2245             : 
    2246             : /*
    2247             :  * Resolve possibly-defaulted operator class specification
    2248             :  *
    2249             :  * Note: This is used to resolve operator class specifications in index and
    2250             :  * partition key definitions.
    2251             :  */
    2252             : Oid
    2253       36998 : ResolveOpClass(const List *opclass, Oid attrType,
    2254             :                const char *accessMethodName, Oid accessMethodId)
    2255             : {
    2256             :     char       *schemaname;
    2257             :     char       *opcname;
    2258             :     HeapTuple   tuple;
    2259             :     Form_pg_opclass opform;
    2260             :     Oid         opClassId,
    2261             :                 opInputType;
    2262             : 
    2263       36998 :     if (opclass == NIL)
    2264             :     {
    2265             :         /* no operator class specified, so find the default */
    2266       19360 :         opClassId = GetDefaultOpClass(attrType, accessMethodId);
    2267       19360 :         if (!OidIsValid(opClassId))
    2268          18 :             ereport(ERROR,
    2269             :                     (errcode(ERRCODE_UNDEFINED_OBJECT),
    2270             :                      errmsg("data type %s has no default operator class for access method \"%s\"",
    2271             :                             format_type_be(attrType), accessMethodName),
    2272             :                      errhint("You must specify an operator class for the index or define a default operator class for the data type.")));
    2273       19342 :         return opClassId;
    2274             :     }
    2275             : 
    2276             :     /*
    2277             :      * Specific opclass name given, so look up the opclass.
    2278             :      */
    2279             : 
    2280             :     /* deconstruct the name list */
    2281       17638 :     DeconstructQualifiedName(opclass, &schemaname, &opcname);
    2282             : 
    2283       17638 :     if (schemaname)
    2284             :     {
    2285             :         /* Look in specific schema only */
    2286             :         Oid         namespaceId;
    2287             : 
    2288          10 :         namespaceId = LookupExplicitNamespace(schemaname, false);
    2289          10 :         tuple = SearchSysCache3(CLAAMNAMENSP,
    2290             :                                 ObjectIdGetDatum(accessMethodId),
    2291             :                                 PointerGetDatum(opcname),
    2292             :                                 ObjectIdGetDatum(namespaceId));
    2293             :     }
    2294             :     else
    2295             :     {
    2296             :         /* Unqualified opclass name, so search the search path */
    2297       17628 :         opClassId = OpclassnameGetOpcid(accessMethodId, opcname);
    2298       17628 :         if (!OidIsValid(opClassId))
    2299          12 :             ereport(ERROR,
    2300             :                     (errcode(ERRCODE_UNDEFINED_OBJECT),
    2301             :                      errmsg("operator class \"%s\" does not exist for access method \"%s\"",
    2302             :                             opcname, accessMethodName)));
    2303       17616 :         tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(opClassId));
    2304             :     }
    2305             : 
    2306       17626 :     if (!HeapTupleIsValid(tuple))
    2307           0 :         ereport(ERROR,
    2308             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
    2309             :                  errmsg("operator class \"%s\" does not exist for access method \"%s\"",
    2310             :                         NameListToString(opclass), accessMethodName)));
    2311             : 
    2312             :     /*
    2313             :      * Verify that the index operator class accepts this datatype.  Note we
    2314             :      * will accept binary compatibility.
    2315             :      */
    2316       17626 :     opform = (Form_pg_opclass) GETSTRUCT(tuple);
    2317       17626 :     opClassId = opform->oid;
    2318       17626 :     opInputType = opform->opcintype;
    2319             : 
    2320       17626 :     if (!IsBinaryCoercible(attrType, opInputType))
    2321           0 :         ereport(ERROR,
    2322             :                 (errcode(ERRCODE_DATATYPE_MISMATCH),
    2323             :                  errmsg("operator class \"%s\" does not accept data type %s",
    2324             :                         NameListToString(opclass), format_type_be(attrType))));
    2325             : 
    2326       17626 :     ReleaseSysCache(tuple);
    2327             : 
    2328       17626 :     return opClassId;
    2329             : }
    2330             : 
    2331             : /*
    2332             :  * GetDefaultOpClass
    2333             :  *
    2334             :  * Given the OIDs of a datatype and an access method, find the default
    2335             :  * operator class, if any.  Returns InvalidOid if there is none.
    2336             :  */
    2337             : Oid
    2338       93774 : GetDefaultOpClass(Oid type_id, Oid am_id)
    2339             : {
    2340       93774 :     Oid         result = InvalidOid;
    2341       93774 :     int         nexact = 0;
    2342       93774 :     int         ncompatible = 0;
    2343       93774 :     int         ncompatiblepreferred = 0;
    2344             :     Relation    rel;
    2345             :     ScanKeyData skey[1];
    2346             :     SysScanDesc scan;
    2347             :     HeapTuple   tup;
    2348             :     TYPCATEGORY tcategory;
    2349             : 
    2350             :     /* If it's a domain, look at the base type instead */
    2351       93774 :     type_id = getBaseType(type_id);
    2352             : 
    2353       93774 :     tcategory = TypeCategory(type_id);
    2354             : 
    2355             :     /*
    2356             :      * We scan through all the opclasses available for the access method,
    2357             :      * looking for one that is marked default and matches the target type
    2358             :      * (either exactly or binary-compatibly, but prefer an exact match).
    2359             :      *
    2360             :      * We could find more than one binary-compatible match.  If just one is
    2361             :      * for a preferred type, use that one; otherwise we fail, forcing the user
    2362             :      * to specify which one he wants.  (The preferred-type special case is a
    2363             :      * kluge for varchar: it's binary-compatible to both text and bpchar, so
    2364             :      * we need a tiebreaker.)  If we find more than one exact match, then
    2365             :      * someone put bogus entries in pg_opclass.
    2366             :      */
    2367       93774 :     rel = table_open(OperatorClassRelationId, AccessShareLock);
    2368             : 
    2369       93774 :     ScanKeyInit(&skey[0],
    2370             :                 Anum_pg_opclass_opcmethod,
    2371             :                 BTEqualStrategyNumber, F_OIDEQ,
    2372             :                 ObjectIdGetDatum(am_id));
    2373             : 
    2374       93774 :     scan = systable_beginscan(rel, OpclassAmNameNspIndexId, true,
    2375             :                               NULL, 1, skey);
    2376             : 
    2377     4112740 :     while (HeapTupleIsValid(tup = systable_getnext(scan)))
    2378             :     {
    2379     4018966 :         Form_pg_opclass opclass = (Form_pg_opclass) GETSTRUCT(tup);
    2380             : 
    2381             :         /* ignore altogether if not a default opclass */
    2382     4018966 :         if (!opclass->opcdefault)
    2383      603708 :             continue;
    2384     3415258 :         if (opclass->opcintype == type_id)
    2385             :         {
    2386       83548 :             nexact++;
    2387       83548 :             result = opclass->oid;
    2388             :         }
    2389     4992062 :         else if (nexact == 0 &&
    2390     1660352 :                  IsBinaryCoercible(type_id, opclass->opcintype))
    2391             :         {
    2392       20468 :             if (IsPreferredType(tcategory, opclass->opcintype))
    2393             :             {
    2394        1748 :                 ncompatiblepreferred++;
    2395        1748 :                 result = opclass->oid;
    2396             :             }
    2397       18720 :             else if (ncompatiblepreferred == 0)
    2398             :             {
    2399       18720 :                 ncompatible++;
    2400       18720 :                 result = opclass->oid;
    2401             :             }
    2402             :         }
    2403             :     }
    2404             : 
    2405       93774 :     systable_endscan(scan);
    2406             : 
    2407       93774 :     table_close(rel, AccessShareLock);
    2408             : 
    2409             :     /* raise error if pg_opclass contains inconsistent data */
    2410       93774 :     if (nexact > 1)
    2411           0 :         ereport(ERROR,
    2412             :                 (errcode(ERRCODE_DUPLICATE_OBJECT),
    2413             :                  errmsg("there are multiple default operator classes for data type %s",
    2414             :                         format_type_be(type_id))));
    2415             : 
    2416       93774 :     if (nexact == 1 ||
    2417        8480 :         ncompatiblepreferred == 1 ||
    2418        8480 :         (ncompatiblepreferred == 0 && ncompatible == 1))
    2419       92782 :         return result;
    2420             : 
    2421         992 :     return InvalidOid;
    2422             : }
    2423             : 
    2424             : /*
    2425             :  * GetOperatorFromWellKnownStrategy
    2426             :  *
    2427             :  * opclass - the opclass to use
    2428             :  * rhstype - the type for the right-hand side, or InvalidOid to use the type of the given opclass.
    2429             :  * opid - holds the operator we found
    2430             :  * strat - holds the input and output strategy number
    2431             :  *
    2432             :  * Finds an operator from a "well-known" strategy number.  This is used for
    2433             :  * temporal index constraints (and other temporal features) to look up
    2434             :  * equality and overlaps operators, since the strategy numbers for non-btree
    2435             :  * indexams need not follow any fixed scheme.  We ask an opclass support
    2436             :  * function to translate from the well-known number to the internal value.  If
    2437             :  * the function isn't defined or it gives no result, we return
    2438             :  * InvalidStrategy.
    2439             :  */
    2440             : void
    2441        1556 : GetOperatorFromWellKnownStrategy(Oid opclass, Oid rhstype,
    2442             :                                  Oid *opid, StrategyNumber *strat)
    2443             : {
    2444             :     Oid         opfamily;
    2445             :     Oid         opcintype;
    2446        1556 :     StrategyNumber instrat = *strat;
    2447             : 
    2448             :     Assert(instrat == RTEqualStrategyNumber || instrat == RTOverlapStrategyNumber || instrat == RTContainedByStrategyNumber);
    2449             : 
    2450        1556 :     *opid = InvalidOid;
    2451             : 
    2452        1556 :     if (get_opclass_opfamily_and_input_type(opclass, &opfamily, &opcintype))
    2453             :     {
    2454             :         /*
    2455             :          * Ask the opclass to translate to its internal stratnum
    2456             :          *
    2457             :          * For now we only need GiST support, but this could support other
    2458             :          * indexams if we wanted.
    2459             :          */
    2460        1556 :         *strat = GistTranslateStratnum(opclass, instrat);
    2461        1556 :         if (*strat == InvalidStrategy)
    2462             :         {
    2463             :             HeapTuple   tuple;
    2464             : 
    2465           0 :             tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclass));
    2466           0 :             if (!HeapTupleIsValid(tuple))
    2467           0 :                 elog(ERROR, "cache lookup failed for operator class %u", opclass);
    2468             : 
    2469           0 :             ereport(ERROR,
    2470             :                     errcode(ERRCODE_UNDEFINED_OBJECT),
    2471             :                     instrat == RTEqualStrategyNumber ? errmsg("could not identify an equality operator for type %s", format_type_be(opcintype)) :
    2472             :                     instrat == RTOverlapStrategyNumber ? errmsg("could not identify an overlaps operator for type %s", format_type_be(opcintype)) :
    2473             :                     instrat == RTContainedByStrategyNumber ? errmsg("could not identify a contained-by operator for type %s", format_type_be(opcintype)) : 0,
    2474             :                     errdetail("Could not translate strategy number %d for operator class \"%s\" for access method \"%s\".",
    2475             :                               instrat, NameStr(((Form_pg_opclass) GETSTRUCT(tuple))->opcname), "gist"));
    2476             :         }
    2477             : 
    2478             :         /*
    2479             :          * We parameterize rhstype so foreign keys can ask for a <@ operator
    2480             :          * whose rhs matches the aggregate function. For example range_agg
    2481             :          * returns anymultirange.
    2482             :          */
    2483        1556 :         if (!OidIsValid(rhstype))
    2484        1154 :             rhstype = opcintype;
    2485        1556 :         *opid = get_opfamily_member(opfamily, opcintype, rhstype, *strat);
    2486             :     }
    2487             : 
    2488        1556 :     if (!OidIsValid(*opid))
    2489             :     {
    2490             :         HeapTuple   tuple;
    2491             : 
    2492           0 :         tuple = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfamily));
    2493           0 :         if (!HeapTupleIsValid(tuple))
    2494           0 :             elog(ERROR, "cache lookup failed for operator family %u", opfamily);
    2495             : 
    2496           0 :         ereport(ERROR,
    2497             :                 errcode(ERRCODE_UNDEFINED_OBJECT),
    2498             :                 instrat == RTEqualStrategyNumber ? errmsg("could not identify an equality operator for type %s", format_type_be(opcintype)) :
    2499             :                 instrat == RTOverlapStrategyNumber ? errmsg("could not identify an overlaps operator for type %s", format_type_be(opcintype)) :
    2500             :                 instrat == RTContainedByStrategyNumber ? errmsg("could not identify a contained-by operator for type %s", format_type_be(opcintype)) : 0,
    2501             :                 errdetail("There is no suitable operator in operator family \"%s\" for access method \"%s\".",
    2502             :                           NameStr(((Form_pg_opfamily) GETSTRUCT(tuple))->opfname), "gist"));
    2503             :     }
    2504        1556 : }
    2505             : 
    2506             : /*
    2507             :  *  makeObjectName()
    2508             :  *
    2509             :  *  Create a name for an implicitly created index, sequence, constraint,
    2510             :  *  extended statistics, etc.
    2511             :  *
    2512             :  *  The parameters are typically: the original table name, the original field
    2513             :  *  name, and a "type" string (such as "seq" or "pkey").    The field name
    2514             :  *  and/or type can be NULL if not relevant.
    2515             :  *
    2516             :  *  The result is a palloc'd string.
    2517             :  *
    2518             :  *  The basic result we want is "name1_name2_label", omitting "_name2" or
    2519             :  *  "_label" when those parameters are NULL.  However, we must generate
    2520             :  *  a name with less than NAMEDATALEN characters!  So, we truncate one or
    2521             :  *  both names if necessary to make a short-enough string.  The label part
    2522             :  *  is never truncated (so it had better be reasonably short).
    2523             :  *
    2524             :  *  The caller is responsible for checking uniqueness of the generated
    2525             :  *  name and retrying as needed; retrying will be done by altering the
    2526             :  *  "label" string (which is why we never truncate that part).
    2527             :  */
    2528             : char *
    2529       85736 : makeObjectName(const char *name1, const char *name2, const char *label)
    2530             : {
    2531             :     char       *name;
    2532       85736 :     int         overhead = 0;   /* chars needed for label and underscores */
    2533             :     int         availchars;     /* chars available for name(s) */
    2534             :     int         name1chars;     /* chars allocated to name1 */
    2535             :     int         name2chars;     /* chars allocated to name2 */
    2536             :     int         ndx;
    2537             : 
    2538       85736 :     name1chars = strlen(name1);
    2539       85736 :     if (name2)
    2540             :     {
    2541       76854 :         name2chars = strlen(name2);
    2542       76854 :         overhead++;             /* allow for separating underscore */
    2543             :     }
    2544             :     else
    2545        8882 :         name2chars = 0;
    2546       85736 :     if (label)
    2547       26118 :         overhead += strlen(label) + 1;
    2548             : 
    2549       85736 :     availchars = NAMEDATALEN - 1 - overhead;
    2550             :     Assert(availchars > 0);      /* else caller chose a bad label */
    2551             : 
    2552             :     /*
    2553             :      * If we must truncate, preferentially truncate the longer name. This
    2554             :      * logic could be expressed without a loop, but it's simple and obvious as
    2555             :      * a loop.
    2556             :      */
    2557       85802 :     while (name1chars + name2chars > availchars)
    2558             :     {
    2559          66 :         if (name1chars > name2chars)
    2560           0 :             name1chars--;
    2561             :         else
    2562          66 :             name2chars--;
    2563             :     }
    2564             : 
    2565       85736 :     name1chars = pg_mbcliplen(name1, name1chars, name1chars);
    2566       85736 :     if (name2)
    2567       76854 :         name2chars = pg_mbcliplen(name2, name2chars, name2chars);
    2568             : 
    2569             :     /* Now construct the string using the chosen lengths */
    2570       85736 :     name = palloc(name1chars + name2chars + overhead + 1);
    2571       85736 :     memcpy(name, name1, name1chars);
    2572       85736 :     ndx = name1chars;
    2573       85736 :     if (name2)
    2574             :     {
    2575       76854 :         name[ndx++] = '_';
    2576       76854 :         memcpy(name + ndx, name2, name2chars);
    2577       76854 :         ndx += name2chars;
    2578             :     }
    2579       85736 :     if (label)
    2580             :     {
    2581       26118 :         name[ndx++] = '_';
    2582       26118 :         strcpy(name + ndx, label);
    2583             :     }
    2584             :     else
    2585       59618 :         name[ndx] = '\0';
    2586             : 
    2587       85736 :     return name;
    2588             : }
    2589             : 
    2590             : /*
    2591             :  * Select a nonconflicting name for a new relation.  This is ordinarily
    2592             :  * used to choose index names (which is why it's here) but it can also
    2593             :  * be used for sequences, or any autogenerated relation kind.
    2594             :  *
    2595             :  * name1, name2, and label are used the same way as for makeObjectName(),
    2596             :  * except that the label can't be NULL; digits will be appended to the label
    2597             :  * if needed to create a name that is unique within the specified namespace.
    2598             :  *
    2599             :  * If isconstraint is true, we also avoid choosing a name matching any
    2600             :  * existing constraint in the same namespace.  (This is stricter than what
    2601             :  * Postgres itself requires, but the SQL standard says that constraint names
    2602             :  * should be unique within schemas, so we follow that for autogenerated
    2603             :  * constraint names.)
    2604             :  *
    2605             :  * Note: it is theoretically possible to get a collision anyway, if someone
    2606             :  * else chooses the same name concurrently.  This is fairly unlikely to be
    2607             :  * a problem in practice, especially if one is holding an exclusive lock on
    2608             :  * the relation identified by name1.  However, if choosing multiple names
    2609             :  * within a single command, you'd better create the new object and do
    2610             :  * CommandCounterIncrement before choosing the next one!
    2611             :  *
    2612             :  * Returns a palloc'd string.
    2613             :  */
    2614             : char *
    2615       13174 : ChooseRelationName(const char *name1, const char *name2,
    2616             :                    const char *label, Oid namespaceid,
    2617             :                    bool isconstraint)
    2618             : {
    2619       13174 :     int         pass = 0;
    2620       13174 :     char       *relname = NULL;
    2621             :     char        modlabel[NAMEDATALEN];
    2622             : 
    2623             :     /* try the unmodified label first */
    2624       13174 :     strlcpy(modlabel, label, sizeof(modlabel));
    2625             : 
    2626             :     for (;;)
    2627             :     {
    2628       14266 :         relname = makeObjectName(name1, name2, modlabel);
    2629             : 
    2630       14266 :         if (!OidIsValid(get_relname_relid(relname, namespaceid)))
    2631             :         {
    2632       13180 :             if (!isconstraint ||
    2633        8306 :                 !ConstraintNameExists(relname, namespaceid))
    2634             :                 break;
    2635             :         }
    2636             : 
    2637             :         /* found a conflict, so try a new name component */
    2638        1092 :         pfree(relname);
    2639        1092 :         snprintf(modlabel, sizeof(modlabel), "%s%d", label, ++pass);
    2640             :     }
    2641             : 
    2642       13174 :     return relname;
    2643             : }
    2644             : 
    2645             : /*
    2646             :  * Select the name to be used for an index.
    2647             :  *
    2648             :  * The argument list is pretty ad-hoc :-(
    2649             :  */
    2650             : static char *
    2651       11074 : ChooseIndexName(const char *tabname, Oid namespaceId,
    2652             :                 const List *colnames, const List *exclusionOpNames,
    2653             :                 bool primary, bool isconstraint)
    2654             : {
    2655             :     char       *indexname;
    2656             : 
    2657       11074 :     if (primary)
    2658             :     {
    2659             :         /* the primary key's name does not depend on the specific column(s) */
    2660        7422 :         indexname = ChooseRelationName(tabname,
    2661             :                                        NULL,
    2662             :                                        "pkey",
    2663             :                                        namespaceId,
    2664             :                                        true);
    2665             :     }
    2666        3652 :     else if (exclusionOpNames != NIL)
    2667             :     {
    2668         206 :         indexname = ChooseRelationName(tabname,
    2669         206 :                                        ChooseIndexNameAddition(colnames),
    2670             :                                        "excl",
    2671             :                                        namespaceId,
    2672             :                                        true);
    2673             :     }
    2674        3446 :     else if (isconstraint)
    2675             :     {
    2676         672 :         indexname = ChooseRelationName(tabname,
    2677         672 :                                        ChooseIndexNameAddition(colnames),
    2678             :                                        "key",
    2679             :                                        namespaceId,
    2680             :                                        true);
    2681             :     }
    2682             :     else
    2683             :     {
    2684        2774 :         indexname = ChooseRelationName(tabname,
    2685        2774 :                                        ChooseIndexNameAddition(colnames),
    2686             :                                        "idx",
    2687             :                                        namespaceId,
    2688             :                                        false);
    2689             :     }
    2690             : 
    2691       11074 :     return indexname;
    2692             : }
    2693             : 
    2694             : /*
    2695             :  * Generate "name2" for a new index given the list of column names for it
    2696             :  * (as produced by ChooseIndexColumnNames).  This will be passed to
    2697             :  * ChooseRelationName along with the parent table name and a suitable label.
    2698             :  *
    2699             :  * We know that less than NAMEDATALEN characters will actually be used,
    2700             :  * so we can truncate the result once we've generated that many.
    2701             :  *
    2702             :  * XXX See also ChooseForeignKeyConstraintNameAddition and
    2703             :  * ChooseExtendedStatisticNameAddition.
    2704             :  */
    2705             : static char *
    2706        3652 : ChooseIndexNameAddition(const List *colnames)
    2707             : {
    2708             :     char        buf[NAMEDATALEN * 2];
    2709        3652 :     int         buflen = 0;
    2710             :     ListCell   *lc;
    2711             : 
    2712        3652 :     buf[0] = '\0';
    2713        8330 :     foreach(lc, colnames)
    2714             :     {
    2715        4678 :         const char *name = (const char *) lfirst(lc);
    2716             : 
    2717        4678 :         if (buflen > 0)
    2718        1026 :             buf[buflen++] = '_';    /* insert _ between names */
    2719             : 
    2720             :         /*
    2721             :          * At this point we have buflen <= NAMEDATALEN.  name should be less
    2722             :          * than NAMEDATALEN already, but use strlcpy for paranoia.
    2723             :          */
    2724        4678 :         strlcpy(buf + buflen, name, NAMEDATALEN);
    2725        4678 :         buflen += strlen(buf + buflen);
    2726        4678 :         if (buflen >= NAMEDATALEN)
    2727           0 :             break;
    2728             :     }
    2729        3652 :     return pstrdup(buf);
    2730             : }
    2731             : 
    2732             : /*
    2733             :  * Select the actual names to be used for the columns of an index, given the
    2734             :  * list of IndexElems for the columns.  This is mostly about ensuring the
    2735             :  * names are unique so we don't get a conflicting-attribute-names error.
    2736             :  *
    2737             :  * Returns a List of plain strings (char *, not String nodes).
    2738             :  */
    2739             : static List *
    2740       27168 : ChooseIndexColumnNames(const List *indexElems)
    2741             : {
    2742       27168 :     List       *result = NIL;
    2743             :     ListCell   *lc;
    2744             : 
    2745       64922 :     foreach(lc, indexElems)
    2746             :     {
    2747       37754 :         IndexElem  *ielem = (IndexElem *) lfirst(lc);
    2748             :         const char *origname;
    2749             :         const char *curname;
    2750             :         int         i;
    2751             :         char        buf[NAMEDATALEN];
    2752             : 
    2753             :         /* Get the preliminary name from the IndexElem */
    2754       37754 :         if (ielem->indexcolname)
    2755        2822 :             origname = ielem->indexcolname; /* caller-specified name */
    2756       34932 :         else if (ielem->name)
    2757       34520 :             origname = ielem->name; /* simple column reference */
    2758             :         else
    2759         412 :             origname = "expr";    /* default name for expression */
    2760             : 
    2761             :         /* If it conflicts with any previous column, tweak it */
    2762       37754 :         curname = origname;
    2763       37754 :         for (i = 1;; i++)
    2764          56 :         {
    2765             :             ListCell   *lc2;
    2766             :             char        nbuf[32];
    2767             :             int         nlen;
    2768             : 
    2769       59850 :             foreach(lc2, result)
    2770             :             {
    2771       22096 :                 if (strcmp(curname, (char *) lfirst(lc2)) == 0)
    2772          56 :                     break;
    2773             :             }
    2774       37810 :             if (lc2 == NULL)
    2775       37754 :                 break;          /* found nonconflicting name */
    2776             : 
    2777          56 :             sprintf(nbuf, "%d", i);
    2778             : 
    2779             :             /* Ensure generated names are shorter than NAMEDATALEN */
    2780          56 :             nlen = pg_mbcliplen(origname, strlen(origname),
    2781          56 :                                 NAMEDATALEN - 1 - strlen(nbuf));
    2782          56 :             memcpy(buf, origname, nlen);
    2783          56 :             strcpy(buf + nlen, nbuf);
    2784          56 :             curname = buf;
    2785             :         }
    2786             : 
    2787             :         /* And attach to the result list */
    2788       37754 :         result = lappend(result, pstrdup(curname));
    2789             :     }
    2790       27168 :     return result;
    2791             : }
    2792             : 
    2793             : /*
    2794             :  * ExecReindex
    2795             :  *
    2796             :  * Primary entry point for manual REINDEX commands.  This is mainly a
    2797             :  * preparation wrapper for the real operations that will happen in
    2798             :  * each subroutine of REINDEX.
    2799             :  */
    2800             : void
    2801        1052 : ExecReindex(ParseState *pstate, const ReindexStmt *stmt, bool isTopLevel)
    2802             : {
    2803        1052 :     ReindexParams params = {0};
    2804             :     ListCell   *lc;
    2805        1052 :     bool        concurrently = false;
    2806        1052 :     bool        verbose = false;
    2807        1052 :     char       *tablespacename = NULL;
    2808             : 
    2809             :     /* Parse option list */
    2810        1770 :     foreach(lc, stmt->params)
    2811             :     {
    2812         718 :         DefElem    *opt = (DefElem *) lfirst(lc);
    2813             : 
    2814         718 :         if (strcmp(opt->defname, "verbose") == 0)
    2815          14 :             verbose = defGetBoolean(opt);
    2816         704 :         else if (strcmp(opt->defname, "concurrently") == 0)
    2817         576 :             concurrently = defGetBoolean(opt);
    2818         128 :         else if (strcmp(opt->defname, "tablespace") == 0)
    2819         128 :             tablespacename = defGetString(opt);
    2820             :         else
    2821           0 :             ereport(ERROR,
    2822             :                     (errcode(ERRCODE_SYNTAX_ERROR),
    2823             :                      errmsg("unrecognized REINDEX option \"%s\"",
    2824             :                             opt->defname),
    2825             :                      parser_errposition(pstate, opt->location)));
    2826             :     }
    2827             : 
    2828        1052 :     if (concurrently)
    2829         576 :         PreventInTransactionBlock(isTopLevel,
    2830             :                                   "REINDEX CONCURRENTLY");
    2831             : 
    2832        1034 :     params.options =
    2833        2068 :         (verbose ? REINDEXOPT_VERBOSE : 0) |
    2834        1034 :         (concurrently ? REINDEXOPT_CONCURRENTLY : 0);
    2835             : 
    2836             :     /*
    2837             :      * Assign the tablespace OID to move indexes to, with InvalidOid to do
    2838             :      * nothing.
    2839             :      */
    2840        1034 :     if (tablespacename != NULL)
    2841             :     {
    2842         128 :         params.tablespaceOid = get_tablespace_oid(tablespacename, false);
    2843             : 
    2844             :         /* Check permissions except when moving to database's default */
    2845         128 :         if (OidIsValid(params.tablespaceOid) &&
    2846         128 :             params.tablespaceOid != MyDatabaseTableSpace)
    2847             :         {
    2848             :             AclResult   aclresult;
    2849             : 
    2850         128 :             aclresult = object_aclcheck(TableSpaceRelationId, params.tablespaceOid,
    2851             :                                         GetUserId(), ACL_CREATE);
    2852         128 :             if (aclresult != ACLCHECK_OK)
    2853          12 :                 aclcheck_error(aclresult, OBJECT_TABLESPACE,
    2854          12 :                                get_tablespace_name(params.tablespaceOid));
    2855             :         }
    2856             :     }
    2857             :     else
    2858         906 :         params.tablespaceOid = InvalidOid;
    2859             : 
    2860        1022 :     switch (stmt->kind)
    2861             :     {
    2862         360 :         case REINDEX_OBJECT_INDEX:
    2863         360 :             ReindexIndex(stmt, &params, isTopLevel);
    2864         254 :             break;
    2865         480 :         case REINDEX_OBJECT_TABLE:
    2866         480 :             ReindexTable(stmt, &params, isTopLevel);
    2867         358 :             break;
    2868         182 :         case REINDEX_OBJECT_SCHEMA:
    2869             :         case REINDEX_OBJECT_SYSTEM:
    2870             :         case REINDEX_OBJECT_DATABASE:
    2871             : 
    2872             :             /*
    2873             :              * This cannot run inside a user transaction block; if we were
    2874             :              * inside a transaction, then its commit- and
    2875             :              * start-transaction-command calls would not have the intended
    2876             :              * effect!
    2877             :              */
    2878         182 :             PreventInTransactionBlock(isTopLevel,
    2879         250 :                                       (stmt->kind == REINDEX_OBJECT_SCHEMA) ? "REINDEX SCHEMA" :
    2880          68 :                                       (stmt->kind == REINDEX_OBJECT_SYSTEM) ? "REINDEX SYSTEM" :
    2881             :                                       "REINDEX DATABASE");
    2882         176 :             ReindexMultipleTables(stmt, &params);
    2883         126 :             break;
    2884           0 :         default:
    2885           0 :             elog(ERROR, "unrecognized object type: %d",
    2886             :                  (int) stmt->kind);
    2887             :             break;
    2888             :     }
    2889         738 : }
    2890             : 
    2891             : /*
    2892             :  * ReindexIndex
    2893             :  *      Recreate a specific index.
    2894             :  */
    2895             : static void
    2896         360 : ReindexIndex(const ReindexStmt *stmt, const ReindexParams *params, bool isTopLevel)
    2897             : {
    2898         360 :     const RangeVar *indexRelation = stmt->relation;
    2899             :     struct ReindexIndexCallbackState state;
    2900             :     Oid         indOid;
    2901             :     char        persistence;
    2902             :     char        relkind;
    2903             : 
    2904             :     /*
    2905             :      * Find and lock index, and check permissions on table; use callback to
    2906             :      * obtain lock on table first, to avoid deadlock hazard.  The lock level
    2907             :      * used here must match the index lock obtained in reindex_index().
    2908             :      *
    2909             :      * If it's a temporary index, we will perform a non-concurrent reindex,
    2910             :      * even if CONCURRENTLY was requested.  In that case, reindex_index() will
    2911             :      * upgrade the lock, but that's OK, because other sessions can't hold
    2912             :      * locks on our temporary table.
    2913             :      */
    2914         360 :     state.params = *params;
    2915         360 :     state.locked_table_oid = InvalidOid;
    2916         360 :     indOid = RangeVarGetRelidExtended(indexRelation,
    2917         360 :                                       (params->options & REINDEXOPT_CONCURRENTLY) != 0 ?
    2918             :                                       ShareUpdateExclusiveLock : AccessExclusiveLock,
    2919             :                                       0,
    2920             :                                       RangeVarCallbackForReindexIndex,
    2921             :                                       &state);
    2922             : 
    2923             :     /*
    2924             :      * Obtain the current persistence and kind of the existing index.  We
    2925             :      * already hold a lock on the index.
    2926             :      */
    2927         312 :     persistence = get_rel_persistence(indOid);
    2928         312 :     relkind = get_rel_relkind(indOid);
    2929             : 
    2930         312 :     if (relkind == RELKIND_PARTITIONED_INDEX)
    2931          36 :         ReindexPartitions(stmt, indOid, params, isTopLevel);
    2932         276 :     else if ((params->options & REINDEXOPT_CONCURRENTLY) != 0 &&
    2933             :              persistence != RELPERSISTENCE_TEMP)
    2934         156 :         ReindexRelationConcurrently(stmt, indOid, params);
    2935             :     else
    2936             :     {
    2937         120 :         ReindexParams newparams = *params;
    2938             : 
    2939         120 :         newparams.options |= REINDEXOPT_REPORT_PROGRESS;
    2940         120 :         reindex_index(stmt, indOid, false, persistence, &newparams);
    2941             :     }
    2942         254 : }
    2943             : 
    2944             : /*
    2945             :  * Check permissions on table before acquiring relation lock; also lock
    2946             :  * the heap before the RangeVarGetRelidExtended takes the index lock, to avoid
    2947             :  * deadlocks.
    2948             :  */
    2949             : static void
    2950         366 : RangeVarCallbackForReindexIndex(const RangeVar *relation,
    2951             :                                 Oid relId, Oid oldRelId, void *arg)
    2952             : {
    2953             :     char        relkind;
    2954         366 :     struct ReindexIndexCallbackState *state = arg;
    2955             :     LOCKMODE    table_lockmode;
    2956             :     Oid         table_oid;
    2957             : 
    2958             :     /*
    2959             :      * Lock level here should match table lock in reindex_index() for
    2960             :      * non-concurrent case and table locks used by index_concurrently_*() for
    2961             :      * concurrent case.
    2962             :      */
    2963         732 :     table_lockmode = (state->params.options & REINDEXOPT_CONCURRENTLY) != 0 ?
    2964         366 :         ShareUpdateExclusiveLock : ShareLock;
    2965             : 
    2966             :     /*
    2967             :      * If we previously locked some other index's heap, and the name we're
    2968             :      * looking up no longer refers to that relation, release the now-useless
    2969             :      * lock.
    2970             :      */
    2971         366 :     if (relId != oldRelId && OidIsValid(oldRelId))
    2972             :     {
    2973           6 :         UnlockRelationOid(state->locked_table_oid, table_lockmode);
    2974           6 :         state->locked_table_oid = InvalidOid;
    2975             :     }
    2976             : 
    2977             :     /* If the relation does not exist, there's nothing more to do. */
    2978         366 :     if (!OidIsValid(relId))
    2979          12 :         return;
    2980             : 
    2981             :     /*
    2982             :      * If the relation does exist, check whether it's an index.  But note that
    2983             :      * the relation might have been dropped between the time we did the name
    2984             :      * lookup and now.  In that case, there's nothing to do.
    2985             :      */
    2986         354 :     relkind = get_rel_relkind(relId);
    2987         354 :     if (!relkind)
    2988           0 :         return;
    2989         354 :     if (relkind != RELKIND_INDEX &&
    2990             :         relkind != RELKIND_PARTITIONED_INDEX)
    2991          24 :         ereport(ERROR,
    2992             :                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
    2993             :                  errmsg("\"%s\" is not an index", relation->relname)));
    2994             : 
    2995             :     /* Check permissions */
    2996         330 :     table_oid = IndexGetRelation(relId, true);
    2997         330 :     if (OidIsValid(table_oid))
    2998             :     {
    2999             :         AclResult   aclresult;
    3000             : 
    3001         330 :         aclresult = pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN);
    3002         330 :         if (aclresult != ACLCHECK_OK)
    3003          12 :             aclcheck_error(aclresult, OBJECT_INDEX, relation->relname);
    3004             :     }
    3005             : 
    3006             :     /* Lock heap before index to avoid deadlock. */
    3007         318 :     if (relId != oldRelId)
    3008             :     {
    3009             :         /*
    3010             :          * If the OID isn't valid, it means the index was concurrently
    3011             :          * dropped, which is not a problem for us; just return normally.
    3012             :          */
    3013         318 :         if (OidIsValid(table_oid))
    3014             :         {
    3015         318 :             LockRelationOid(table_oid, table_lockmode);
    3016         318 :             state->locked_table_oid = table_oid;
    3017             :         }
    3018             :     }
    3019             : }
    3020             : 
    3021             : /*
    3022             :  * ReindexTable
    3023             :  *      Recreate all indexes of a table (and of its toast table, if any)
    3024             :  */
    3025             : static Oid
    3026         480 : ReindexTable(const ReindexStmt *stmt, const ReindexParams *params, bool isTopLevel)
    3027             : {
    3028             :     Oid         heapOid;
    3029             :     bool        result;
    3030         480 :     const RangeVar *relation = stmt->relation;
    3031             : 
    3032             :     /*
    3033             :      * The lock level used here should match reindex_relation().
    3034             :      *
    3035             :      * If it's a temporary table, we will perform a non-concurrent reindex,
    3036             :      * even if CONCURRENTLY was requested.  In that case, reindex_relation()
    3037             :      * will upgrade the lock, but that's OK, because other sessions can't hold
    3038             :      * locks on our temporary table.
    3039             :      */
    3040         480 :     heapOid = RangeVarGetRelidExtended(relation,
    3041         480 :                                        (params->options & REINDEXOPT_CONCURRENTLY) != 0 ?
    3042             :                                        ShareUpdateExclusiveLock : ShareLock,
    3043             :                                        0,
    3044             :                                        RangeVarCallbackMaintainsTable, NULL);
    3045             : 
    3046         434 :     if (get_rel_relkind(heapOid) == RELKIND_PARTITIONED_TABLE)
    3047          70 :         ReindexPartitions(stmt, heapOid, params, isTopLevel);
    3048         598 :     else if ((params->options & REINDEXOPT_CONCURRENTLY) != 0 &&
    3049         234 :              get_rel_persistence(heapOid) != RELPERSISTENCE_TEMP)
    3050             :     {
    3051         222 :         result = ReindexRelationConcurrently(stmt, heapOid, params);
    3052             : 
    3053         184 :         if (!result)
    3054          18 :             ereport(NOTICE,
    3055             :                     (errmsg("table \"%s\" has no indexes that can be reindexed concurrently",
    3056             :                             relation->relname)));
    3057             :     }
    3058             :     else
    3059             :     {
    3060         142 :         ReindexParams newparams = *params;
    3061             : 
    3062         142 :         newparams.options |= REINDEXOPT_REPORT_PROGRESS;
    3063         142 :         result = reindex_relation(stmt, heapOid,
    3064             :                                   REINDEX_REL_PROCESS_TOAST |
    3065             :                                   REINDEX_REL_CHECK_CONSTRAINTS,
    3066             :                                   &newparams);
    3067         110 :         if (!result)
    3068          12 :             ereport(NOTICE,
    3069             :                     (errmsg("table \"%s\" has no indexes to reindex",
    3070             :                             relation->relname)));
    3071             :     }
    3072             : 
    3073         358 :     return heapOid;
    3074             : }
    3075             : 
    3076             : /*
    3077             :  * ReindexMultipleTables
    3078             :  *      Recreate indexes of tables selected by objectName/objectKind.
    3079             :  *
    3080             :  * To reduce the probability of deadlocks, each table is reindexed in a
    3081             :  * separate transaction, so we can release the lock on it right away.
    3082             :  * That means this must not be called within a user transaction block!
    3083             :  */
    3084             : static void
    3085         176 : ReindexMultipleTables(const ReindexStmt *stmt, const ReindexParams *params)
    3086             : {
    3087             : 
    3088             :     Oid         objectOid;
    3089             :     Relation    relationRelation;
    3090             :     TableScanDesc scan;
    3091             :     ScanKeyData scan_keys[1];
    3092             :     HeapTuple   tuple;
    3093             :     MemoryContext private_context;
    3094             :     MemoryContext old;
    3095         176 :     List       *relids = NIL;
    3096             :     int         num_keys;
    3097         176 :     bool        concurrent_warning = false;
    3098         176 :     bool        tablespace_warning = false;
    3099         176 :     const char *objectName = stmt->name;
    3100         176 :     const ReindexObjectType objectKind = stmt->kind;
    3101             : 
    3102             :     Assert(objectKind == REINDEX_OBJECT_SCHEMA ||
    3103             :            objectKind == REINDEX_OBJECT_SYSTEM ||
    3104             :            objectKind == REINDEX_OBJECT_DATABASE);
    3105             : 
    3106             :     /*
    3107             :      * This matches the options enforced by the grammar, where the object name
    3108             :      * is optional for DATABASE and SYSTEM.
    3109             :      */
    3110             :     Assert(objectName || objectKind != REINDEX_OBJECT_SCHEMA);
    3111             : 
    3112         176 :     if (objectKind == REINDEX_OBJECT_SYSTEM &&
    3113          34 :         (params->options & REINDEXOPT_CONCURRENTLY) != 0)
    3114          20 :         ereport(ERROR,
    3115             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3116             :                  errmsg("cannot reindex system catalogs concurrently")));
    3117             : 
    3118             :     /*
    3119             :      * Get OID of object to reindex, being the database currently being used
    3120             :      * by session for a database or for system catalogs, or the schema defined
    3121             :      * by caller. At the same time do permission checks that need different
    3122             :      * processing depending on the object type.
    3123             :      */
    3124         156 :     if (objectKind == REINDEX_OBJECT_SCHEMA)
    3125             :     {
    3126         108 :         objectOid = get_namespace_oid(objectName, false);
    3127             : 
    3128         102 :         if (!object_ownercheck(NamespaceRelationId, objectOid, GetUserId()) &&
    3129          24 :             !has_privs_of_role(GetUserId(), ROLE_PG_MAINTAIN))
    3130          18 :             aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
    3131             :                            objectName);
    3132             :     }
    3133             :     else
    3134             :     {
    3135          48 :         objectOid = MyDatabaseId;
    3136             : 
    3137          48 :         if (objectName && strcmp(objectName, get_database_name(objectOid)) != 0)
    3138           6 :             ereport(ERROR,
    3139             :                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3140             :                      errmsg("can only reindex the currently open database")));
    3141          42 :         if (!object_ownercheck(DatabaseRelationId, objectOid, GetUserId()) &&
    3142           0 :             !has_privs_of_role(GetUserId(), ROLE_PG_MAINTAIN))
    3143           0 :             aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
    3144           0 :                            get_database_name(objectOid));
    3145             :     }
    3146             : 
    3147             :     /*
    3148             :      * Create a memory context that will survive forced transaction commits we
    3149             :      * do below.  Since it is a child of PortalContext, it will go away
    3150             :      * eventually even if we suffer an error; there's no need for special
    3151             :      * abort cleanup logic.
    3152             :      */
    3153         126 :     private_context = AllocSetContextCreate(PortalContext,
    3154             :                                             "ReindexMultipleTables",
    3155             :                                             ALLOCSET_SMALL_SIZES);
    3156             : 
    3157             :     /*
    3158             :      * Define the search keys to find the objects to reindex. For a schema, we
    3159             :      * select target relations using relnamespace, something not necessary for
    3160             :      * a database-wide operation.
    3161             :      */
    3162         126 :     if (objectKind == REINDEX_OBJECT_SCHEMA)
    3163             :     {
    3164          84 :         num_keys = 1;
    3165          84 :         ScanKeyInit(&scan_keys[0],
    3166             :                     Anum_pg_class_relnamespace,
    3167             :                     BTEqualStrategyNumber, F_OIDEQ,
    3168             :                     ObjectIdGetDatum(objectOid));
    3169             :     }
    3170             :     else
    3171          42 :         num_keys = 0;
    3172             : 
    3173             :     /*
    3174             :      * Scan pg_class to build a list of the relations we need to reindex.
    3175             :      *
    3176             :      * We only consider plain relations and materialized views here (toast
    3177             :      * rels will be processed indirectly by reindex_relation).
    3178             :      */
    3179         126 :     relationRelation = table_open(RelationRelationId, AccessShareLock);
    3180         126 :     scan = table_beginscan_catalog(relationRelation, num_keys, scan_keys);
    3181       20176 :     while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
    3182             :     {
    3183       20050 :         Form_pg_class classtuple = (Form_pg_class) GETSTRUCT(tuple);
    3184       20050 :         Oid         relid = classtuple->oid;
    3185             : 
    3186             :         /*
    3187             :          * Only regular tables and matviews can have indexes, so ignore any
    3188             :          * other kind of relation.
    3189             :          *
    3190             :          * Partitioned tables/indexes are skipped but matching leaf partitions
    3191             :          * are processed.
    3192             :          */
    3193       20050 :         if (classtuple->relkind != RELKIND_RELATION &&
    3194       16492 :             classtuple->relkind != RELKIND_MATVIEW)
    3195       16474 :             continue;
    3196             : 
    3197             :         /* Skip temp tables of other backends; we can't reindex them at all */
    3198        3576 :         if (classtuple->relpersistence == RELPERSISTENCE_TEMP &&
    3199          36 :             !isTempNamespace(classtuple->relnamespace))
    3200           0 :             continue;
    3201             : 
    3202             :         /*
    3203             :          * Check user/system classification.  SYSTEM processes all the
    3204             :          * catalogs, and DATABASE processes everything that's not a catalog.
    3205             :          */
    3206        3576 :         if (objectKind == REINDEX_OBJECT_SYSTEM &&
    3207         984 :             !IsCatalogRelationOid(relid))
    3208          88 :             continue;
    3209        5408 :         else if (objectKind == REINDEX_OBJECT_DATABASE &&
    3210        1920 :                  IsCatalogRelationOid(relid))
    3211        1792 :             continue;
    3212             : 
    3213             :         /*
    3214             :          * We already checked privileges on the database or schema, but we
    3215             :          * further restrict reindexing shared catalogs to roles with the
    3216             :          * MAINTAIN privilege on the relation.
    3217             :          */
    3218        1938 :         if (classtuple->relisshared &&
    3219         242 :             pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
    3220           0 :             continue;
    3221             : 
    3222             :         /*
    3223             :          * Skip system tables, since index_create() would reject indexing them
    3224             :          * concurrently (and it would likely fail if we tried).
    3225             :          */
    3226        2178 :         if ((params->options & REINDEXOPT_CONCURRENTLY) != 0 &&
    3227         482 :             IsCatalogRelationOid(relid))
    3228             :         {
    3229         384 :             if (!concurrent_warning)
    3230           6 :                 ereport(WARNING,
    3231             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3232             :                          errmsg("cannot reindex system catalogs concurrently, skipping all")));
    3233         384 :             concurrent_warning = true;
    3234         384 :             continue;
    3235             :         }
    3236             : 
    3237             :         /*
    3238             :          * If a new tablespace is set, check if this relation has to be
    3239             :          * skipped.
    3240             :          */
    3241        1312 :         if (OidIsValid(params->tablespaceOid))
    3242             :         {
    3243           0 :             bool        skip_rel = false;
    3244             : 
    3245             :             /*
    3246             :              * Mapped relations cannot be moved to different tablespaces (in
    3247             :              * particular this eliminates all shared catalogs.).
    3248             :              */
    3249           0 :             if (RELKIND_HAS_STORAGE(classtuple->relkind) &&
    3250           0 :                 !RelFileNumberIsValid(classtuple->relfilenode))
    3251           0 :                 skip_rel = true;
    3252             : 
    3253             :             /*
    3254             :              * A system relation is always skipped, even with
    3255             :              * allow_system_table_mods enabled.
    3256             :              */
    3257           0 :             if (IsSystemClass(relid, classtuple))
    3258           0 :                 skip_rel = true;
    3259             : 
    3260           0 :             if (skip_rel)
    3261             :             {
    3262           0 :                 if (!tablespace_warning)
    3263           0 :                     ereport(WARNING,
    3264             :                             (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
    3265             :                              errmsg("cannot move system relations, skipping all")));
    3266           0 :                 tablespace_warning = true;
    3267           0 :                 continue;
    3268             :             }
    3269             :         }
    3270             : 
    3271             :         /* Save the list of relation OIDs in private context */
    3272        1312 :         old = MemoryContextSwitchTo(private_context);
    3273             : 
    3274             :         /*
    3275             :          * We always want to reindex pg_class first if it's selected to be
    3276             :          * reindexed.  This ensures that if there is any corruption in
    3277             :          * pg_class' indexes, they will be fixed before we process any other
    3278             :          * tables.  This is critical because reindexing itself will try to
    3279             :          * update pg_class.
    3280             :          */
    3281        1312 :         if (relid == RelationRelationId)
    3282          16 :             relids = lcons_oid(relid, relids);
    3283             :         else
    3284        1296 :             relids = lappend_oid(relids, relid);
    3285             : 
    3286        1312 :         MemoryContextSwitchTo(old);
    3287             :     }
    3288         126 :     table_endscan(scan);
    3289         126 :     table_close(relationRelation, AccessShareLock);
    3290             : 
    3291             :     /*
    3292             :      * Process each relation listed in a separate transaction.  Note that this
    3293             :      * commits and then starts a new transaction immediately.
    3294             :      */
    3295         126 :     ReindexMultipleInternal(stmt, relids, params);
    3296             : 
    3297         126 :     MemoryContextDelete(private_context);
    3298         126 : }
    3299             : 
    3300             : /*
    3301             :  * Error callback specific to ReindexPartitions().
    3302             :  */
    3303             : static void
    3304          12 : reindex_error_callback(void *arg)
    3305             : {
    3306          12 :     ReindexErrorInfo *errinfo = (ReindexErrorInfo *) arg;
    3307             : 
    3308             :     Assert(RELKIND_HAS_PARTITIONS(errinfo->relkind));
    3309             : 
    3310          12 :     if (errinfo->relkind == RELKIND_PARTITIONED_TABLE)
    3311           6 :         errcontext("while reindexing partitioned table \"%s.%s\"",
    3312             :                    errinfo->relnamespace, errinfo->relname);
    3313           6 :     else if (errinfo->relkind == RELKIND_PARTITIONED_INDEX)
    3314           6 :         errcontext("while reindexing partitioned index \"%s.%s\"",
    3315             :                    errinfo->relnamespace, errinfo->relname);
    3316          12 : }
    3317             : 
    3318             : /*
    3319             :  * ReindexPartitions
    3320             :  *
    3321             :  * Reindex a set of partitions, per the partitioned index or table given
    3322             :  * by the caller.
    3323             :  */
    3324             : static void
    3325         106 : ReindexPartitions(const ReindexStmt *stmt, Oid relid, const ReindexParams *params, bool isTopLevel)
    3326             : {
    3327         106 :     List       *partitions = NIL;
    3328         106 :     char        relkind = get_rel_relkind(relid);
    3329         106 :     char       *relname = get_rel_name(relid);
    3330         106 :     char       *relnamespace = get_namespace_name(get_rel_namespace(relid));
    3331             :     MemoryContext reindex_context;
    3332             :     List       *inhoids;
    3333             :     ListCell   *lc;
    3334             :     ErrorContextCallback errcallback;
    3335             :     ReindexErrorInfo errinfo;
    3336             : 
    3337             :     Assert(RELKIND_HAS_PARTITIONS(relkind));
    3338             : 
    3339             :     /*
    3340             :      * Check if this runs in a transaction block, with an error callback to
    3341             :      * provide more context under which a problem happens.
    3342             :      */
    3343         106 :     errinfo.relname = pstrdup(relname);
    3344         106 :     errinfo.relnamespace = pstrdup(relnamespace);
    3345         106 :     errinfo.relkind = relkind;
    3346         106 :     errcallback.callback = reindex_error_callback;
    3347         106 :     errcallback.arg = (void *) &errinfo;
    3348         106 :     errcallback.previous = error_context_stack;
    3349         106 :     error_context_stack = &errcallback;
    3350             : 
    3351         106 :     PreventInTransactionBlock(isTopLevel,
    3352             :                               relkind == RELKIND_PARTITIONED_TABLE ?
    3353             :                               "REINDEX TABLE" : "REINDEX INDEX");
    3354             : 
    3355             :     /* Pop the error context stack */
    3356          94 :     error_context_stack = errcallback.previous;
    3357             : 
    3358             :     /*
    3359             :      * Create special memory context for cross-transaction storage.
    3360             :      *
    3361             :      * Since it is a child of PortalContext, it will go away eventually even
    3362             :      * if we suffer an error so there is no need for special abort cleanup
    3363             :      * logic.
    3364             :      */
    3365          94 :     reindex_context = AllocSetContextCreate(PortalContext, "Reindex",
    3366             :                                             ALLOCSET_DEFAULT_SIZES);
    3367             : 
    3368             :     /* ShareLock is enough to prevent schema modifications */
    3369          94 :     inhoids = find_all_inheritors(relid, ShareLock, NULL);
    3370             : 
    3371             :     /*
    3372             :      * The list of relations to reindex are the physical partitions of the
    3373             :      * tree so discard any partitioned table or index.
    3374             :      */
    3375         364 :     foreach(lc, inhoids)
    3376             :     {
    3377         270 :         Oid         partoid = lfirst_oid(lc);
    3378         270 :         char        partkind = get_rel_relkind(partoid);
    3379             :         MemoryContext old_context;
    3380             : 
    3381             :         /*
    3382             :          * This discards partitioned tables, partitioned indexes and foreign
    3383             :          * tables.
    3384             :          */
    3385         270 :         if (!RELKIND_HAS_STORAGE(partkind))
    3386         158 :             continue;
    3387             : 
    3388             :         Assert(partkind == RELKIND_INDEX ||
    3389             :                partkind == RELKIND_RELATION);
    3390             : 
    3391             :         /* Save partition OID */
    3392         112 :         old_context = MemoryContextSwitchTo(reindex_context);
    3393         112 :         partitions = lappend_oid(partitions, partoid);
    3394         112 :         MemoryContextSwitchTo(old_context);
    3395             :     }
    3396             : 
    3397             :     /*
    3398             :      * Process each partition listed in a separate transaction.  Note that
    3399             :      * this commits and then starts a new transaction immediately.
    3400             :      */
    3401          94 :     ReindexMultipleInternal(stmt, partitions, params);
    3402             : 
    3403             :     /*
    3404             :      * Clean up working storage --- note we must do this after
    3405             :      * StartTransactionCommand, else we might be trying to delete the active
    3406             :      * context!
    3407             :      */
    3408          94 :     MemoryContextDelete(reindex_context);
    3409          94 : }
    3410             : 
    3411             : /*
    3412             :  * ReindexMultipleInternal
    3413             :  *
    3414             :  * Reindex a list of relations, each one being processed in its own
    3415             :  * transaction.  This commits the existing transaction immediately,
    3416             :  * and starts a new transaction when finished.
    3417             :  */
    3418             : static void
    3419         220 : ReindexMultipleInternal(const ReindexStmt *stmt, const List *relids, const ReindexParams *params)
    3420             : {
    3421             :     ListCell   *l;
    3422             : 
    3423         220 :     PopActiveSnapshot();
    3424         220 :     CommitTransactionCommand();
    3425             : 
    3426        1644 :     foreach(l, relids)
    3427             :     {
    3428        1424 :         Oid         relid = lfirst_oid(l);
    3429             :         char        relkind;
    3430             :         char        relpersistence;
    3431             : 
    3432        1424 :         StartTransactionCommand();
    3433             : 
    3434             :         /* functions in indexes may want a snapshot set */
    3435        1424 :         PushActiveSnapshot(GetTransactionSnapshot());
    3436             : 
    3437             :         /* check if the relation still exists */
    3438        1424 :         if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(relid)))
    3439             :         {
    3440           4 :             PopActiveSnapshot();
    3441           4 :             CommitTransactionCommand();
    3442           4 :             continue;
    3443             :         }
    3444             : 
    3445             :         /*
    3446             :          * Check permissions except when moving to database's default if a new
    3447             :          * tablespace is chosen.  Note that this check also happens in
    3448             :          * ExecReindex(), but we do an extra check here as this runs across
    3449             :          * multiple transactions.
    3450             :          */
    3451        1420 :         if (OidIsValid(params->tablespaceOid) &&
    3452          12 :             params->tablespaceOid != MyDatabaseTableSpace)
    3453             :         {
    3454             :             AclResult   aclresult;
    3455             : 
    3456          12 :             aclresult = object_aclcheck(TableSpaceRelationId, params->tablespaceOid,
    3457             :                                         GetUserId(), ACL_CREATE);
    3458          12 :             if (aclresult != ACLCHECK_OK)
    3459           0 :                 aclcheck_error(aclresult, OBJECT_TABLESPACE,
    3460           0 :                                get_tablespace_name(params->tablespaceOid));
    3461             :         }
    3462             : 
    3463        1420 :         relkind = get_rel_relkind(relid);
    3464        1420 :         relpersistence = get_rel_persistence(relid);
    3465             : 
    3466             :         /*
    3467             :          * Partitioned tables and indexes can never be processed directly, and
    3468             :          * a list of their leaves should be built first.
    3469             :          */
    3470             :         Assert(!RELKIND_HAS_PARTITIONS(relkind));
    3471             : 
    3472        1420 :         if ((params->options & REINDEXOPT_CONCURRENTLY) != 0 &&
    3473             :             relpersistence != RELPERSISTENCE_TEMP)
    3474         122 :         {
    3475         122 :             ReindexParams newparams = *params;
    3476             : 
    3477         122 :             newparams.options |= REINDEXOPT_MISSING_OK;
    3478         122 :             (void) ReindexRelationConcurrently(stmt, relid, &newparams);
    3479         122 :             if (ActiveSnapshotSet())
    3480          26 :                 PopActiveSnapshot();
    3481             :             /* ReindexRelationConcurrently() does the verbose output */
    3482             :         }
    3483        1298 :         else if (relkind == RELKIND_INDEX)
    3484             :         {
    3485          18 :             ReindexParams newparams = *params;
    3486             : 
    3487          18 :             newparams.options |=
    3488             :                 REINDEXOPT_REPORT_PROGRESS | REINDEXOPT_MISSING_OK;
    3489          18 :             reindex_index(stmt, relid, false, relpersistence, &newparams);
    3490          18 :             PopActiveSnapshot();
    3491             :             /* reindex_index() does the verbose output */
    3492             :         }
    3493             :         else
    3494             :         {
    3495             :             bool        result;
    3496        1280 :             ReindexParams newparams = *params;
    3497             : 
    3498        1280 :             newparams.options |=
    3499             :                 REINDEXOPT_REPORT_PROGRESS | REINDEXOPT_MISSING_OK;
    3500        1280 :             result = reindex_relation(stmt, relid,
    3501             :                                       REINDEX_REL_PROCESS_TOAST |
    3502             :                                       REINDEX_REL_CHECK_CONSTRAINTS,
    3503             :                                       &newparams);
    3504             : 
    3505        1280 :             if (result && (params->options & REINDEXOPT_VERBOSE) != 0)
    3506           0 :                 ereport(INFO,
    3507             :                         (errmsg("table \"%s.%s\" was reindexed",
    3508             :                                 get_namespace_name(get_rel_namespace(relid)),
    3509             :                                 get_rel_name(relid))));
    3510             : 
    3511        1280 :             PopActiveSnapshot();
    3512             :         }
    3513             : 
    3514        1420 :         CommitTransactionCommand();
    3515             :     }
    3516             : 
    3517         220 :     StartTransactionCommand();
    3518         220 : }
    3519             : 
    3520             : 
    3521             : /*
    3522             :  * ReindexRelationConcurrently - process REINDEX CONCURRENTLY for given
    3523             :  * relation OID
    3524             :  *
    3525             :  * 'relationOid' can either belong to an index, a table or a materialized
    3526             :  * view.  For tables and materialized views, all its indexes will be rebuilt,
    3527             :  * excluding invalid indexes and any indexes used in exclusion constraints,
    3528             :  * but including its associated toast table indexes.  For indexes, the index
    3529             :  * itself will be rebuilt.
    3530             :  *
    3531             :  * The locks taken on parent tables and involved indexes are kept until the
    3532             :  * transaction is committed, at which point a session lock is taken on each
    3533             :  * relation.  Both of these protect against concurrent schema changes.
    3534             :  *
    3535             :  * Returns true if any indexes have been rebuilt (including toast table's
    3536             :  * indexes, when relevant), otherwise returns false.
    3537             :  *
    3538             :  * NOTE: This cannot be used on temporary relations.  A concurrent build would
    3539             :  * cause issues with ON COMMIT actions triggered by the transactions of the
    3540             :  * concurrent build.  Temporary relations are not subject to concurrent
    3541             :  * concerns, so there's no need for the more complicated concurrent build,
    3542             :  * anyway, and a non-concurrent reindex is more efficient.
    3543             :  */
    3544             : static bool
    3545         500 : ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const ReindexParams *params)
    3546             : {
    3547             :     typedef struct ReindexIndexInfo
    3548             :     {
    3549             :         Oid         indexId;
    3550             :         Oid         tableId;
    3551             :         Oid         amId;
    3552             :         bool        safe;       /* for set_indexsafe_procflags */
    3553             :     } ReindexIndexInfo;
    3554         500 :     List       *heapRelationIds = NIL;
    3555         500 :     List       *indexIds = NIL;
    3556         500 :     List       *newIndexIds = NIL;
    3557         500 :     List       *relationLocks = NIL;
    3558         500 :     List       *lockTags = NIL;
    3559             :     ListCell   *lc,
    3560             :                *lc2;
    3561             :     MemoryContext private_context;
    3562             :     MemoryContext oldcontext;
    3563             :     char        relkind;
    3564         500 :     char       *relationName = NULL;
    3565         500 :     char       *relationNamespace = NULL;
    3566             :     PGRUsage    ru0;
    3567         500 :     const int   progress_index[] = {
    3568             :         PROGRESS_CREATEIDX_COMMAND,
    3569             :         PROGRESS_CREATEIDX_PHASE,
    3570             :         PROGRESS_CREATEIDX_INDEX_OID,
    3571             :         PROGRESS_CREATEIDX_ACCESS_METHOD_OID
    3572             :     };
    3573             :     int64       progress_vals[4];
    3574             : 
    3575             :     /*
    3576             :      * Create a memory context that will survive forced transaction commits we
    3577             :      * do below.  Since it is a child of PortalContext, it will go away
    3578             :      * eventually even if we suffer an error; there's no need for special
    3579             :      * abort cleanup logic.
    3580             :      */
    3581         500 :     private_context = AllocSetContextCreate(PortalContext,
    3582             :                                             "ReindexConcurrent",
    3583             :                                             ALLOCSET_SMALL_SIZES);
    3584             : 
    3585         500 :     if ((params->options & REINDEXOPT_VERBOSE) != 0)
    3586             :     {
    3587             :         /* Save data needed by REINDEX VERBOSE in private context */
    3588           4 :         oldcontext = MemoryContextSwitchTo(private_context);
    3589             : 
    3590           4 :         relationName = get_rel_name(relationOid);
    3591           4 :         relationNamespace = get_namespace_name(get_rel_namespace(relationOid));
    3592             : 
    3593           4 :         pg_rusage_init(&ru0);
    3594             : 
    3595           4 :         MemoryContextSwitchTo(oldcontext);
    3596             :     }
    3597             : 
    3598         500 :     relkind = get_rel_relkind(relationOid);
    3599             : 
    3600             :     /*
    3601             :      * Extract the list of indexes that are going to be rebuilt based on the
    3602             :      * relation Oid given by caller.
    3603             :      */
    3604         500 :     switch (relkind)
    3605             :     {
    3606         320 :         case RELKIND_RELATION:
    3607             :         case RELKIND_MATVIEW:
    3608             :         case RELKIND_TOASTVALUE:
    3609             :             {
    3610             :                 /*
    3611             :                  * In the case of a relation, find all its indexes including
    3612             :                  * toast indexes.
    3613             :                  */
    3614             :                 Relation    heapRelation;
    3615             : 
    3616             :                 /* Save the list of relation OIDs in private context */
    3617         320 :                 oldcontext = MemoryContextSwitchTo(private_context);
    3618             : 
    3619             :                 /* Track this relation for session locks */
    3620         320 :                 heapRelationIds = lappend_oid(heapRelationIds, relationOid);
    3621             : 
    3622         320 :                 MemoryContextSwitchTo(oldcontext);
    3623             : 
    3624         320 :                 if (IsCatalogRelationOid(relationOid))
    3625          36 :                     ereport(ERROR,
    3626             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3627             :                              errmsg("cannot reindex system catalogs concurrently")));
    3628             : 
    3629             :                 /* Open relation to get its indexes */
    3630         284 :                 if ((params->options & REINDEXOPT_MISSING_OK) != 0)
    3631             :                 {
    3632          98 :                     heapRelation = try_table_open(relationOid,
    3633             :                                                   ShareUpdateExclusiveLock);
    3634             :                     /* leave if relation does not exist */
    3635          98 :                     if (!heapRelation)
    3636           0 :                         break;
    3637             :                 }
    3638             :                 else
    3639         186 :                     heapRelation = table_open(relationOid,
    3640             :                                               ShareUpdateExclusiveLock);
    3641             : 
    3642         306 :                 if (OidIsValid(params->tablespaceOid) &&
    3643          22 :                     IsSystemRelation(heapRelation))
    3644           2 :                     ereport(ERROR,
    3645             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3646             :                              errmsg("cannot move system relation \"%s\"",
    3647             :                                     RelationGetRelationName(heapRelation))));
    3648             : 
    3649             :                 /* Add all the valid indexes of relation to list */
    3650         544 :                 foreach(lc, RelationGetIndexList(heapRelation))
    3651             :                 {
    3652         262 :                     Oid         cellOid = lfirst_oid(lc);
    3653         262 :                     Relation    indexRelation = index_open(cellOid,
    3654             :                                                            ShareUpdateExclusiveLock);
    3655             : 
    3656         262 :                     if (!indexRelation->rd_index->indisvalid)
    3657           6 :                         ereport(WARNING,
    3658             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    3659             :                                  errmsg("skipping reindex of invalid index \"%s.%s\"",
    3660             :                                         get_namespace_name(get_rel_namespace(cellOid)),
    3661             :                                         get_rel_name(cellOid)),
    3662             :                                  errhint("Use DROP INDEX or REINDEX INDEX.")));
    3663         256 :                     else if (indexRelation->rd_index->indisexclusion)
    3664           6 :                         ereport(WARNING,
    3665             :                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3666             :                                  errmsg("cannot reindex exclusion constraint index \"%s.%s\" concurrently, skipping",
    3667             :                                         get_namespace_name(get_rel_namespace(cellOid)),
    3668             :                                         get_rel_name(cellOid))));
    3669             :                     else
    3670             :                     {
    3671             :                         ReindexIndexInfo *idx;
    3672             : 
    3673             :                         /* Save the list of relation OIDs in private context */
    3674         250 :                         oldcontext = MemoryContextSwitchTo(private_context);
    3675             : 
    3676         250 :                         idx = palloc_object(ReindexIndexInfo);
    3677         250 :                         idx->indexId = cellOid;
    3678             :                         /* other fields set later */
    3679             : 
    3680         250 :                         indexIds = lappend(indexIds, idx);
    3681             : 
    3682         250 :                         MemoryContextSwitchTo(oldcontext);
    3683             :                     }
    3684             : 
    3685         262 :                     index_close(indexRelation, NoLock);
    3686             :                 }
    3687             : 
    3688             :                 /* Also add the toast indexes */
    3689         282 :                 if (OidIsValid(heapRelation->rd_rel->reltoastrelid))
    3690             :                 {
    3691          82 :                     Oid         toastOid = heapRelation->rd_rel->reltoastrelid;
    3692          82 :                     Relation    toastRelation = table_open(toastOid,
    3693             :                                                            ShareUpdateExclusiveLock);
    3694             : 
    3695             :                     /* Save the list of relation OIDs in private context */
    3696          82 :                     oldcontext = MemoryContextSwitchTo(private_context);
    3697             : 
    3698             :                     /* Track this relation for session locks */
    3699          82 :                     heapRelationIds = lappend_oid(heapRelationIds, toastOid);
    3700             : 
    3701          82 :                     MemoryContextSwitchTo(oldcontext);
    3702             : 
    3703         164 :                     foreach(lc2, RelationGetIndexList(toastRelation))
    3704             :                     {
    3705          82 :                         Oid         cellOid = lfirst_oid(lc2);
    3706          82 :                         Relation    indexRelation = index_open(cellOid,
    3707             :                                                                ShareUpdateExclusiveLock);
    3708             : 
    3709          82 :                         if (!indexRelation->rd_index->indisvalid)
    3710           0 :                             ereport(WARNING,
    3711             :                                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    3712             :                                      errmsg("skipping reindex of invalid index \"%s.%s\"",
    3713             :                                             get_namespace_name(get_rel_namespace(cellOid)),
    3714             :                                             get_rel_name(cellOid)),
    3715             :                                      errhint("Use DROP INDEX or REINDEX INDEX.")));
    3716             :                         else
    3717             :                         {
    3718             :                             ReindexIndexInfo *idx;
    3719             : 
    3720             :                             /*
    3721             :                              * Save the list of relation OIDs in private
    3722             :                              * context
    3723             :                              */
    3724          82 :                             oldcontext = MemoryContextSwitchTo(private_context);
    3725             : 
    3726          82 :                             idx = palloc_object(ReindexIndexInfo);
    3727          82 :                             idx->indexId = cellOid;
    3728          82 :                             indexIds = lappend(indexIds, idx);
    3729             :                             /* other fields set later */
    3730             : 
    3731          82 :                             MemoryContextSwitchTo(oldcontext);
    3732             :                         }
    3733             : 
    3734          82 :                         index_close(indexRelation, NoLock);
    3735             :                     }
    3736             : 
    3737          82 :                     table_close(toastRelation, NoLock);
    3738             :                 }
    3739             : 
    3740         282 :                 table_close(heapRelation, NoLock);
    3741         282 :                 break;
    3742             :             }
    3743         180 :         case RELKIND_INDEX:
    3744             :             {
    3745         180 :                 Oid         heapId = IndexGetRelation(relationOid,
    3746         180 :                                                       (params->options & REINDEXOPT_MISSING_OK) != 0);
    3747             :                 Relation    heapRelation;
    3748             :                 ReindexIndexInfo *idx;
    3749             : 
    3750             :                 /* if relation is missing, leave */
    3751         180 :                 if (!OidIsValid(heapId))
    3752           0 :                     break;
    3753             : 
    3754         180 :                 if (IsCatalogRelationOid(heapId))
    3755          18 :                     ereport(ERROR,
    3756             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3757             :                              errmsg("cannot reindex system catalogs concurrently")));
    3758             : 
    3759             :                 /*
    3760             :                  * Don't allow reindex for an invalid index on TOAST table, as
    3761             :                  * if rebuilt it would not be possible to drop it.  Match
    3762             :                  * error message in reindex_index().
    3763             :                  */
    3764         162 :                 if (IsToastNamespace(get_rel_namespace(relationOid)) &&
    3765          56 :                     !get_index_isvalid(relationOid))
    3766           0 :                     ereport(ERROR,
    3767             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3768             :                              errmsg("cannot reindex invalid index on TOAST table")));
    3769             : 
    3770             :                 /*
    3771             :                  * Check if parent relation can be locked and if it exists,
    3772             :                  * this needs to be done at this stage as the list of indexes
    3773             :                  * to rebuild is not complete yet, and REINDEXOPT_MISSING_OK
    3774             :                  * should not be used once all the session locks are taken.
    3775             :                  */
    3776         162 :                 if ((params->options & REINDEXOPT_MISSING_OK) != 0)
    3777             :                 {
    3778          24 :                     heapRelation = try_table_open(heapId,
    3779             :                                                   ShareUpdateExclusiveLock);
    3780             :                     /* leave if relation does not exist */
    3781          24 :                     if (!heapRelation)
    3782           0 :                         break;
    3783             :                 }
    3784             :                 else
    3785         138 :                     heapRelation = table_open(heapId,
    3786             :                                               ShareUpdateExclusiveLock);
    3787             : 
    3788         170 :                 if (OidIsValid(params->tablespaceOid) &&
    3789           8 :                     IsSystemRelation(heapRelation))
    3790           2 :                     ereport(ERROR,
    3791             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3792             :                              errmsg("cannot move system relation \"%s\"",
    3793             :                                     get_rel_name(relationOid))));
    3794             : 
    3795         160 :                 table_close(heapRelation, NoLock);
    3796             : 
    3797             :                 /* Save the list of relation OIDs in private context */
    3798         160 :                 oldcontext = MemoryContextSwitchTo(private_context);
    3799             : 
    3800             :                 /* Track the heap relation of this index for session locks */
    3801         160 :                 heapRelationIds = list_make1_oid(heapId);
    3802             : 
    3803             :                 /*
    3804             :                  * Save the list of relation OIDs in private context.  Note
    3805             :                  * that invalid indexes are allowed here.
    3806             :                  */
    3807         160 :                 idx = palloc_object(ReindexIndexInfo);
    3808         160 :                 idx->indexId = relationOid;
    3809         160 :                 indexIds = lappend(indexIds, idx);
    3810             :                 /* other fields set later */
    3811             : 
    3812         160 :                 MemoryContextSwitchTo(oldcontext);
    3813         160 :                 break;
    3814             :             }
    3815             : 
    3816           0 :         case RELKIND_PARTITIONED_TABLE:
    3817             :         case RELKIND_PARTITIONED_INDEX:
    3818             :         default:
    3819             :             /* Return error if type of relation is not supported */
    3820           0 :             ereport(ERROR,
    3821             :                     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
    3822             :                      errmsg("cannot reindex this type of relation concurrently")));
    3823             :             break;
    3824             :     }
    3825             : 
    3826             :     /*
    3827             :      * Definitely no indexes, so leave.  Any checks based on
    3828             :      * REINDEXOPT_MISSING_OK should be done only while the list of indexes to
    3829             :      * work on is built as the session locks taken before this transaction
    3830             :      * commits will make sure that they cannot be dropped by a concurrent
    3831             :      * session until this operation completes.
    3832             :      */
    3833         442 :     if (indexIds == NIL)
    3834          44 :         return false;
    3835             : 
    3836             :     /* It's not a shared catalog, so refuse to move it to shared tablespace */
    3837         398 :     if (params->tablespaceOid == GLOBALTABLESPACE_OID)
    3838           6 :         ereport(ERROR,
    3839             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    3840             :                  errmsg("cannot move non-shared relation to tablespace \"%s\"",
    3841             :                         get_tablespace_name(params->tablespaceOid))));
    3842             : 
    3843             :     Assert(heapRelationIds != NIL);
    3844             : 
    3845             :     /*-----
    3846             :      * Now we have all the indexes we want to process in indexIds.
    3847             :      *
    3848             :      * The phases now are:
    3849             :      *
    3850             :      * 1. create new indexes in the catalog
    3851             :      * 2. build new indexes
    3852             :      * 3. let new indexes catch up with tuples inserted in the meantime
    3853             :      * 4. swap index names
    3854             :      * 5. mark old indexes as dead
    3855             :      * 6. drop old indexes
    3856             :      *
    3857             :      * We process each phase for all indexes before moving to the next phase,
    3858             :      * for efficiency.
    3859             :      */
    3860             : 
    3861             :     /*
    3862             :      * Phase 1 of REINDEX CONCURRENTLY
    3863             :      *
    3864             :      * Create a new index with the same properties as the old one, but it is
    3865             :      * only registered in catalogs and will be built later.  Then get session
    3866             :      * locks on all involved tables.  See analogous code in DefineIndex() for
    3867             :      * more detailed comments.
    3868             :      */
    3869             : 
    3870         872 :     foreach(lc, indexIds)
    3871             :     {
    3872             :         char       *concurrentName;
    3873         486 :         ReindexIndexInfo *idx = lfirst(lc);
    3874             :         ReindexIndexInfo *newidx;
    3875             :         Oid         newIndexId;
    3876             :         Relation    indexRel;
    3877             :         Relation    heapRel;
    3878             :         Oid         save_userid;
    3879             :         int         save_sec_context;
    3880             :         int         save_nestlevel;
    3881             :         Relation    newIndexRel;
    3882             :         LockRelId  *lockrelid;
    3883             :         Oid         tablespaceid;
    3884             : 
    3885         486 :         indexRel = index_open(idx->indexId, ShareUpdateExclusiveLock);
    3886         486 :         heapRel = table_open(indexRel->rd_index->indrelid,
    3887             :                              ShareUpdateExclusiveLock);
    3888             : 
    3889             :         /*
    3890             :          * Switch to the table owner's userid, so that any index functions are
    3891             :          * run as that user.  Also lock down security-restricted operations
    3892             :          * and arrange to make GUC variable changes local to this command.
    3893             :          */
    3894         486 :         GetUserIdAndSecContext(&save_userid, &save_sec_context);
    3895         486 :         SetUserIdAndSecContext(heapRel->rd_rel->relowner,
    3896             :                                save_sec_context | SECURITY_RESTRICTED_OPERATION);
    3897         486 :         save_nestlevel = NewGUCNestLevel();
    3898         486 :         RestrictSearchPath();
    3899             : 
    3900             :         /* determine safety of this index for set_indexsafe_procflags */
    3901         954 :         idx->safe = (indexRel->rd_indexprs == NIL &&
    3902         468 :                      indexRel->rd_indpred == NIL);
    3903         486 :         idx->tableId = RelationGetRelid(heapRel);
    3904         486 :         idx->amId = indexRel->rd_rel->relam;
    3905             : 
    3906             :         /* This function shouldn't be called for temporary relations. */
    3907         486 :         if (indexRel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
    3908           0 :             elog(ERROR, "cannot reindex a temporary table concurrently");
    3909             : 
    3910         486 :         pgstat_progress_start_command(PROGRESS_COMMAND_CREATE_INDEX, idx->tableId);
    3911             : 
    3912         486 :         progress_vals[0] = PROGRESS_CREATEIDX_COMMAND_REINDEX_CONCURRENTLY;
    3913         486 :         progress_vals[1] = 0;   /* initializing */
    3914         486 :         progress_vals[2] = idx->indexId;
    3915         486 :         progress_vals[3] = idx->amId;
    3916         486 :         pgstat_progress_update_multi_param(4, progress_index, progress_vals);
    3917             : 
    3918             :         /* Choose a temporary relation name for the new index */
    3919         486 :         concurrentName = ChooseRelationName(get_rel_name(idx->indexId),
    3920             :                                             NULL,
    3921             :                                             "ccnew",
    3922         486 :                                             get_rel_namespace(indexRel->rd_index->indrelid),
    3923             :                                             false);
    3924             : 
    3925             :         /* Choose the new tablespace, indexes of toast tables are not moved */
    3926         486 :         if (OidIsValid(params->tablespaceOid) &&
    3927          28 :             heapRel->rd_rel->relkind != RELKIND_TOASTVALUE)
    3928          20 :             tablespaceid = params->tablespaceOid;
    3929             :         else
    3930         466 :             tablespaceid = indexRel->rd_rel->reltablespace;
    3931             : 
    3932             :         /* Create new index definition based on given index */
    3933         486 :         newIndexId = index_concurrently_create_copy(heapRel,
    3934             :                                                     idx->indexId,
    3935             :                                                     tablespaceid,
    3936             :                                                     concurrentName);
    3937             : 
    3938             :         /*
    3939             :          * Now open the relation of the new index, a session-level lock is
    3940             :          * also needed on it.
    3941             :          */
    3942         480 :         newIndexRel = index_open(newIndexId, ShareUpdateExclusiveLock);
    3943             : 
    3944             :         /*
    3945             :          * Save the list of OIDs and locks in private context
    3946             :          */
    3947         480 :         oldcontext = MemoryContextSwitchTo(private_context);
    3948             : 
    3949         480 :         newidx = palloc_object(ReindexIndexInfo);
    3950         480 :         newidx->indexId = newIndexId;
    3951         480 :         newidx->safe = idx->safe;
    3952         480 :         newidx->tableId = idx->tableId;
    3953         480 :         newidx->amId = idx->amId;
    3954             : 
    3955         480 :         newIndexIds = lappend(newIndexIds, newidx);
    3956             : 
    3957             :         /*
    3958             :          * Save lockrelid to protect each relation from drop then close
    3959             :          * relations. The lockrelid on parent relation is not taken here to
    3960             :          * avoid multiple locks taken on the same relation, instead we rely on
    3961             :          * parentRelationIds built earlier.
    3962             :          */
    3963         480 :         lockrelid = palloc_object(LockRelId);
    3964         480 :         *lockrelid = indexRel->rd_lockInfo.lockRelId;
    3965         480 :         relationLocks = lappend(relationLocks, lockrelid);
    3966         480 :         lockrelid = palloc_object(LockRelId);
    3967         480 :         *lockrelid = newIndexRel->rd_lockInfo.lockRelId;
    3968         480 :         relationLocks = lappend(relationLocks, lockrelid);
    3969             : 
    3970         480 :         MemoryContextSwitchTo(oldcontext);
    3971             : 
    3972         480 :         index_close(indexRel, NoLock);
    3973         480 :         index_close(newIndexRel, NoLock);
    3974             : 
    3975             :         /* Roll back any GUC changes executed by index functions */
    3976         480 :         AtEOXact_GUC(false, save_nestlevel);
    3977             : 
    3978             :         /* Restore userid and security context */
    3979         480 :         SetUserIdAndSecContext(save_userid, save_sec_context);
    3980             : 
    3981         480 :         table_close(heapRel, NoLock);
    3982             : 
    3983             :         /*
    3984             :          * If a statement is available, telling that this comes from a REINDEX
    3985             :          * command, collect the new index for event triggers.
    3986             :          */
    3987         480 :         if (stmt)
    3988             :         {
    3989             :             ObjectAddress address;
    3990             : 
    3991         480 :             ObjectAddressSet(address, RelationRelationId, newIndexId);
    3992         480 :             EventTriggerCollectSimpleCommand(address,
    3993             :                                              InvalidObjectAddress,
    3994             :                                              (Node *) stmt);
    3995             :         }
    3996             :     }
    3997             : 
    3998             :     /*
    3999             :      * Save the heap lock for following visibility checks with other backends
    4000             :      * might conflict with this session.
    4001             :      */
    4002         854 :     foreach(lc, heapRelationIds)
    4003             :     {
    4004         468 :         Relation    heapRelation = table_open(lfirst_oid(lc), ShareUpdateExclusiveLock);
    4005             :         LockRelId  *lockrelid;
    4006             :         LOCKTAG    *heaplocktag;
    4007             : 
    4008             :         /* Save the list of locks in private context */
    4009         468 :         oldcontext = MemoryContextSwitchTo(private_context);
    4010             : 
    4011             :         /* Add lockrelid of heap relation to the list of locked relations */
    4012         468 :         lockrelid = palloc_object(LockRelId);
    4013         468 :         *lockrelid = heapRelation->rd_lockInfo.lockRelId;
    4014         468 :         relationLocks = lappend(relationLocks, lockrelid);
    4015             : 
    4016         468 :         heaplocktag = palloc_object(LOCKTAG);
    4017             : 
    4018             :         /* Save the LOCKTAG for this parent relation for the wait phase */
    4019         468 :         SET_LOCKTAG_RELATION(*heaplocktag, lockrelid->dbId, lockrelid->relId);
    4020         468 :         lockTags = lappend(lockTags, heaplocktag);
    4021             : 
    4022         468 :         MemoryContextSwitchTo(oldcontext);
    4023             : 
    4024             :         /* Close heap relation */
    4025         468 :         table_close(heapRelation, NoLock);
    4026             :     }
    4027             : 
    4028             :     /* Get a session-level lock on each table. */
    4029        1814 :     foreach(lc, relationLocks)
    4030             :     {
    4031        1428 :         LockRelId  *lockrelid = (LockRelId *) lfirst(lc);
    4032             : 
    4033        1428 :         LockRelationIdForSession(lockrelid, ShareUpdateExclusiveLock);
    4034             :     }
    4035             : 
    4036         386 :     PopActiveSnapshot();
    4037         386 :     CommitTransactionCommand();
    4038         386 :     StartTransactionCommand();
    4039             : 
    4040             :     /*
    4041             :      * Because we don't take a snapshot in this transaction, there's no need
    4042             :      * to set the PROC_IN_SAFE_IC flag here.
    4043             :      */
    4044             : 
    4045             :     /*
    4046             :      * Phase 2 of REINDEX CONCURRENTLY
    4047             :      *
    4048             :      * Build the new indexes in a separate transaction for each index to avoid
    4049             :      * having open transactions for an unnecessary long time.  But before
    4050             :      * doing that, wait until no running transactions could have the table of
    4051             :      * the index open with the old list of indexes.  See "phase 2" in
    4052             :      * DefineIndex() for more details.
    4053             :      */
    4054             : 
    4055         386 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    4056             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_1);
    4057         386 :     WaitForLockersMultiple(lockTags, ShareLock, true);
    4058         386 :     CommitTransactionCommand();
    4059             : 
    4060         860 :     foreach(lc, newIndexIds)
    4061             :     {
    4062         480 :         ReindexIndexInfo *newidx = lfirst(lc);
    4063             : 
    4064             :         /* Start new transaction for this index's concurrent build */
    4065         480 :         StartTransactionCommand();
    4066             : 
    4067             :         /*
    4068             :          * Check for user-requested abort.  This is inside a transaction so as
    4069             :          * xact.c does not issue a useless WARNING, and ensures that
    4070             :          * session-level locks are cleaned up on abort.
    4071             :          */
    4072         480 :         CHECK_FOR_INTERRUPTS();
    4073             : 
    4074             :         /* Tell concurrent indexing to ignore us, if index qualifies */
    4075         480 :         if (newidx->safe)
    4076         456 :             set_indexsafe_procflags();
    4077             : 
    4078             :         /* Set ActiveSnapshot since functions in the indexes may need it */
    4079         480 :         PushActiveSnapshot(GetTransactionSnapshot());
    4080             : 
    4081             :         /*
    4082             :          * Update progress for the index to build, with the correct parent
    4083             :          * table involved.
    4084             :          */
    4085         480 :         pgstat_progress_start_command(PROGRESS_COMMAND_CREATE_INDEX, newidx->tableId);
    4086         480 :         progress_vals[0] = PROGRESS_CREATEIDX_COMMAND_REINDEX_CONCURRENTLY;
    4087         480 :         progress_vals[1] = PROGRESS_CREATEIDX_PHASE_BUILD;
    4088         480 :         progress_vals[2] = newidx->indexId;
    4089         480 :         progress_vals[3] = newidx->amId;
    4090         480 :         pgstat_progress_update_multi_param(4, progress_index, progress_vals);
    4091             : 
    4092             :         /* Perform concurrent build of new index */
    4093         480 :         index_concurrently_build(newidx->tableId, newidx->indexId);
    4094             : 
    4095         474 :         PopActiveSnapshot();
    4096         474 :         CommitTransactionCommand();
    4097             :     }
    4098             : 
    4099         380 :     StartTransactionCommand();
    4100             : 
    4101             :     /*
    4102             :      * Because we don't take a snapshot or Xid in this transaction, there's no
    4103             :      * need to set the PROC_IN_SAFE_IC flag here.
    4104             :      */
    4105             : 
    4106             :     /*
    4107             :      * Phase 3 of REINDEX CONCURRENTLY
    4108             :      *
    4109             :      * During this phase the old indexes catch up with any new tuples that
    4110             :      * were created during the previous phase.  See "phase 3" in DefineIndex()
    4111             :      * for more details.
    4112             :      */
    4113             : 
    4114         380 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    4115             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_2);
    4116         380 :     WaitForLockersMultiple(lockTags, ShareLock, true);
    4117         380 :     CommitTransactionCommand();
    4118             : 
    4119         854 :     foreach(lc, newIndexIds)
    4120             :     {
    4121         474 :         ReindexIndexInfo *newidx = lfirst(lc);
    4122             :         TransactionId limitXmin;
    4123             :         Snapshot    snapshot;
    4124             : 
    4125         474 :         StartTransactionCommand();
    4126             : 
    4127             :         /*
    4128             :          * Check for user-requested abort.  This is inside a transaction so as
    4129             :          * xact.c does not issue a useless WARNING, and ensures that
    4130             :          * session-level locks are cleaned up on abort.
    4131             :          */
    4132         474 :         CHECK_FOR_INTERRUPTS();
    4133             : 
    4134             :         /* Tell concurrent indexing to ignore us, if index qualifies */
    4135         474 :         if (newidx->safe)
    4136         450 :             set_indexsafe_procflags();
    4137             : 
    4138             :         /*
    4139             :          * Take the "reference snapshot" that will be used by validate_index()
    4140             :          * to filter candidate tuples.
    4141             :          */
    4142         474 :         snapshot = RegisterSnapshot(GetTransactionSnapshot());
    4143         474 :         PushActiveSnapshot(snapshot);
    4144             : 
    4145             :         /*
    4146             :          * Update progress for the index to build, with the correct parent
    4147             :          * table involved.
    4148             :          */
    4149         474 :         pgstat_progress_start_command(PROGRESS_COMMAND_CREATE_INDEX, newidx->tableId);
    4150         474 :         progress_vals[0] = PROGRESS_CREATEIDX_COMMAND_REINDEX_CONCURRENTLY;
    4151         474 :         progress_vals[1] = PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXSCAN;
    4152         474 :         progress_vals[2] = newidx->indexId;
    4153         474 :         progress_vals[3] = newidx->amId;
    4154         474 :         pgstat_progress_update_multi_param(4, progress_index, progress_vals);
    4155             : 
    4156         474 :         validate_index(newidx->tableId, newidx->indexId, snapshot);
    4157             : 
    4158             :         /*
    4159             :          * We can now do away with our active snapshot, we still need to save
    4160             :          * the xmin limit to wait for older snapshots.
    4161             :          */
    4162         474 :         limitXmin = snapshot->xmin;
    4163             : 
    4164         474 :         PopActiveSnapshot();
    4165         474 :         UnregisterSnapshot(snapshot);
    4166             : 
    4167             :         /*
    4168             :          * To ensure no deadlocks, we must commit and start yet another
    4169             :          * transaction, and do our wait before any snapshot has been taken in
    4170             :          * it.
    4171             :          */
    4172         474 :         CommitTransactionCommand();
    4173         474 :         StartTransactionCommand();
    4174             : 
    4175             :         /*
    4176             :          * The index is now valid in the sense that it contains all currently
    4177             :          * interesting tuples.  But since it might not contain tuples deleted
    4178             :          * just before the reference snap was taken, we have to wait out any
    4179             :          * transactions that might have older snapshots.
    4180             :          *
    4181             :          * Because we don't take a snapshot or Xid in this transaction,
    4182             :          * there's no need to set the PROC_IN_SAFE_IC flag here.
    4183             :          */
    4184         474 :         pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    4185             :                                      PROGRESS_CREATEIDX_PHASE_WAIT_3);
    4186         474 :         WaitForOlderSnapshots(limitXmin, true);
    4187             : 
    4188         474 :         CommitTransactionCommand();
    4189             :     }
    4190             : 
    4191             :     /*
    4192             :      * Phase 4 of REINDEX CONCURRENTLY
    4193             :      *
    4194             :      * Now that the new indexes have been validated, swap each new index with
    4195             :      * its corresponding old index.
    4196             :      *
    4197             :      * We mark the new indexes as valid and the old indexes as not valid at
    4198             :      * the same time to make sure we only get constraint violations from the
    4199             :      * indexes with the correct names.
    4200             :      */
    4201             : 
    4202         380 :     StartTransactionCommand();
    4203             : 
    4204             :     /*
    4205             :      * Because this transaction only does catalog manipulations and doesn't do
    4206             :      * any index operations, we can set the PROC_IN_SAFE_IC flag here
    4207             :      * unconditionally.
    4208             :      */
    4209         380 :     set_indexsafe_procflags();
    4210             : 
    4211         854 :     forboth(lc, indexIds, lc2, newIndexIds)
    4212             :     {
    4213         474 :         ReindexIndexInfo *oldidx = lfirst(lc);
    4214         474 :         ReindexIndexInfo *newidx = lfirst(lc2);
    4215             :         char       *oldName;
    4216             : 
    4217             :         /*
    4218             :          * Check for user-requested abort.  This is inside a transaction so as
    4219             :          * xact.c does not issue a useless WARNING, and ensures that
    4220             :          * session-level locks are cleaned up on abort.
    4221             :          */
    4222         474 :         CHECK_FOR_INTERRUPTS();
    4223             : 
    4224             :         /* Choose a relation name for old index */
    4225         474 :         oldName = ChooseRelationName(get_rel_name(oldidx->indexId),
    4226             :                                      NULL,
    4227             :                                      "ccold",
    4228             :                                      get_rel_namespace(oldidx->tableId),
    4229             :                                      false);
    4230             : 
    4231             :         /*
    4232             :          * Swap old index with the new one.  This also marks the new one as
    4233             :          * valid and the old one as not valid.
    4234             :          */
    4235         474 :         index_concurrently_swap(newidx->indexId, oldidx->indexId, oldName);
    4236             : 
    4237             :         /*
    4238             :          * Invalidate the relcache for the table, so that after this commit
    4239             :          * all sessions will refresh any cached plans that might reference the
    4240             :          * index.
    4241             :          */
    4242         474 :         CacheInvalidateRelcacheByRelid(oldidx->tableId);
    4243             : 
    4244             :         /*
    4245             :          * CCI here so that subsequent iterations see the oldName in the
    4246             :          * catalog and can choose a nonconflicting name for their oldName.
    4247             :          * Otherwise, this could lead to conflicts if a table has two indexes
    4248             :          * whose names are equal for the first NAMEDATALEN-minus-a-few
    4249             :          * characters.
    4250             :          */
    4251         474 :         CommandCounterIncrement();
    4252             :     }
    4253             : 
    4254             :     /* Commit this transaction and make index swaps visible */
    4255         380 :     CommitTransactionCommand();
    4256         380 :     StartTransactionCommand();
    4257             : 
    4258             :     /*
    4259             :      * While we could set PROC_IN_SAFE_IC if all indexes qualified, there's no
    4260             :      * real need for that, because we only acquire an Xid after the wait is
    4261             :      * done, and that lasts for a very short period.
    4262             :      */
    4263             : 
    4264             :     /*
    4265             :      * Phase 5 of REINDEX CONCURRENTLY
    4266             :      *
    4267             :      * Mark the old indexes as dead.  First we must wait until no running
    4268             :      * transaction could be using the index for a query.  See also
    4269             :      * index_drop() for more details.
    4270             :      */
    4271             : 
    4272         380 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    4273             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_4);
    4274         380 :     WaitForLockersMultiple(lockTags, AccessExclusiveLock, true);
    4275             : 
    4276         854 :     foreach(lc, indexIds)
    4277             :     {
    4278         474 :         ReindexIndexInfo *oldidx = lfirst(lc);
    4279             : 
    4280             :         /*
    4281             :          * Check for user-requested abort.  This is inside a transaction so as
    4282             :          * xact.c does not issue a useless WARNING, and ensures that
    4283             :          * session-level locks are cleaned up on abort.
    4284             :          */
    4285         474 :         CHECK_FOR_INTERRUPTS();
    4286             : 
    4287         474 :         index_concurrently_set_dead(oldidx->tableId, oldidx->indexId);
    4288             :     }
    4289             : 
    4290             :     /* Commit this transaction to make the updates visible. */
    4291         380 :     CommitTransactionCommand();
    4292         380 :     StartTransactionCommand();
    4293             : 
    4294             :     /*
    4295             :      * While we could set PROC_IN_SAFE_IC if all indexes qualified, there's no
    4296             :      * real need for that, because we only acquire an Xid after the wait is
    4297             :      * done, and that lasts for a very short period.
    4298             :      */
    4299             : 
    4300             :     /*
    4301             :      * Phase 6 of REINDEX CONCURRENTLY
    4302             :      *
    4303             :      * Drop the old indexes.
    4304             :      */
    4305             : 
    4306         380 :     pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
    4307             :                                  PROGRESS_CREATEIDX_PHASE_WAIT_5);
    4308         380 :     WaitForLockersMultiple(lockTags, AccessExclusiveLock, true);
    4309             : 
    4310         380 :     PushActiveSnapshot(GetTransactionSnapshot());
    4311             : 
    4312             :     {
    4313         380 :         ObjectAddresses *objects = new_object_addresses();
    4314             : 
    4315         854 :         foreach(lc, indexIds)
    4316             :         {
    4317         474 :             ReindexIndexInfo *idx = lfirst(lc);
    4318             :             ObjectAddress object;
    4319             : 
    4320         474 :             object.classId = RelationRelationId;
    4321         474 :             object.objectId = idx->indexId;
    4322         474 :             object.objectSubId = 0;
    4323             : 
    4324         474 :             add_exact_object_address(&object, objects);
    4325             :         }
    4326             : 
    4327             :         /*
    4328             :          * Use PERFORM_DELETION_CONCURRENT_LOCK so that index_drop() uses the
    4329             :          * right lock level.
    4330             :          */
    4331         380 :         performMultipleDeletions(objects, DROP_RESTRICT,
    4332             :                                  PERFORM_DELETION_CONCURRENT_LOCK | PERFORM_DELETION_INTERNAL);
    4333             :     }
    4334             : 
    4335         380 :     PopActiveSnapshot();
    4336         380 :     CommitTransactionCommand();
    4337             : 
    4338             :     /*
    4339             :      * Finally, release the session-level lock on the table.
    4340             :      */
    4341        1790 :     foreach(lc, relationLocks)
    4342             :     {
    4343        1410 :         LockRelId  *lockrelid = (LockRelId *) lfirst(lc);
    4344             : 
    4345        1410 :         UnlockRelationIdForSession(lockrelid, ShareUpdateExclusiveLock);
    4346             :     }
    4347             : 
    4348             :     /* Start a new transaction to finish process properly */
    4349         380 :     StartTransactionCommand();
    4350             : 
    4351             :     /* Log what we did */
    4352         380 :     if ((params->options & REINDEXOPT_VERBOSE) != 0)
    4353             :     {
    4354           4 :         if (relkind == RELKIND_INDEX)
    4355           0 :             ereport(INFO,
    4356             :                     (errmsg("index \"%s.%s\" was reindexed",
    4357             :                             relationNamespace, relationName),
    4358             :                      errdetail("%s.",
    4359             :                                pg_rusage_show(&ru0))));
    4360             :         else
    4361             :         {
    4362          12 :             foreach(lc, newIndexIds)
    4363             :             {
    4364           8 :                 ReindexIndexInfo *idx = lfirst(lc);
    4365           8 :                 Oid         indOid = idx->indexId;
    4366             : 
    4367           8 :                 ereport(INFO,
    4368             :                         (errmsg("index \"%s.%s\" was reindexed",
    4369             :                                 get_namespace_name(get_rel_namespace(indOid)),
    4370             :                                 get_rel_name(indOid))));
    4371             :                 /* Don't show rusage here, since it's not per index. */
    4372             :             }
    4373             : 
    4374           4 :             ereport(INFO,
    4375             :                     (errmsg("table \"%s.%s\" was reindexed",
    4376             :                             relationNamespace, relationName),
    4377             :                      errdetail("%s.",
    4378             :                                pg_rusage_show(&ru0))));
    4379             :         }
    4380             :     }
    4381             : 
    4382         380 :     MemoryContextDelete(private_context);
    4383             : 
    4384         380 :     pgstat_progress_end_command();
    4385             : 
    4386         380 :     return true;
    4387             : }
    4388             : 
    4389             : /*
    4390             :  * Insert or delete an appropriate pg_inherits tuple to make the given index
    4391             :  * be a partition of the indicated parent index.
    4392             :  *
    4393             :  * This also corrects the pg_depend information for the affected index.
    4394             :  */
    4395             : void
    4396         846 : IndexSetParentIndex(Relation partitionIdx, Oid parentOid)
    4397             : {
    4398             :     Relation    pg_inherits;
    4399             :     ScanKeyData key[2];
    4400             :     SysScanDesc scan;
    4401         846 :     Oid         partRelid = RelationGetRelid(partitionIdx);
    4402             :     HeapTuple   tuple;
    4403             :     bool        fix_dependencies;
    4404             : 
    4405             :     /* Make sure this is an index */
    4406             :     Assert(partitionIdx->rd_rel->relkind == RELKIND_INDEX ||
    4407             :            partitionIdx->rd_rel->relkind == RELKIND_PARTITIONED_INDEX);
    4408             : 
    4409             :     /*
    4410             :      * Scan pg_inherits for rows linking our index to some parent.
    4411             :      */
    4412         846 :     pg_inherits = relation_open(InheritsRelationId, RowExclusiveLock);
    4413         846 :     ScanKeyInit(&key[0],
    4414             :                 Anum_pg_inherits_inhrelid,
    4415             :                 BTEqualStrategyNumber, F_OIDEQ,
    4416             :                 ObjectIdGetDatum(partRelid));
    4417         846 :     ScanKeyInit(&key[1],
    4418             :                 Anum_pg_inherits_inhseqno,
    4419             :                 BTEqualStrategyNumber, F_INT4EQ,
    4420             :                 Int32GetDatum(1));
    4421         846 :     scan = systable_beginscan(pg_inherits, InheritsRelidSeqnoIndexId, true,
    4422             :                               NULL, 2, key);
    4423         846 :     tuple = systable_getnext(scan);
    4424             : 
    4425         846 :     if (!HeapTupleIsValid(tuple))
    4426             :     {
    4427         544 :         if (parentOid == InvalidOid)
    4428             :         {
    4429             :             /*
    4430             :              * No pg_inherits row, and no parent wanted: nothing to do in this
    4431             :              * case.
    4432             :              */
    4433           0 :             fix_dependencies = false;
    4434             :         }
    4435             :         else
    4436             :         {
    4437         544 :             StoreSingleInheritance(partRelid, parentOid, 1);
    4438         544 :             fix_dependencies = true;
    4439             :         }
    4440             :     }
    4441             :     else
    4442             :     {
    4443         302 :         Form_pg_inherits inhForm = (Form_pg_inherits) GETSTRUCT(tuple);
    4444             : 
    4445         302 :         if (parentOid == InvalidOid)
    4446             :         {
    4447             :             /*
    4448             :              * There exists a pg_inherits row, which we want to clear; do so.
    4449             :              */
    4450         302 :             CatalogTupleDelete(pg_inherits, &tuple->t_self);
    4451         302 :             fix_dependencies = true;
    4452             :         }
    4453             :         else
    4454             :         {
    4455             :             /*
    4456             :              * A pg_inherits row exists.  If it's the same we want, then we're
    4457             :              * good; if it differs, that amounts to a corrupt catalog and
    4458             :              * should not happen.
    4459             :              */
    4460           0 :             if (inhForm->inhparent != parentOid)
    4461             :             {
    4462             :                 /* unexpected: we should not get called in this case */
    4463           0 :                 elog(ERROR, "bogus pg_inherit row: inhrelid %u inhparent %u",
    4464             :                      inhForm->inhrelid, inhForm->inhparent);
    4465             :             }
    4466             : 
    4467             :             /* already in the right state */
    4468           0 :             fix_dependencies = false;
    4469             :         }
    4470             :     }
    4471             : 
    4472             :     /* done with pg_inherits */
    4473         846 :     systable_endscan(scan);
    4474         846 :     relation_close(pg_inherits, RowExclusiveLock);
    4475             : 
    4476             :     /* set relhassubclass if an index partition has been added to the parent */
    4477         846 :     if (OidIsValid(parentOid))
    4478         544 :         SetRelationHasSubclass(parentOid, true);
    4479             : 
    4480             :     /* set relispartition correctly on the partition */
    4481         846 :     update_relispartition(partRelid, OidIsValid(parentOid));
    4482             : 
    4483         846 :     if (fix_dependencies)
    4484             :     {
    4485             :         /*
    4486             :          * Insert/delete pg_depend rows.  If setting a parent, add PARTITION
    4487             :          * dependencies on the parent index and the table; if removing a
    4488             :          * parent, delete PARTITION dependencies.
    4489             :          */
    4490         846 :         if (OidIsValid(parentOid))
    4491             :         {
    4492             :             ObjectAddress partIdx;
    4493             :             ObjectAddress parentIdx;
    4494             :             ObjectAddress partitionTbl;
    4495             : 
    4496         544 :             ObjectAddressSet(partIdx, RelationRelationId, partRelid);
    4497         544 :             ObjectAddressSet(parentIdx, RelationRelationId, parentOid);
    4498         544 :             ObjectAddressSet(partitionTbl, RelationRelationId,
    4499             :                              partitionIdx->rd_index->indrelid);
    4500         544 :             recordDependencyOn(&partIdx, &parentIdx,
    4501             :                                DEPENDENCY_PARTITION_PRI);
    4502         544 :             recordDependencyOn(&partIdx, &partitionTbl,
    4503             :                                DEPENDENCY_PARTITION_SEC);
    4504             :         }
    4505             :         else
    4506             :         {
    4507         302 :             deleteDependencyRecordsForClass(RelationRelationId, partRelid,
    4508             :                                             RelationRelationId,
    4509             :                                             DEPENDENCY_PARTITION_PRI);
    4510         302 :             deleteDependencyRecordsForClass(RelationRelationId, partRelid,
    4511             :                                             RelationRelationId,
    4512             :                                             DEPENDENCY_PARTITION_SEC);
    4513             :         }
    4514             : 
    4515             :         /* make our updates visible */
    4516         846 :         CommandCounterIncrement();
    4517             :     }
    4518         846 : }
    4519             : 
    4520             : /*
    4521             :  * Subroutine of IndexSetParentIndex to update the relispartition flag of the
    4522             :  * given index to the given value.
    4523             :  */
    4524             : static void
    4525         846 : update_relispartition(Oid relationId, bool newval)
    4526             : {
    4527             :     HeapTuple   tup;
    4528             :     Relation    classRel;
    4529             : 
    4530         846 :     classRel = table_open(RelationRelationId, RowExclusiveLock);
    4531         846 :     tup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relationId));
    4532         846 :     if (!HeapTupleIsValid(tup))
    4533           0 :         elog(ERROR, "cache lookup failed for relation %u", relationId);
    4534             :     Assert(((Form_pg_class) GETSTRUCT(tup))->relispartition != newval);
    4535         846 :     ((Form_pg_class) GETSTRUCT(tup))->relispartition = newval;
    4536         846 :     CatalogTupleUpdate(classRel, &tup->t_self, tup);
    4537         846 :     heap_freetuple(tup);
    4538         846 :     table_close(classRel, RowExclusiveLock);
    4539         846 : }
    4540             : 
    4541             : /*
    4542             :  * Set the PROC_IN_SAFE_IC flag in MyProc->statusFlags.
    4543             :  *
    4544             :  * When doing concurrent index builds, we can set this flag
    4545             :  * to tell other processes concurrently running CREATE
    4546             :  * INDEX CONCURRENTLY or REINDEX CONCURRENTLY to ignore us when
    4547             :  * doing their waits for concurrent snapshots.  On one hand it
    4548             :  * avoids pointlessly waiting for a process that's not interesting
    4549             :  * anyway; but more importantly it avoids deadlocks in some cases.
    4550             :  *
    4551             :  * This can be done safely only for indexes that don't execute any
    4552             :  * expressions that could access other tables, so index must not be
    4553             :  * expressional nor partial.  Caller is responsible for only calling
    4554             :  * this routine when that assumption holds true.
    4555             :  *
    4556             :  * (The flag is reset automatically at transaction end, so it must be
    4557             :  * set for each transaction.)
    4558             :  */
    4559             : static inline void
    4560        1592 : set_indexsafe_procflags(void)
    4561             : {
    4562             :     /*
    4563             :      * This should only be called before installing xid or xmin in MyProc;
    4564             :      * otherwise, concurrent processes could see an Xmin that moves backwards.
    4565             :      */
    4566             :     Assert(MyProc->xid == InvalidTransactionId &&
    4567             :            MyProc->xmin == InvalidTransactionId);
    4568             : 
    4569        1592 :     LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
    4570        1592 :     MyProc->statusFlags |= PROC_IN_SAFE_IC;
    4571        1592 :     ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
    4572        1592 :     LWLockRelease(ProcArrayLock);
    4573        1592 : }

Generated by: LCOV version 1.14