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

Generated by: LCOV version 1.14