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

 *

 * amvalidate.c

 *    Support routines for index access methods' amvalidate and

 *    amadjustmembers functions.

 *

 * Copyright (c) 2016-2025, PostgreSQL Global Development Group

 *

 *

 * IDENTIFICATION

 *    src/backend/access/index/amvalidate.c

 *

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

 */

#include "postgres.h"

#include "access/amvalidate.h"

#include "access/htup_details.h"

#include "catalog/pg_am.h"

#include "catalog/pg_amop.h"

#include "catalog/pg_amproc.h"

#include "catalog/pg_opclass.h"

#include "catalog/pg_operator.h"

#include "catalog/pg_proc.h"

#include "catalog/pg_type.h"

#include "parser/parse_coerce.h"

#include "utils/syscache.h"

/*

 * identify_opfamily_groups() returns a List of OpFamilyOpFuncGroup structs,

 * one for each combination of lefttype/righttype present in the family's

 * operator and support function lists.  If amopstrategy K is present for

 * this datatype combination, we set bit 1 << K in operatorset, and similarly

 * for the support functions.  With uint64 fields we can handle operator and

 * function numbers up to 63, which is plenty for the foreseeable future.

 *

 * The given CatCLists are expected to represent a single opfamily fetched

 * from the AMOPSTRATEGY and AMPROCNUM caches, so that they will be in

 * order by those caches' second and third cache keys, namely the datatypes.

 */

List *

identify_opfamily_groups(CatCList *oprlist, CatCList *proclist)

{

  List     *result = NIL;

  OpFamilyOpFuncGroup *thisgroup;

  Form_pg_amop oprform;

  Form_pg_amproc procform;

  int     io,

        ip;

  /* We need the lists to be ordered; should be true in normal operation */

  if (!oprlist->ordered || !proclist->ordered)

    elog(ERROR, "cannot validate operator family without ordered data");

  /*

   * Advance through the lists concurrently.  Thanks to the ordering, we

   * should see all operators and functions of a given datatype pair

   * consecutively.

   */

  thisgroup = NULL;

  io = ip = 0;

  if (io < oprlist->n_members)

  {

    oprform = (Form_pg_amop) GETSTRUCT(&oprlist->members[io]->tuple);

    io++;

  }

  else

    oprform = NULL;

  if (ip < proclist->n_members)

  {

    procform = (Form_pg_amproc) GETSTRUCT(&proclist->members[ip]->tuple);

    ip++;

  }

  else

    procform = NULL;

  while (oprform || procform)

  {

    if (oprform && thisgroup &&

      oprform->amoplefttype == thisgroup->lefttype &&

      oprform->amoprighttype == thisgroup->righttype)

    {

      /* Operator belongs to current group; include it and advance */

      /* Ignore strategy numbers outside supported range */

      if (oprform->amopstrategy > 0 && oprform->amopstrategy < 64)

        thisgroup->operatorset |= ((uint64) 1) << oprform->amopstrategy;

      if (io < oprlist->n_members)

      {

        oprform = (Form_pg_amop) GETSTRUCT(&oprlist->members[io]->tuple);

        io++;

      }

      else

        oprform = NULL;

      continue;

    }

    if (procform && thisgroup &&

      procform->amproclefttype == thisgroup->lefttype &&

      procform->amprocrighttype == thisgroup->righttype)

    {

      /* Procedure belongs to current group; include it and advance */

      /* Ignore function numbers outside supported range */

      if (procform->amprocnum > 0 && procform->amprocnum < 64)

        thisgroup->functionset |= ((uint64) 1) << procform->amprocnum;

      if (ip < proclist->n_members)

      {

        procform = (Form_pg_amproc) GETSTRUCT(&proclist->members[ip]->tuple);

        ip++;

      }

      else

        procform = NULL;

      continue;

    }

    /* Time for a new group */

    thisgroup = (OpFamilyOpFuncGroup *) palloc(sizeof(OpFamilyOpFuncGroup));

    if (oprform &&

      (!procform ||

       (oprform->amoplefttype < procform->amproclefttype ||

        (oprform->amoplefttype == procform->amproclefttype &&

         oprform->amoprighttype < procform->amprocrighttype))))

    {

      thisgroup->lefttype = oprform->amoplefttype;

      thisgroup->righttype = oprform->amoprighttype;

    }

    else

    {

      thisgroup->lefttype = procform->amproclefttype;

      thisgroup->righttype = procform->amprocrighttype;

    }

    thisgroup->operatorset = thisgroup->functionset = 0;

    result = lappend(result, thisgroup);

  }

  return result;

}

/*

 * Validate the signature (argument and result types) of an opclass support

 * function.  Return true if OK, false if not.

 *

 * The "..." represents maxargs argument-type OIDs.  If "exact" is true, they

 * must match the function arg types exactly, else only binary-coercibly.

 * In any case the function result type must match restype exactly.

 */

bool

check_amproc_signature(Oid funcid, Oid restype, bool exact,

             int minargs, int maxargs,...)

{

  bool    result = true;

  HeapTuple tp;

  Form_pg_proc procform;

  va_list   ap;

  int     i;

  tp = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));

  if (!HeapTupleIsValid(tp))

    elog(ERROR, "cache lookup failed for function %u", funcid);

  procform = (Form_pg_proc) GETSTRUCT(tp);

  if (procform->prorettype != restype || procform->proretset ||

    procform->pronargs < minargs || procform->pronargs > maxargs)

    result = false;

  va_start(ap, maxargs);

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

  {

    Oid     argtype = va_arg(ap, Oid);

    if (i >= procform->pronargs)

      continue;

    if (exact ? (argtype != procform->proargtypes.values[i]) :

      !IsBinaryCoercible(argtype, procform->proargtypes.values[i]))

      result = false;

  }

  va_end(ap);

  ReleaseSysCache(tp);

  return result;

}

/*

 * Validate the signature of an opclass options support function, that should

 * be 'void(internal)'.

 */

bool

check_amoptsproc_signature(Oid funcid)

{

  return check_amproc_signature(funcid, VOIDOID, true, 1, 1, INTERNALOID);

}

/*

 * Validate the signature (argument and result types) of an opclass operator.

 * Return true if OK, false if not.

 *

 * Currently, we can hard-wire this as accepting only binary operators.  Also,

 * we can insist on exact type matches, since the given lefttype/righttype

 * come from pg_amop and should always match the operator exactly.

 */

bool

check_amop_signature(Oid opno, Oid restype, Oid lefttype, Oid righttype)

{

  bool    result = true;

  HeapTuple tp;

  Form_pg_operator opform;

  tp = SearchSysCache1(OPEROID, ObjectIdGetDatum(opno));

  if (!HeapTupleIsValid(tp)) /* shouldn't happen */

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

  opform = (Form_pg_operator) GETSTRUCT(tp);

  if (opform->oprresult != restype || opform->oprkind != 'b' ||

    opform->oprleft != lefttype || opform->oprright != righttype)

    result = false;

  ReleaseSysCache(tp);

  return result;

}

/*

 * Get the OID of the opclass belonging to an opfamily and accepting

 * the specified type as input type.  Returns InvalidOid if no such opclass.

 *

 * If there is more than one such opclass, you get a random one of them.

 * Since that shouldn't happen, we don't waste cycles checking.

 *

 * We could look up the AM's OID from the opfamily, but all existing callers

 * know that or can get it without an extra lookup, so we make them pass it.

 */

Oid

opclass_for_family_datatype(Oid amoid, Oid opfamilyoid, Oid datatypeoid)

{

  Oid     result = InvalidOid;

  CatCList   *opclist;

  int     i;

  /*

   * We search through all the AM's opclasses to see if one matches.  This

   * is a bit inefficient but there is no better index available.  It also

   * saves making an explicit check that the opfamily belongs to the AM.

   */

  opclist = SearchSysCacheList1(CLAAMNAMENSP, ObjectIdGetDatum(amoid));

  for (i = 0; i < opclist->n_members; i++)

  {

    HeapTuple classtup = &opclist->members[i]->tuple;

    Form_pg_opclass classform = (Form_pg_opclass) GETSTRUCT(classtup);

    if (classform->opcfamily == opfamilyoid &&

      classform->opcintype == datatypeoid)

    {

      result = classform->oid;

      break;

    }

  }

  ReleaseCatCacheList(opclist);

  return result;

}

/*

 * Is the datatype a legitimate input type for the btree opfamily?

 */

bool

opfamily_can_sort_type(Oid opfamilyoid, Oid datatypeoid)

{

  return OidIsValid(opclass_for_family_datatype(BTREE_AM_OID,

                          opfamilyoid,

                          datatypeoid));

}