/*-------------------------------------------------------------------------

 *

 * index.c

 *    code to create and destroy POSTGRES index relations

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 *

 * IDENTIFICATION

 *    src/backend/catalog/index.c

 *

 *

 * INTERFACE ROUTINES

 *    index_create()      - Create a cataloged index relation

 *    index_drop()      - Removes index relation from catalogs

 *    BuildIndexInfo()    - Prepare to insert index tuples

 *    FormIndexDatum()    - Construct datum vector for one index tuple

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include <unistd.h>

#include "access/amapi.h"

#include "access/attmap.h"

#include "access/heapam.h"

#include "access/multixact.h"

#include "access/relscan.h"

#include "access/tableam.h"

#include "access/toast_compression.h"

#include "access/transam.h"

#include "access/visibilitymap.h"

#include "access/xact.h"

#include "bootstrap/bootstrap.h"

#include "catalog/binary_upgrade.h"

#include "catalog/catalog.h"

#include "catalog/dependency.h"

#include "catalog/heap.h"

#include "catalog/index.h"

#include "catalog/objectaccess.h"

#include "catalog/partition.h"

#include "catalog/pg_am.h"

#include "catalog/pg_collation.h"

#include "catalog/pg_constraint.h"

#include "catalog/pg_description.h"

#include "catalog/pg_inherits.h"

#include "catalog/pg_opclass.h"

#include "catalog/pg_operator.h"

#include "catalog/pg_tablespace.h"

#include "catalog/pg_trigger.h"

#include "catalog/pg_type.h"

#include "catalog/storage.h"

#include "catalog/storage_xlog.h"

#include "commands/event_trigger.h"

#include "commands/progress.h"

#include "commands/tablecmds.h"

#include "commands/trigger.h"

#include "executor/executor.h"

#include "miscadmin.h"

#include "nodes/makefuncs.h"

#include "nodes/nodeFuncs.h"

#include "optimizer/optimizer.h"

#include "parser/parser.h"

#include "pgstat.h"

#include "postmaster/autovacuum.h"

#include "rewrite/rewriteManip.h"

#include "storage/bufmgr.h"

#include "storage/lmgr.h"

#include "storage/predicate.h"

#include "storage/smgr.h"

#include "utils/builtins.h"

#include "utils/fmgroids.h"

#include "utils/guc.h"

#include "utils/inval.h"

#include "utils/lsyscache.h"

#include "utils/memutils.h"

#include "utils/pg_rusage.h"

#include "utils/rel.h"

#include "utils/snapmgr.h"

#include "utils/syscache.h"

#include "utils/tuplesort.h"

/* Potentially set by pg_upgrade_support functions */

Oid     binary_upgrade_next_index_pg_class_oid = InvalidOid;

RelFileNumber binary_upgrade_next_index_pg_class_relfilenumber =

InvalidRelFileNumber;

/*

 * Pointer-free representation of variables used when reindexing system

 * catalogs; we use this to propagate those values to parallel workers.

 */

typedef struct

{

  Oid     currentlyReindexedHeap;

  Oid     currentlyReindexedIndex;

  int     numPendingReindexedIndexes;

  Oid     pendingReindexedIndexes[FLEXIBLE_ARRAY_MEMBER];

} SerializedReindexState;

/* non-export function prototypes */

static bool relationHasPrimaryKey(Relation rel);

static TupleDesc ConstructTupleDescriptor(Relation heapRelation,

                      const IndexInfo *indexInfo,

                      const List *indexColNames,

                      Oid accessMethodId,

                      const Oid *collationIds,

                      const Oid *opclassIds);

static void InitializeAttributeOids(Relation indexRelation,

                  int numatts, Oid indexoid);

static void AppendAttributeTuples(Relation indexRelation, const Datum *attopts, const NullableDatum *stattargets);

static void UpdateIndexRelation(Oid indexoid, Oid heapoid,

                Oid parentIndexId,

                const IndexInfo *indexInfo,

                const Oid *collationOids,

                const Oid *opclassOids,

                const int16 *coloptions,

                bool primary,

                bool isexclusion,

                bool immediate,

                bool isvalid,

                bool isready);

static void index_update_stats(Relation rel,

                 bool hasindex,

                 double reltuples);

static void IndexCheckExclusion(Relation heapRelation,

                Relation indexRelation,

                IndexInfo *indexInfo);

static bool validate_index_callback(ItemPointer itemptr, void *opaque);

static bool ReindexIsCurrentlyProcessingIndex(Oid indexOid);

static void SetReindexProcessing(Oid heapOid, Oid indexOid);

static void ResetReindexProcessing(void);

static void SetReindexPending(List *indexes);

static void RemoveReindexPending(Oid indexOid);

/*

 * relationHasPrimaryKey

 *    See whether an existing relation has a primary key.

 *

 * Caller must have suitable lock on the relation.

 *

 * Note: we intentionally do not check indisvalid here; that's because this

 * is used to enforce the rule that there can be only one indisprimary index,

 * and we want that to be true even if said index is invalid.

 */

static bool

relationHasPrimaryKey(Relation rel)

{

  bool    result = false;

  List     *indexoidlist;

  ListCell   *indexoidscan;

  /*

   * Get the list of index OIDs for the table from the relcache, and look up

   * each one in the pg_index syscache until we find one marked primary key

   * (hopefully there isn't more than one such).

   */

  indexoidlist = RelationGetIndexList(rel);

  foreach(indexoidscan, indexoidlist)

  {

    Oid     indexoid = lfirst_oid(indexoidscan);

    HeapTuple indexTuple;

    indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));

    if (!HeapTupleIsValid(indexTuple)) /* should not happen */

      elog(ERROR, "cache lookup failed for index %u", indexoid);

    result = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary;

    ReleaseSysCache(indexTuple);

    if (result)

      break;

  }

  list_free(indexoidlist);

  return result;

}

/*

 * index_check_primary_key

 *    Apply special checks needed before creating a PRIMARY KEY index

 *

 * This processing used to be in DefineIndex(), but has been split out

 * so that it can be applied during ALTER TABLE ADD PRIMARY KEY USING INDEX.

 *

 * We check for a pre-existing primary key, and that all columns of the index

 * are simple column references (not expressions), and that all those

 * columns are marked NOT NULL.  If not, fail.

 *

 * We used to automatically change unmarked columns to NOT NULL here by doing

 * our own local ALTER TABLE command.  But that doesn't work well if we're

 * executing one subcommand of an ALTER TABLE: the operations may not get

 * performed in the right order overall.  Now we expect that the parser

 * inserted any required ALTER TABLE SET NOT NULL operations before trying

 * to create a primary-key index.

 *

 * Caller had better have at least ShareLock on the table, else the not-null

 * checking isn't trustworthy.

 */

void

index_check_primary_key(Relation heapRel,

            const IndexInfo *indexInfo,

            bool is_alter_table,

            const IndexStmt *stmt)

{

  int     i;

  /*

   * If ALTER TABLE or CREATE TABLE .. PARTITION OF, check that there isn't

   * already a PRIMARY KEY.  In CREATE TABLE for an ordinary relation, we

   * have faith that the parser rejected multiple pkey clauses; and CREATE

   * INDEX doesn't have a way to say PRIMARY KEY, so it's no problem either.

   */

  if ((is_alter_table || heapRel->rd_rel->relispartition) &&

    relationHasPrimaryKey(heapRel))

  {

    ereport(ERROR,

        (errcode(ERRCODE_INVALID_TABLE_DEFINITION),

         errmsg("multiple primary keys for table \"%s\" are not allowed",

            RelationGetRelationName(heapRel))));

  }

  /*

   * Indexes created with NULLS NOT DISTINCT cannot be used for primary key

   * constraints. While there is no direct syntax to reach here, it can be

   * done by creating a separate index and attaching it via ALTER TABLE ..

   * USING INDEX.

   */

  if (indexInfo->ii_NullsNotDistinct)

  {

    ereport(ERROR,

        (errcode(ERRCODE_INVALID_TABLE_DEFINITION),

         errmsg("primary keys cannot use NULLS NOT DISTINCT indexes")));

  }

  /*

   * Check that all of the attributes in a primary key are marked as not

   * null.  (We don't really expect to see that; it'd mean the parser messed

   * up.  But it seems wise to check anyway.)

   */

  for (i = 0; i < indexInfo->ii_NumIndexKeyAttrs; i++)

  {

    AttrNumber  attnum = indexInfo->ii_IndexAttrNumbers[i];

    HeapTuple atttuple;

    Form_pg_attribute attform;

    if (attnum == 0)

      ereport(ERROR,

          (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

           errmsg("primary keys cannot be expressions")));

    /* System attributes are never null, so no need to check */

    if (attnum < 0)

      continue;

    atttuple = SearchSysCache2(ATTNUM,

                   ObjectIdGetDatum(RelationGetRelid(heapRel)),

                   Int16GetDatum(attnum));

    if (!HeapTupleIsValid(atttuple))

      elog(ERROR, "cache lookup failed for attribute %d of relation %u",

         attnum, RelationGetRelid(heapRel));

    attform = (Form_pg_attribute) GETSTRUCT(atttuple);

    if (!attform->attnotnull)

      ereport(ERROR,

          (errcode(ERRCODE_INVALID_TABLE_DEFINITION),

           errmsg("primary key column \"%s\" is not marked NOT NULL",

              NameStr(attform->attname))));

    ReleaseSysCache(atttuple);

  }

}

/*

 *    ConstructTupleDescriptor

 *

 * Build an index tuple descriptor for a new index

 */

static TupleDesc

ConstructTupleDescriptor(Relation heapRelation,

             const IndexInfo *indexInfo,

             const List *indexColNames,

             Oid accessMethodId,

             const Oid *collationIds,

             const Oid *opclassIds)

{

  int     numatts = indexInfo->ii_NumIndexAttrs;

  int     numkeyatts = indexInfo->ii_NumIndexKeyAttrs;

  ListCell   *colnames_item = list_head(indexColNames);

  ListCell   *indexpr_item = list_head(indexInfo->ii_Expressions);

  const IndexAmRoutine *amroutine;

  TupleDesc heapTupDesc;

  TupleDesc indexTupDesc;

  int     natts;      /* #atts in heap rel --- for error checks */

  int     i;

  /* We need access to the index AM's API struct */

  amroutine = GetIndexAmRoutineByAmId(accessMethodId, false);

  /* ... and to the table's tuple descriptor */

  heapTupDesc = RelationGetDescr(heapRelation);

  natts = RelationGetForm(heapRelation)->relnatts;

  /*

   * allocate the new tuple descriptor

   */

  indexTupDesc = CreateTemplateTupleDesc(numatts);

  /*

   * Fill in the pg_attribute row.

   */

  for (i = 0; i < numatts; i++)

  {

    AttrNumber  atnum = indexInfo->ii_IndexAttrNumbers[i];

    Form_pg_attribute to = TupleDescAttr(indexTupDesc, i);

    HeapTuple tuple;

    Form_pg_type typeTup;

    Form_pg_opclass opclassTup;

    Oid     keyType;

    MemSet(to, 0, ATTRIBUTE_FIXED_PART_SIZE);

    to->attnum = i + 1;

    to->attislocal = true;

    to->attcollation = (i < numkeyatts) ? collationIds[i] : InvalidOid;

    /*

     * Set the attribute name as specified by caller.

     */

    if (colnames_item == NULL) /* shouldn't happen */

      elog(ERROR, "too few entries in colnames list");

    namestrcpy(&to->attname, (const char *) lfirst(colnames_item));

    colnames_item = lnext(indexColNames, colnames_item);

    /*

     * For simple index columns, we copy some pg_attribute fields from the

     * parent relation.  For expressions we have to look at the expression

     * result.

     */

    if (atnum != 0)

    {

      /* Simple index column */

      const FormData_pg_attribute *from;

      Assert(atnum > 0);  /* should've been caught above */

      if (atnum > natts) /* safety check */

        elog(ERROR, "invalid column number %d", atnum);

      from = TupleDescAttr(heapTupDesc,

                 AttrNumberGetAttrOffset(atnum));

      to->atttypid = from->atttypid;

      to->attlen = from->attlen;

      to->attndims = from->attndims;

      to->atttypmod = from->atttypmod;

      to->attbyval = from->attbyval;

      to->attalign = from->attalign;

      to->attstorage = from->attstorage;

      to->attcompression = from->attcompression;

    }

    else

    {

      /* Expressional index */

      Node     *indexkey;

      if (indexpr_item == NULL) /* shouldn't happen */

        elog(ERROR, "too few entries in indexprs list");

      indexkey = (Node *) lfirst(indexpr_item);

      indexpr_item = lnext(indexInfo->ii_Expressions, indexpr_item);

      /*

       * Lookup the expression type in pg_type for the type length etc.

       */

      keyType = exprType(indexkey);

      tuple = SearchSysCache1(TYPEOID, ObjectIdGetDatum(keyType));

      if (!HeapTupleIsValid(tuple))

        elog(ERROR, "cache lookup failed for type %u", keyType);

      typeTup = (Form_pg_type) GETSTRUCT(tuple);

      /*

       * Assign some of the attributes values. Leave the rest.

       */

      to->atttypid = keyType;

      to->attlen = typeTup->typlen;

      to->atttypmod = exprTypmod(indexkey);

      to->attbyval = typeTup->typbyval;

      to->attalign = typeTup->typalign;

      to->attstorage = typeTup->typstorage;

      /*

       * For expression columns, set attcompression invalid, since

       * there's no table column from which to copy the value. Whenever

       * we actually need to compress a value, we'll use whatever the

       * current value of default_toast_compression is at that point in

       * time.

       */

      to->attcompression = InvalidCompressionMethod;

      ReleaseSysCache(tuple);

      /*

       * Make sure the expression yields a type that's safe to store in

       * an index.  We need this defense because we have index opclasses

       * for pseudo-types such as "record", and the actually stored type

       * had better be safe; eg, a named composite type is okay, an

       * anonymous record type is not.  The test is the same as for

       * whether a table column is of a safe type (which is why we

       * needn't check for the non-expression case).

       */

      CheckAttributeType(NameStr(to->attname),

                 to->atttypid, to->attcollation,

                 NIL, 0);

    }

    /*

     * We do not yet have the correct relation OID for the index, so just

     * set it invalid for now.  InitializeAttributeOids() will fix it

     * later.

     */

    to->attrelid = InvalidOid;

    /*

     * Check the opclass and index AM to see if either provides a keytype

     * (overriding the attribute type).  Opclass (if exists) takes

     * precedence.

     */

    keyType = amroutine->amkeytype;

    if (i < indexInfo->ii_NumIndexKeyAttrs)

    {

      tuple = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclassIds[i]));

      if (!HeapTupleIsValid(tuple))

        elog(ERROR, "cache lookup failed for opclass %u", opclassIds[i]);

      opclassTup = (Form_pg_opclass) GETSTRUCT(tuple);

      if (OidIsValid(opclassTup->opckeytype))

        keyType = opclassTup->opckeytype;

      /*

       * If keytype is specified as ANYELEMENT, and opcintype is

       * ANYARRAY, then the attribute type must be an array (else it'd

       * not have matched this opclass); use its element type.

       *

       * We could also allow ANYCOMPATIBLE/ANYCOMPATIBLEARRAY here, but

       * there seems no need to do so; there's no reason to declare an

       * opclass as taking ANYCOMPATIBLEARRAY rather than ANYARRAY.

       */

      if (keyType == ANYELEMENTOID && opclassTup->opcintype == ANYARRAYOID)

      {

        keyType = get_base_element_type(to->atttypid);

        if (!OidIsValid(keyType))

          elog(ERROR, "could not get element type of array type %u",

             to->atttypid);

      }

      ReleaseSysCache(tuple);

    }

    /*

     * If a key type different from the heap value is specified, update

     * the type-related fields in the index tupdesc.

     */

    if (OidIsValid(keyType) && keyType != to->atttypid)

    {

      tuple = SearchSysCache1(TYPEOID, ObjectIdGetDatum(keyType));

      if (!HeapTupleIsValid(tuple))

        elog(ERROR, "cache lookup failed for type %u", keyType);

      typeTup = (Form_pg_type) GETSTRUCT(tuple);

      to->atttypid = keyType;

      to->atttypmod = -1;

      to->attlen = typeTup->typlen;

      to->attbyval = typeTup->typbyval;

      to->attalign = typeTup->typalign;

      to->attstorage = typeTup->typstorage;

      /* As above, use the default compression method in this case */

      to->attcompression = InvalidCompressionMethod;

      ReleaseSysCache(tuple);

    }

    populate_compact_attribute(indexTupDesc, i);

  }

  TupleDescFinalize(indexTupDesc);

  return indexTupDesc;

}

/* ----------------------------------------------------------------

 *    InitializeAttributeOids

 * ----------------------------------------------------------------

 */

static void

InitializeAttributeOids(Relation indexRelation,

            int numatts,

            Oid indexoid)

{

  TupleDesc tupleDescriptor;

  int     i;

  tupleDescriptor = RelationGetDescr(indexRelation);

  for (i = 0; i < numatts; i += 1)

    TupleDescAttr(tupleDescriptor, i)->attrelid = indexoid;

}

/* ----------------------------------------------------------------

 *    AppendAttributeTuples

 * ----------------------------------------------------------------

 */

static void

AppendAttributeTuples(Relation indexRelation, const Datum *attopts, const NullableDatum *stattargets)

{

  Relation  pg_attribute;

  CatalogIndexState indstate;

  TupleDesc indexTupDesc;

  FormExtraData_pg_attribute *attrs_extra = NULL;

  if (attopts)

  {

    attrs_extra = palloc0_array(FormExtraData_pg_attribute, indexRelation->rd_att->natts);

    for (int i = 0; i < indexRelation->rd_att->natts; i++)

    {

      if (attopts[i])

        attrs_extra[i].attoptions.value = attopts[i];

      else

        attrs_extra[i].attoptions.isnull = true;

      if (stattargets)

        attrs_extra[i].attstattarget = stattargets[i];

      else

        attrs_extra[i].attstattarget.isnull = true;

    }

  }

  /*

   * open the attribute relation and its indexes

   */

  pg_attribute = table_open(AttributeRelationId, RowExclusiveLock);

  indstate = CatalogOpenIndexes(pg_attribute);

  /*

   * insert data from new index's tupdesc into pg_attribute

   */

  indexTupDesc = RelationGetDescr(indexRelation);

  InsertPgAttributeTuples(pg_attribute, indexTupDesc, InvalidOid, attrs_extra, indstate);

  CatalogCloseIndexes(indstate);

  table_close(pg_attribute, RowExclusiveLock);

}

/* ----------------------------------------------------------------

 *    UpdateIndexRelation

 *

 * Construct and insert a new entry in the pg_index catalog

 * ----------------------------------------------------------------

 */

static void

UpdateIndexRelation(Oid indexoid,

          Oid heapoid,

          Oid parentIndexId,

          const IndexInfo *indexInfo,

          const Oid *collationOids,

          const Oid *opclassOids,

          const int16 *coloptions,

          bool primary,

          bool isexclusion,

          bool immediate,

          bool isvalid,

          bool isready)

{

  int2vector *indkey;

  oidvector  *indcollation;

  oidvector  *indclass;

  int2vector *indoption;

  Datum   exprsDatum;

  Datum   predDatum;

  Datum   values[Natts_pg_index];

  bool    nulls[Natts_pg_index] = {0};

  Relation  pg_index;

  HeapTuple tuple;

  int     i;

  /*

   * Copy the index key, opclass, and indoption info into arrays (should we

   * make the caller pass them like this to start with?)

   */

  indkey = buildint2vector(NULL, indexInfo->ii_NumIndexAttrs);

  for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)

    indkey->values[i] = indexInfo->ii_IndexAttrNumbers[i];

  indcollation = buildoidvector(collationOids, indexInfo->ii_NumIndexKeyAttrs);

  indclass = buildoidvector(opclassOids, indexInfo->ii_NumIndexKeyAttrs);

  indoption = buildint2vector(coloptions, indexInfo->ii_NumIndexKeyAttrs);

  /*

   * Convert the index expressions (if any) to a text datum

   */

  if (indexInfo->ii_Expressions != NIL)

  {

    char     *exprsString;

    exprsString = nodeToString(indexInfo->ii_Expressions);

    exprsDatum = CStringGetTextDatum(exprsString);

    pfree(exprsString);

  }

  else

    exprsDatum = (Datum) 0;

  /*

   * Convert the index predicate (if any) to a text datum.  Note we convert

   * implicit-AND format to normal explicit-AND for storage.

   */

  if (indexInfo->ii_Predicate != NIL)

  {

    char     *predString;

    predString = nodeToString(make_ands_explicit(indexInfo->ii_Predicate));

    predDatum = CStringGetTextDatum(predString);

    pfree(predString);

  }

  else

    predDatum = (Datum) 0;

  /*

   * open the system catalog index relation

   */

  pg_index = table_open(IndexRelationId, RowExclusiveLock);

  /*

   * Build a pg_index tuple

   */

  values[Anum_pg_index_indexrelid - 1] = ObjectIdGetDatum(indexoid);

  values[Anum_pg_index_indrelid - 1] = ObjectIdGetDatum(heapoid);

  values[Anum_pg_index_indnatts - 1] = Int16GetDatum(indexInfo->ii_NumIndexAttrs);

  values[Anum_pg_index_indnkeyatts - 1] = Int16GetDatum(indexInfo->ii_NumIndexKeyAttrs);

  values[Anum_pg_index_indisunique - 1] = BoolGetDatum(indexInfo->ii_Unique);

  values[Anum_pg_index_indnullsnotdistinct - 1] = BoolGetDatum(indexInfo->ii_NullsNotDistinct);

  values[Anum_pg_index_indisprimary - 1] = BoolGetDatum(primary);

  values[Anum_pg_index_indisexclusion - 1] = BoolGetDatum(isexclusion);

  values[Anum_pg_index_indimmediate - 1] = BoolGetDatum(immediate);

  values[Anum_pg_index_indisclustered - 1] = BoolGetDatum(false);

  values[Anum_pg_index_indisvalid - 1] = BoolGetDatum(isvalid);

  values[Anum_pg_index_indcheckxmin - 1] = BoolGetDatum(false);

  values[Anum_pg_index_indisready - 1] = BoolGetDatum(isready);

  values[Anum_pg_index_indislive - 1] = BoolGetDatum(true);

  values[Anum_pg_index_indisreplident - 1] = BoolGetDatum(false);

  values[Anum_pg_index_indkey - 1] = PointerGetDatum(indkey);

  values[Anum_pg_index_indcollation - 1] = PointerGetDatum(indcollation);

  values[Anum_pg_index_indclass - 1] = PointerGetDatum(indclass);

  values[Anum_pg_index_indoption - 1] = PointerGetDatum(indoption);

  values[Anum_pg_index_indexprs - 1] = exprsDatum;

  if (exprsDatum == (Datum) 0)

    nulls[Anum_pg_index_indexprs - 1] = true;

  values[Anum_pg_index_indpred - 1] = predDatum;

  if (predDatum == (Datum) 0)

    nulls[Anum_pg_index_indpred - 1] = true;

  tuple = heap_form_tuple(RelationGetDescr(pg_index), values, nulls);

  /*

   * insert the tuple into the pg_index catalog

   */

  CatalogTupleInsert(pg_index, tuple);

  /*

   * close the relation and free the tuple

   */

  table_close(pg_index, RowExclusiveLock);

  heap_freetuple(tuple);

}

/*

 * index_create

 *

 * heapRelation: table to build index on (suitably locked by caller)

 * indexRelationName: what it say

 * indexRelationId: normally, pass InvalidOid to let this routine

 *    generate an OID for the index.  During bootstrap this may be

 *    nonzero to specify a preselected OID.

 * parentIndexRelid: if creating an index partition, the OID of the

 *    parent index; otherwise InvalidOid.

 * parentConstraintId: if creating a constraint on a partition, the OID

 *    of the constraint in the parent; otherwise InvalidOid.

 * relFileNumber: normally, pass InvalidRelFileNumber to get new storage.

 *    May be nonzero to attach an existing valid build.

 * indexInfo: same info executor uses to insert into the index

 * indexColNames: column names to use for index (List of char *)

 * accessMethodId: OID of index AM to use

 * tableSpaceId: OID of tablespace to use

 * collationIds: array of collation OIDs, one per index column

 * opclassIds: array of index opclass OIDs, one per index column

 * coloptions: array of per-index-column indoption settings

 * reloptions: AM-specific options

 * flags: bitmask that can include any combination of these bits:

 *    INDEX_CREATE_IS_PRIMARY

 *      the index is a primary key

 *    INDEX_CREATE_ADD_CONSTRAINT:

 *      invoke index_constraint_create also

 *    INDEX_CREATE_SKIP_BUILD:

 *      skip the index_build() step for the moment; caller must do it

 *      later (typically via reindex_index())

 *    INDEX_CREATE_CONCURRENT:

 *      do not lock the table against writers.  The index will be

 *      marked "invalid" and the caller must take additional steps

 *      to fix it up.

 *    INDEX_CREATE_IF_NOT_EXISTS:

 *      do not throw an error if a relation with the same name

 *      already exists.

 *    INDEX_CREATE_PARTITIONED:

 *      create a partitioned index (table must be partitioned)

 *    INDEX_CREATE_SUPPRESS_PROGRESS:

 *      don't report progress during the index build.

 *    INDEX_CREATE_DEFERRABLE:

 *      index supports a deferrable constraint, mark it as

 *      non-immediate (indimmediate = false).

 *

 * constr_flags: flags passed to index_constraint_create

 *    (only if INDEX_CREATE_ADD_CONSTRAINT is set)

 * allow_system_table_mods: allow table to be a system catalog

 * is_internal: if true, post creation hook for new index

 * constraintId: if not NULL, receives OID of created constraint

 *

 * Returns the OID of the created index.

 *

 * NB: Caller is responsible for ensuring the user has USAGE on all types

 * indexInfo->ii_{Expressions,Predicate} depend on.

 */

Oid

index_create(Relation heapRelation,

       const char *indexRelationName,

       Oid indexRelationId,

       Oid parentIndexRelid,

       Oid parentConstraintId,

       RelFileNumber relFileNumber,

       IndexInfo *indexInfo,

       const List *indexColNames,

       Oid accessMethodId,

       Oid tableSpaceId,

       const Oid *collationIds,

       const Oid *opclassIds,

       const Datum *opclassOptions,

       const int16 *coloptions,

       const NullableDatum *stattargets,

       Datum reloptions,

       uint16 flags,

       uint16 constr_flags,

       bool allow_system_table_mods,

       bool is_internal,

       Oid *constraintId)

{

  Oid     heapRelationId = RelationGetRelid(heapRelation);

  Relation  pg_class;

  Relation  indexRelation;

  TupleDesc indexTupDesc;

  bool    shared_relation;

  bool    mapped_relation;

  bool    is_exclusion;

  Oid     namespaceId;

  int     i;

  char    relpersistence;

  bool    isprimary = (flags & INDEX_CREATE_IS_PRIMARY) != 0;

  bool    invalid = (flags & INDEX_CREATE_INVALID) != 0;

  bool    concurrent = (flags & INDEX_CREATE_CONCURRENT) != 0;

  bool    partitioned = (flags & INDEX_CREATE_PARTITIONED) != 0;

  bool    progress = (flags & INDEX_CREATE_SUPPRESS_PROGRESS) == 0;

  char    relkind;

  TransactionId relfrozenxid;

  MultiXactId relminmxid;

  bool    create_storage = !RelFileNumberIsValid(relFileNumber);

  /* constraint flags can only be set when a constraint is requested */

  Assert((constr_flags == 0) ||

       ((flags & INDEX_CREATE_ADD_CONSTRAINT) != 0));

  /* partitioned indexes must never be "built" by themselves */

  Assert(!partitioned || (flags & INDEX_CREATE_SKIP_BUILD));

  relkind = partitioned ? RELKIND_PARTITIONED_INDEX : RELKIND_INDEX;

  is_exclusion = (indexInfo->ii_ExclusionOps != NULL);

  pg_class = table_open(RelationRelationId, RowExclusiveLock);

  /*

   * The index will be in the same namespace as its parent table, and is

   * shared across databases if and only if the parent is.  Likewise, it

   * will use the relfilenumber map if and only if the parent does; and it

   * inherits the parent's relpersistence.

   */

  namespaceId = RelationGetNamespace(heapRelation);

  shared_relation = heapRelation->rd_rel->relisshared;

  mapped_relation = RelationIsMapped(heapRelation);

  relpersistence = heapRelation->rd_rel->relpersistence;

  /*

   * check parameters

   */

  if (indexInfo->ii_NumIndexAttrs < 1)

    elog(ERROR, "must index at least one column");

  if (!allow_system_table_mods &&

    IsSystemRelation(heapRelation) &&

    IsNormalProcessingMode())

    ereport(ERROR,

        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

         errmsg("user-defined indexes on system catalog tables are not supported")));

  /*

   * Btree text_pattern_ops uses texteq as the equality operator, which is

   * fine as long as the collation is deterministic; texteq then reduces to

   * bitwise equality and so it is semantically compatible with the other

   * operators and functions in that opclass.  But with a nondeterministic

   * collation, texteq could yield results that are incompatible with the

   * actual behavior of the index (which is determined by the opclass's

   * comparison function).  We prevent such problems by refusing creation of

   * an index with that opclass and a nondeterministic collation.

   *

   * The same applies to varchar_pattern_ops and bpchar_pattern_ops.  If we

   * find more cases, we might decide to create a real mechanism for marking

   * opclasses as incompatible with nondeterminism; but for now, this small

   * hack suffices.

   *

   * Another solution is to use a special operator, not texteq, as the

   * equality opclass member; but that is undesirable because it would

   * prevent index usage in many queries that work fine today.

   */

  for (i = 0; i < indexInfo->ii_NumIndexKeyAttrs; i++)

  {

    Oid     collation = collationIds[i];

    Oid     opclass = opclassIds[i];

    if (collation)

    {

      if ((opclass == TEXT_BTREE_PATTERN_OPS_OID ||

         opclass == VARCHAR_BTREE_PATTERN_OPS_OID ||

         opclass == BPCHAR_BTREE_PATTERN_OPS_OID) &&

        !get_collation_isdeterministic(collation))

      {

        HeapTuple classtup;

        classtup = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclass));

        if (!HeapTupleIsValid(classtup))

          elog(ERROR, "cache lookup failed for operator class %u", opclass);

        ereport(ERROR,

            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

             errmsg("nondeterministic collations are not supported for operator class \"%s\"",

                NameStr(((Form_pg_opclass) GETSTRUCT(classtup))->opcname))));

        ReleaseSysCache(classtup);

      }

    }

  }

  /*

   * Concurrent index build on a system catalog is unsafe because we tend to

   * release locks before committing in catalogs.

   */

  if (concurrent &&

    IsCatalogRelation(heapRelation))

    ereport(ERROR,

        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

         errmsg("concurrent index creation on system catalog tables is not supported")));

  /*

   * This case is currently not supported.  There's no way to ask for it in

   * the grammar with CREATE INDEX, but it can happen with REINDEX.

   */

  if (concurrent && is_exclusion)

    ereport(ERROR,

        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

         errmsg("concurrent index creation for exclusion constraints is not supported")));

  /*

   * We cannot allow indexing a shared relation after initdb (because

   * there's no way to make the entry in other databases' pg_class).

   */

  if (shared_relation && !IsBootstrapProcessingMode())

    ereport(ERROR,

        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),

         errmsg("shared indexes cannot be created after initdb")));

  /*

   * Shared relations must be in pg_global, too (last-ditch check)

   */

  if (shared_relation && tableSpaceId != GLOBALTABLESPACE_OID)

    elog(ERROR, "shared relations must be placed in pg_global tablespace");

  /*

   * Check for duplicate name (both as to the index, and as to the

   * associated constraint if any).  Such cases would fail on the relevant

   * catalogs' unique indexes anyway, but we prefer to give a friendlier

   * error message.

   */

  if (get_relname_relid(indexRelationName, namespaceId))

  {

    if ((flags & INDEX_CREATE_IF_NOT_EXISTS) != 0)

    {

      ereport(NOTICE,

          (errcode(ERRCODE_DUPLICATE_TABLE),

           errmsg("relation \"%s\" already exists, skipping",

              indexRelationName)));

      table_close(pg_class, RowExclusiveLock);

      return InvalidOid;

    }

    ereport(ERROR,

        (errcode(ERRCODE_DUPLICATE_TABLE),

         errmsg("relation \"%s\" already exists",

            indexRelationName)));

  }

  if ((flags & INDEX_CREATE_ADD_CONSTRAINT) != 0 &&

    ConstraintNameIsUsed(CONSTRAINT_RELATION, heapRelationId,

               indexRelationName))

  {

    /*

     * INDEX_CREATE_IF_NOT_EXISTS does not apply here, since the

     * conflicting constraint is not an index.

     */

    ereport(ERROR,

        (errcode(ERRCODE_DUPLICATE_OBJECT),

         errmsg("constraint \"%s\" for relation \"%s\" already exists",

            indexRelationName, RelationGetRelationName(heapRelation))));

  }

  /*

   * construct tuple descriptor for index tuples

   */

  indexTupDesc = ConstructTupleDescriptor(heapRelation,

                      indexInfo,

                      indexColNames,

                      accessMethodId,

                      collationIds,

                      opclassIds);

  /*

   * Allocate an OID for the index, unless we were told what to use.

   *

   * The OID will be the relfilenumber as well, so make sure it doesn't

   * collide with either pg_class OIDs or existing physical files.

   */

  if (!OidIsValid(indexRelationId))

  {

    /* Use binary-upgrade override for pg_class.oid and relfilenumber */

    if (IsBinaryUpgrade)

    {

      if (!OidIsValid(binary_upgrade_next_index_pg_class_oid))

        ereport(ERROR,

            (errcode(ERRCODE_INVALID_PARAMETER_VALUE),

             errmsg("pg_class index OID value not set when in binary upgrade mode")));

      indexRelationId = binary_upgrade_next_index_pg_class_oid;