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

 *

 * hashutil.c

 *    Utility code for Postgres hash implementation.

 *

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

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

 *

 *

 * IDENTIFICATION

 *    src/backend/access/hash/hashutil.c

 *

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

 */

#include "postgres.h"

#include "access/hash.h"

#include "access/reloptions.h"

#include "access/relscan.h"

#include "port/pg_bitutils.h"

#include "utils/lsyscache.h"

#include "utils/rel.h"

#define CALC_NEW_BUCKET(old_bucket, lowmask) \

      old_bucket | (lowmask + 1)

/*

 * _hash_checkqual -- does the index tuple satisfy the scan conditions?

 */

bool

_hash_checkqual(IndexScanDesc scan, IndexTuple itup)

{

  /*

   * Currently, we can't check any of the scan conditions since we do not

   * have the original index entry value to supply to the sk_func. Always

   * return true; we expect that hashgettuple already set the recheck flag

   * to make the main indexscan code do it.

   */

#ifdef NOT_USED

  TupleDesc tupdesc = RelationGetDescr(scan->indexRelation);

  ScanKey   key = scan->keyData;

  int     scanKeySize = scan->numberOfKeys;

  while (scanKeySize > 0)

  {

    Datum   datum;

    bool    isNull;

    Datum   test;

    datum = index_getattr(itup,

                key->sk_attno,

                tupdesc,

                &isNull);

    /* assume sk_func is strict */

    if (isNull)

      return false;

    if (key->sk_flags & SK_ISNULL)

      return false;

    test = FunctionCall2Coll(&key->sk_func, key->sk_collation,

                 datum, key->sk_argument);

    if (!DatumGetBool(test))

      return false;

    key++;

    scanKeySize--;

  }

#endif

  return true;

}

/*

 * _hash_datum2hashkey -- given a Datum, call the index's hash function

 *

 * The Datum is assumed to be of the index's column type, so we can use the

 * "primary" hash function that's tracked for us by the generic index code.

 */

uint32

_hash_datum2hashkey(Relation rel, Datum key)

{

  FmgrInfo   *procinfo;

  Oid     collation;

  /* XXX assumes index has only one attribute */

  procinfo = index_getprocinfo(rel, 1, HASHSTANDARD_PROC);

  collation = rel->rd_indcollation[0];

  return DatumGetUInt32(FunctionCall1Coll(procinfo, collation, key));

}

/*

 * _hash_datum2hashkey_type -- given a Datum of a specified type,

 *      hash it in a fashion compatible with this index

 *

 * This is much more expensive than _hash_datum2hashkey, so use it only in

 * cross-type situations.

 */

uint32

_hash_datum2hashkey_type(Relation rel, Datum key, Oid keytype)

{

  RegProcedure hash_proc;

  Oid     collation;

  /* XXX assumes index has only one attribute */

  hash_proc = get_opfamily_proc(rel->rd_opfamily[0],

                  keytype,

                  keytype,

                  HASHSTANDARD_PROC);

  if (!RegProcedureIsValid(hash_proc))

    elog(ERROR, "missing support function %d(%u,%u) for index \"%s\"",

       HASHSTANDARD_PROC, keytype, keytype,

       RelationGetRelationName(rel));

  collation = rel->rd_indcollation[0];

  return DatumGetUInt32(OidFunctionCall1Coll(hash_proc, collation, key));

}

/*

 * _hash_hashkey2bucket -- determine which bucket the hashkey maps to.

 */

Bucket

_hash_hashkey2bucket(uint32 hashkey, uint32 maxbucket,

           uint32 highmask, uint32 lowmask)

{

  Bucket    bucket;

  bucket = hashkey & highmask;

  if (bucket > maxbucket)

    bucket = bucket & lowmask;

  return bucket;

}

/*

 * _hash_spareindex -- returns spare index / global splitpoint phase of the

 *             bucket

 */

uint32

_hash_spareindex(uint32 num_bucket)

{

  uint32    splitpoint_group;

  uint32    splitpoint_phases;

  splitpoint_group = pg_ceil_log2_32(num_bucket);

  if (splitpoint_group < HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE)

    return splitpoint_group;

  /* account for single-phase groups */

  splitpoint_phases = HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE;

  /* account for multi-phase groups before splitpoint_group */

  splitpoint_phases +=

    ((splitpoint_group - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) <<

     HASH_SPLITPOINT_PHASE_BITS);

  /* account for phases within current group */

  splitpoint_phases +=

    (((num_bucket - 1) >>

      (splitpoint_group - (HASH_SPLITPOINT_PHASE_BITS + 1))) &

     HASH_SPLITPOINT_PHASE_MASK);  /* to 0-based value. */

  return splitpoint_phases;

}

/*

 *  _hash_get_totalbuckets -- returns total number of buckets allocated till

 *              the given splitpoint phase.

 */

uint32

_hash_get_totalbuckets(uint32 splitpoint_phase)

{

  uint32    splitpoint_group;

  uint32    total_buckets;

  uint32    phases_within_splitpoint_group;

  if (splitpoint_phase < HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE)

    return (1 << splitpoint_phase);

  /* get splitpoint's group */

  splitpoint_group = HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE;

  splitpoint_group +=

    ((splitpoint_phase - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) >>

     HASH_SPLITPOINT_PHASE_BITS);

  /* account for buckets before splitpoint_group */

  total_buckets = (1 << (splitpoint_group - 1));

  /* account for buckets within splitpoint_group */

  phases_within_splitpoint_group =

    (((splitpoint_phase - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) &

      HASH_SPLITPOINT_PHASE_MASK) + 1); /* from 0-based to 1-based */

  total_buckets +=

    (((1 << (splitpoint_group - 1)) >> HASH_SPLITPOINT_PHASE_BITS) *

     phases_within_splitpoint_group);

  return total_buckets;

}

/*

 * _hash_checkpage -- sanity checks on the format of all hash pages

 *

 * If flags is not zero, it is a bitwise OR of the acceptable page types

 * (values of hasho_flag & LH_PAGE_TYPE).

 */

void

_hash_checkpage(Relation rel, Buffer buf, int flags)

{

  Page    page = BufferGetPage(buf);

  /*

   * ReadBuffer verifies that every newly-read page passes

   * PageHeaderIsValid, which means it either contains a reasonably sane

   * page header or is all-zero.  We have to defend against the all-zero

   * case, however.

   */

  if (PageIsNew(page))

    ereport(ERROR,

        (errcode(ERRCODE_INDEX_CORRUPTED),

         errmsg("index \"%s\" contains unexpected zero page at block %u",

            RelationGetRelationName(rel),

            BufferGetBlockNumber(buf)),

         errhint("Please REINDEX it.")));

  /*

   * Additionally check that the special area looks sane.

   */

  if (PageGetSpecialSize(page) != MAXALIGN(sizeof(HashPageOpaqueData)))

    ereport(ERROR,

        (errcode(ERRCODE_INDEX_CORRUPTED),

         errmsg("index \"%s\" contains corrupted page at block %u",

            RelationGetRelationName(rel),

            BufferGetBlockNumber(buf)),

         errhint("Please REINDEX it.")));

  if (flags)

  {

    HashPageOpaque opaque = HashPageGetOpaque(page);

    if ((opaque->hasho_flag & flags) == 0)

      ereport(ERROR,

          (errcode(ERRCODE_INDEX_CORRUPTED),

           errmsg("index \"%s\" contains corrupted page at block %u",

              RelationGetRelationName(rel),

              BufferGetBlockNumber(buf)),

           errhint("Please REINDEX it.")));

  }

  /*

   * When checking the metapage, also verify magic number and version.

   */

  if (flags == LH_META_PAGE)

  {

    HashMetaPage metap = HashPageGetMeta(page);

    if (metap->hashm_magic != HASH_MAGIC)

      ereport(ERROR,

          (errcode(ERRCODE_INDEX_CORRUPTED),

           errmsg("index \"%s\" is not a hash index",

              RelationGetRelationName(rel))));

    if (metap->hashm_version != HASH_VERSION)

      ereport(ERROR,

          (errcode(ERRCODE_INDEX_CORRUPTED),

           errmsg("index \"%s\" has wrong hash version",

              RelationGetRelationName(rel)),

           errhint("Please REINDEX it.")));

  }

}

bytea *

hashoptions(Datum reloptions, bool validate)

{

  static const relopt_parse_elt tab[] = {

    {"fillfactor", RELOPT_TYPE_INT, offsetof(HashOptions, fillfactor)},

  };

  return (bytea *) build_reloptions(reloptions, validate,

                    RELOPT_KIND_HASH,

                    sizeof(HashOptions),

                    tab, lengthof(tab));

}

/*

 * _hash_get_indextuple_hashkey - get the hash index tuple's hash key value

 */

uint32

_hash_get_indextuple_hashkey(IndexTuple itup)

{

  char     *attp;

  /*

   * We assume the hash key is the first attribute and can't be null, so

   * this can be done crudely but very very cheaply ...

   */

  attp = (char *) itup + IndexInfoFindDataOffset(itup->t_info);

  return *((uint32 *) attp);

}

/*

 * _hash_convert_tuple - convert raw index data to hash key

 *

 * Inputs: values and isnull arrays for the user data column(s)

 * Outputs: values and isnull arrays for the index tuple, suitable for

 *    passing to index_form_tuple().

 *

 * Returns true if successful, false if not (because there are null values).

 * On a false result, the given data need not be indexed.

 *

 * Note: callers know that the index-column arrays are always of length 1.

 * In principle, there could be more than one input column, though we do not

 * currently support that.

 */

bool

_hash_convert_tuple(Relation index,

          const Datum *user_values, const bool *user_isnull,

          Datum *index_values, bool *index_isnull)

{

  uint32    hashkey;

  /*

   * We do not insert null values into hash indexes.  This is okay because

   * the only supported search operator is '=', and we assume it is strict.

   */

  if (user_isnull[0])

    return false;

  hashkey = _hash_datum2hashkey(index, user_values[0]);

  index_values[0] = UInt32GetDatum(hashkey);

  index_isnull[0] = false;

  return true;

}

/*

 * _hash_binsearch - Return the offset number in the page where the

 *           specified hash value should be sought or inserted.

 *

 * We use binary search, relying on the assumption that the existing entries

 * are ordered by hash key.

 *

 * Returns the offset of the first index entry having hashkey >= hash_value,

 * or the page's max offset plus one if hash_value is greater than all

 * existing hash keys in the page.  This is the appropriate place to start

 * a search, or to insert a new item.

 */

OffsetNumber

_hash_binsearch(Page page, uint32 hash_value)

{

  OffsetNumber upper;

  OffsetNumber lower;

  /* Loop invariant: lower <= desired place <= upper */

  upper = PageGetMaxOffsetNumber(page) + 1;

  lower = FirstOffsetNumber;

  while (upper > lower)

  {

    OffsetNumber off;

    IndexTuple  itup;

    uint32    hashkey;

    off = (upper + lower) / 2;

    Assert(OffsetNumberIsValid(off));

    itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, off));

    hashkey = _hash_get_indextuple_hashkey(itup);

    if (hashkey < hash_value)

      lower = off + 1;

    else

      upper = off;

  }

  return lower;

}

/*

 * _hash_binsearch_last

 *

 * Same as above, except that if there are multiple matching items in the

 * page, we return the offset of the last one instead of the first one,

 * and the possible range of outputs is 0..maxoffset not 1..maxoffset+1.

 * This is handy for starting a new page in a backwards scan.

 */

OffsetNumber

_hash_binsearch_last(Page page, uint32 hash_value)

{

  OffsetNumber upper;

  OffsetNumber lower;

  /* Loop invariant: lower <= desired place <= upper */

  upper = PageGetMaxOffsetNumber(page);

  lower = FirstOffsetNumber - 1;

  while (upper > lower)

  {

    IndexTuple  itup;

    OffsetNumber off;

    uint32    hashkey;

    off = (upper + lower + 1) / 2;

    Assert(OffsetNumberIsValid(off));

    itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, off));

    hashkey = _hash_get_indextuple_hashkey(itup);

    if (hashkey > hash_value)

      upper = off - 1;

    else

      lower = off;

  }

  return lower;

}

/*

 *  _hash_get_oldblock_from_newbucket() -- get the block number of a bucket

 *      from which current (new) bucket is being split.

 */

BlockNumber

_hash_get_oldblock_from_newbucket(Relation rel, Bucket new_bucket)

{

  Bucket    old_bucket;

  uint32    mask;

  Buffer    metabuf;

  HashMetaPage metap;

  BlockNumber blkno;

  /*

   * To get the old bucket from the current bucket, we need a mask to modulo

   * into lower half of table.  This mask is stored in meta page as

   * hashm_lowmask, but here we can't rely on the same, because we need a

   * value of lowmask that was prevalent at the time when bucket split was

   * started.  Masking the most significant bit of new bucket would give us

   * old bucket.

   */

  mask = (((uint32) 1) << pg_leftmost_one_pos32(new_bucket)) - 1;

  old_bucket = new_bucket & mask;

  metabuf = _hash_getbuf(rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);

  metap = HashPageGetMeta(BufferGetPage(metabuf));

  blkno = BUCKET_TO_BLKNO(metap, old_bucket);

  _hash_relbuf(rel, metabuf);

  return blkno;

}

/*

 *  _hash_get_newblock_from_oldbucket() -- get the block number of a bucket

 *      that will be generated after split from old bucket.

 *

 * This is used to find the new bucket from old bucket based on current table

 * half.  It is mainly required to finish the incomplete splits where we are

 * sure that not more than one bucket could have split in progress from old

 * bucket.

 */

BlockNumber

_hash_get_newblock_from_oldbucket(Relation rel, Bucket old_bucket)

{

  Bucket    new_bucket;

  Buffer    metabuf;

  HashMetaPage metap;

  BlockNumber blkno;

  metabuf = _hash_getbuf(rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);

  metap = HashPageGetMeta(BufferGetPage(metabuf));

  new_bucket = _hash_get_newbucket_from_oldbucket(rel, old_bucket,

                          metap->hashm_lowmask,

                          metap->hashm_maxbucket);

  blkno = BUCKET_TO_BLKNO(metap, new_bucket);

  _hash_relbuf(rel, metabuf);

  return blkno;

}

/*

 *  _hash_get_newbucket_from_oldbucket() -- get the new bucket that will be

 *      generated after split from current (old) bucket.

 *

 * This is used to find the new bucket from old bucket.  New bucket can be

 * obtained by OR'ing old bucket with most significant bit of current table

 * half (lowmask passed in this function can be used to identify msb of

 * current table half).  There could be multiple buckets that could have

 * been split from current bucket.  We need the first such bucket that exists.

 * Caller must ensure that no more than one split has happened from old

 * bucket.

 */

Bucket

_hash_get_newbucket_from_oldbucket(Relation rel, Bucket old_bucket,

                   uint32 lowmask, uint32 maxbucket)

{

  Bucket    new_bucket;

  new_bucket = CALC_NEW_BUCKET(old_bucket, lowmask);

  if (new_bucket > maxbucket)

  {

    lowmask = lowmask >> 1;

    new_bucket = CALC_NEW_BUCKET(old_bucket, lowmask);

  }

  return new_bucket;

}

/*

 * _hash_kill_items - set LP_DEAD state for items an indexscan caller has

 * told us were killed.

 *

 * scan->opaque, referenced locally through so, contains information about the

 * current page and killed tuples thereon (generally, this should only be

 * called if so->numKilled > 0).

 *

 * The caller does not have a lock on the page and may or may not have the

 * page pinned in a buffer.  Note that read-lock is sufficient for setting

 * LP_DEAD status (which is only a hint).

 *

 * The caller must have pin on bucket buffer, but may or may not have pin

 * on overflow buffer, as indicated by HashScanPosIsPinned(so->currPos).

 *

 * We match items by heap TID before assuming they are the right ones to

 * delete.

 *

 * There are never any scans active in a bucket at the time VACUUM begins,

 * because VACUUM takes a cleanup lock on the primary bucket page and scans

 * hold a pin.  A scan can begin after VACUUM leaves the primary bucket page

 * but before it finishes the entire bucket, but it can never pass VACUUM,

 * because VACUUM always locks the next page before releasing the lock on

 * the previous one.  Therefore, we don't have to worry about accidentally

 * killing a TID that has been reused for an unrelated tuple.

 */

void

_hash_kill_items(IndexScanDesc scan)

{

  HashScanOpaque so = (HashScanOpaque) scan->opaque;

  Relation  rel = scan->indexRelation;

  BlockNumber blkno;

  Buffer    buf;

  Page    page;

  HashPageOpaque opaque;

  OffsetNumber offnum,

        maxoff;

  int     numKilled = so->numKilled;

  int     i;

  bool    killedsomething = false;

  bool    havePin = false;

  Assert(so->numKilled > 0);

  Assert(so->killedItems != NULL);

  Assert(HashScanPosIsValid(so->currPos));

  /*

   * Always reset the scan state, so we don't look for same items on other

   * pages.

   */

  so->numKilled = 0;

  blkno = so->currPos.currPage;

  if (HashScanPosIsPinned(so->currPos))

  {

    /*

     * We already have pin on this buffer, so, all we need to do is

     * acquire lock on it.

     */

    havePin = true;

    buf = so->currPos.buf;

    LockBuffer(buf, BUFFER_LOCK_SHARE);

  }

  else

    buf = _hash_getbuf(rel, blkno, HASH_READ, LH_OVERFLOW_PAGE);

  page = BufferGetPage(buf);

  opaque = HashPageGetOpaque(page);

  maxoff = PageGetMaxOffsetNumber(page);

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

  {

    int     itemIndex = so->killedItems[i];

    HashScanPosItem *currItem = &so->currPos.items[itemIndex];

    offnum = currItem->indexOffset;

    Assert(itemIndex >= so->currPos.firstItem &&

         itemIndex <= so->currPos.lastItem);

    while (offnum <= maxoff)

    {

      ItemId    iid = PageGetItemId(page, offnum);

      IndexTuple  ituple = (IndexTuple) PageGetItem(page, iid);

      if (ItemPointerEquals(&ituple->t_tid, &currItem->heapTid))

      {

        if (!killedsomething)

        {

          /*

           * Use the hint bit infrastructure to check if we can

           * update the page while just holding a share lock. If we

           * are not allowed, there's no point continuing.

           */

          if (!BufferBeginSetHintBits(buf))

            goto unlock_page;

        }

        /* found the item */

        ItemIdMarkDead(iid);

        killedsomething = true;

        break;     /* out of inner search loop */

      }

      offnum = OffsetNumberNext(offnum);

    }

  }

  /*

   * Since this can be redone later if needed, mark as dirty hint. Whenever

   * we mark anything LP_DEAD, we also set the page's

   * LH_PAGE_HAS_DEAD_TUPLES flag, which is likewise just a hint.

   */

  if (killedsomething)

  {

    opaque->hasho_flag |= LH_PAGE_HAS_DEAD_TUPLES;

    BufferFinishSetHintBits(buf, true, true);

  }

unlock_page:

  if (havePin)

    LockBuffer(so->currPos.buf, BUFFER_LOCK_UNLOCK);

  else

    _hash_relbuf(rel, buf);

}