Quelle  amutils.c

/*-------------------------------------------------------------------------
 *
 * amutils.c
 *   SQL-level APIs related to index access methods.
 *
 * Copyright (c) 2016-2025, PostgreSQL Global Development Group
 *
 *
 * IDENTIFICATION
 *   src/backend/utils/adt/amutils.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/amapi.h"
#include "access/htup_details.h"
#include "catalog/pg_class.h"
#include "catalog/pg_index.h"
#include "utils/builtins.h"
#include "utils/syscache.h"


/* Convert string property name to enum, for efficiency */
struct am_propname
{
 const char *name;
 IndexAMProperty prop;
};

static const struct am_propname am_propnames[] =
{
 {
  "asc", AMPROP_ASC
 },
 {
  "desc", AMPROP_DESC
 },
 {
  "nulls_first", AMPROP_NULLS_FIRST
 },
 {
  "nulls_last", AMPROP_NULLS_LAST
 },
 {
  "orderable", AMPROP_ORDERABLE
 },
 {
  "distance_orderable", AMPROP_DISTANCE_ORDERABLE
 },
 {
  "returnable", AMPROP_RETURNABLE
 },
 {
  "search_array", AMPROP_SEARCH_ARRAY
 },
 {
  "search_nulls", AMPROP_SEARCH_NULLS
 },
 {
  "clusterable", AMPROP_CLUSTERABLE
 },
 {
  "index_scan", AMPROP_INDEX_SCAN
 },
 {
  "bitmap_scan", AMPROP_BITMAP_SCAN
 },
 {
  "backward_scan", AMPROP_BACKWARD_SCAN
 },
 {
  "can_order", AMPROP_CAN_ORDER
 },
 {
  "can_unique", AMPROP_CAN_UNIQUE
 },
 {
  "can_multi_col", AMPROP_CAN_MULTI_COL
 },
 {
  "can_exclude", AMPROP_CAN_EXCLUDE
 },
 {
  "can_include", AMPROP_CAN_INCLUDE
 },
};

static IndexAMProperty
lookup_prop_name(const char *name)
{
 int   i;

 for (i = 0; i < lengthof(am_propnames); i++)
 {
  if (pg_strcasecmp(am_propnames[i].name, name) == 0)
   return am_propnames[i].prop;
 }

 /* We do not throw an error, so that AMs can define their own properties */
 return AMPROP_UNKNOWN;
}

/*
 * Common code for properties that are just bit tests of indoptions.
 *
 * tuple: the pg_index heaptuple
 * attno: identify the index column to test the indoptions of.
 * guard: if false, a boolean false result is forced (saves code in caller).
 * iopt_mask: mask for interesting indoption bit.
 * iopt_expect: value for a "true" result (should be 0 or iopt_mask).
 *
 * Returns false to indicate a NULL result (for "unknown/inapplicable"),
 * otherwise sets *res to the boolean value to return.
 */
static bool
test_indoption(HeapTuple tuple, int attno, bool guard,
      int16 iopt_mask, int16 iopt_expect,
      bool *res)
{
 Datum  datum;
 int2vector *indoption;
 int16  indoption_val;

 if (!guard)
 {
  *res = false;
  return true;
 }

 datum = SysCacheGetAttrNotNull(INDEXRELID, tuple, Anum_pg_index_indoption);

 indoption = ((int2vector *) DatumGetPointer(datum));
 indoption_val = indoption->values[attno - 1];

 *res = (indoption_val & iopt_mask) == iopt_expect;

 return true;
}


/*
 * Test property of an index AM, index, or index column.
 *
 * This is common code for different SQL-level funcs, so the amoid and
 * index_oid parameters are mutually exclusive; we look up the amoid from the
 * index_oid if needed, or if no index oid is given, we're looking at AM-wide
 * properties.
 */
static Datum
indexam_property(FunctionCallInfo fcinfo,
     const char *propname,
     Oid amoid, Oid index_oid, int attno)
{
 bool  res = false;
 bool  isnull = false;
 int   natts = 0;
 IndexAMProperty prop;
 IndexAmRoutine *routine;

 /* Try to convert property name to enum (no error if not known) */
 prop = lookup_prop_name(propname);

 /* If we have an index OID, look up the AM, and get # of columns too */
 if (OidIsValid(index_oid))
 {
  HeapTuple tuple;
  Form_pg_class rd_rel;

  Assert(!OidIsValid(amoid));
  tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(index_oid));
  if (!HeapTupleIsValid(tuple))
   PG_RETURN_NULL();
  rd_rel = (Form_pg_class) GETSTRUCT(tuple);
  if (rd_rel->relkind != RELKIND_INDEX &&
   rd_rel->relkind != RELKIND_PARTITIONED_INDEX)
  {
   ReleaseSysCache(tuple);
   PG_RETURN_NULL();
  }
  amoid = rd_rel->relam;
  natts = rd_rel->relnatts;
  ReleaseSysCache(tuple);
 }

 /*
  * At this point, either index_oid == InvalidOid or it's a valid index
  * OID. Also, after this test and the one below, either attno == 0 for
  * index-wide or AM-wide tests, or it's a valid column number in a valid
  * index.
 */
 if (attno < 0 || attno > natts)
  PG_RETURN_NULL();

 /*
  * Get AM information.  If we don't have a valid AM OID, return NULL.
 */
 routine = GetIndexAmRoutineByAmId(amoid, true);
 if (routine == NULL)
  PG_RETURN_NULL();

 /*
  * If there's an AM property routine, give it a chance to override the
  * generic logic.  Proceed if it returns false.
 */
 if (routine->amproperty &&
  routine->amproperty(index_oid, attno, prop, propname,
       &res, &isnull))
 {
  if (isnull)
   PG_RETURN_NULL();
  PG_RETURN_BOOL(res);
 }

 if (attno > 0)
 {
  HeapTuple tuple;
  Form_pg_index rd_index;
  bool  iskey = true;

  /*
   * Handle column-level properties. Many of these need the pg_index row
   * (which we also need to use to check for nonkey atts) so we fetch
   * that first.
 */
  tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(index_oid));
  if (!HeapTupleIsValid(tuple))
   PG_RETURN_NULL();
  rd_index = (Form_pg_index) GETSTRUCT(tuple);

  Assert(index_oid == rd_index->indexrelid);
  Assert(attno > 0 && attno <= rd_index->indnatts);

  isnull = true;

  /*
   * If amcaninclude, we might be looking at an attno for a nonkey
   * column, for which we (generically) assume that most properties are
   * null.
 */
  if (routine->amcaninclude
   && attno > rd_index->indnkeyatts)
   iskey = false;

  switch (prop)
  {
   case AMPROP_ASC:
    if (iskey &&
     test_indoption(tuple, attno, routine->amcanorder,
           INDOPTION_DESC, 0, &res))
     isnull = false;
    break;

   case AMPROP_DESC:
    if (iskey &&
     test_indoption(tuple, attno, routine->amcanorder,
           INDOPTION_DESC, INDOPTION_DESC, &res))
     isnull = false;
    break;

   case AMPROP_NULLS_FIRST:
    if (iskey &&
     test_indoption(tuple, attno, routine->amcanorder,
           INDOPTION_NULLS_FIRST, INDOPTION_NULLS_FIRST, &res))
     isnull = false;
    break;

   case AMPROP_NULLS_LAST:
    if (iskey &&
     test_indoption(tuple, attno, routine->amcanorder,
           INDOPTION_NULLS_FIRST, 0, &res))
     isnull = false;
    break;

   case AMPROP_ORDERABLE:

    /*
     * generic assumption is that nonkey columns are not orderable
 */
    res = iskey ? routine->amcanorder : false;
    isnull = false;
    break;

   case AMPROP_DISTANCE_ORDERABLE:

    /*
     * The conditions for whether a column is distance-orderable
     * are really up to the AM (at time of writing, only GiST
     * supports it at all). The planner has its own idea based on
     * whether it finds an operator with amoppurpose 'o', but
     * getting there from just the index column type seems like a
     * lot of work. So instead we expect the AM to handle this in
     * its amproperty routine. The generic result is to return
     * false if the AM says it never supports this, or if this is
     * a nonkey column, and null otherwise (meaning we don't
     * know).
 */
    if (!iskey || !routine->amcanorderbyop)
    {
     res = false;
     isnull = false;
    }
    break;

   case AMPROP_RETURNABLE:

    /* note that we ignore iskey for this property */

    isnull = false;
    res = false;

    if (routine->amcanreturn)
    {
     /*
      * If possible, the AM should handle this test in its
      * amproperty function without opening the rel. But this
      * is the generic fallback if it does not.
 */
     Relation indexrel = index_open(index_oid, AccessShareLock);

     res = index_can_return(indexrel, attno);
     index_close(indexrel, AccessShareLock);
    }
    break;

   case AMPROP_SEARCH_ARRAY:
    if (iskey)
    {
     res = routine->amsearcharray;
     isnull = false;
    }
    break;

   case AMPROP_SEARCH_NULLS:
    if (iskey)
    {
     res = routine->amsearchnulls;
     isnull = false;
    }
    break;

   default:
    break;
  }

  ReleaseSysCache(tuple);

  if (!isnull)
   PG_RETURN_BOOL(res);
  PG_RETURN_NULL();
 }

 if (OidIsValid(index_oid))
 {
  /*
   * Handle index-level properties.  Currently, these only depend on the
   * AM, but that might not be true forever, so we make users name an
   * index not just an AM.
 */
  switch (prop)
  {
   case AMPROP_CLUSTERABLE:
    PG_RETURN_BOOL(routine->amclusterable);

   case AMPROP_INDEX_SCAN:
    PG_RETURN_BOOL(routine->amgettuple ? true : false);

   case AMPROP_BITMAP_SCAN:
    PG_RETURN_BOOL(routine->amgetbitmap ? true : false);

   case AMPROP_BACKWARD_SCAN:
    PG_RETURN_BOOL(routine->amcanbackward);

   default:
    PG_RETURN_NULL();
  }
 }

 /*
  * Handle AM-level properties (those that control what you can say in
  * CREATE INDEX).
 */
 switch (prop)
 {
  case AMPROP_CAN_ORDER:
   PG_RETURN_BOOL(routine->amcanorder);

  case AMPROP_CAN_UNIQUE:
   PG_RETURN_BOOL(routine->amcanunique);

  case AMPROP_CAN_MULTI_COL:
   PG_RETURN_BOOL(routine->amcanmulticol);

  case AMPROP_CAN_EXCLUDE:
   PG_RETURN_BOOL(routine->amgettuple ? true : false);

  case AMPROP_CAN_INCLUDE:
   PG_RETURN_BOOL(routine->amcaninclude);

  default:
   PG_RETURN_NULL();
 }
}

/*
 * Test property of an AM specified by AM OID
 */
Datum
pg_indexam_has_property(PG_FUNCTION_ARGS)
{
 Oid   amoid = PG_GETARG_OID(0);
 char    *propname = text_to_cstring(PG_GETARG_TEXT_PP(1));

 return indexam_property(fcinfo, propname, amoid, InvalidOid, 0);
}

/*
 * Test property of an index specified by index OID
 */
Datum
pg_index_has_property(PG_FUNCTION_ARGS)
{
 Oid   relid = PG_GETARG_OID(0);
 char    *propname = text_to_cstring(PG_GETARG_TEXT_PP(1));

 return indexam_property(fcinfo, propname, InvalidOid, relid, 0);
}

/*
 * Test property of an index column specified by index OID and column number
 */
Datum
pg_index_column_has_property(PG_FUNCTION_ARGS)
{
 Oid   relid = PG_GETARG_OID(0);
 int32  attno = PG_GETARG_INT32(1);
 char    *propname = text_to_cstring(PG_GETARG_TEXT_PP(2));

 /* Reject attno 0 immediately, so that attno > 0 identifies this case */
 if (attno <= 0)
  PG_RETURN_NULL();

 return indexam_property(fcinfo, propname, InvalidOid, relid, attno);
}

/*
 * Return the name of the given phase, as used for progress reporting by the
 * given AM.
 */
Datum
pg_indexam_progress_phasename(PG_FUNCTION_ARGS)
{
 Oid   amoid = PG_GETARG_OID(0);
 int32  phasenum = PG_GETARG_INT32(1);
 IndexAmRoutine *routine;
 char    *name;

 routine = GetIndexAmRoutineByAmId(amoid, true);
 if (routine == NULL || !routine->ambuildphasename)
  PG_RETURN_NULL();

 name = routine->ambuildphasename(phasenum);
 if (!name)
  PG_RETURN_NULL();

 PG_RETURN_DATUM(CStringGetTextDatum(name));
}