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

 *

 * hash.c

 *    Implementation of Margo Seltzer's Hashing package for postgres.

 *

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

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

 *

 *

 * IDENTIFICATION

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

 *

 * NOTES

 *    This file contains only the public interface routines.

 *

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

 */

#include "postgres.h"

#include "access/hash.h"

#include "access/hash_xlog.h"

#include "access/relscan.h"

#include "access/stratnum.h"

#include "access/tableam.h"

#include "access/xloginsert.h"

#include "commands/progress.h"

#include "commands/vacuum.h"

#include "miscadmin.h"

#include "nodes/execnodes.h"

#include "optimizer/plancat.h"

#include "pgstat.h"

#include "utils/fmgrprotos.h"

#include "utils/index_selfuncs.h"

#include "utils/rel.h"

/* Working state for hashbuild and its callback */

typedef struct

{

  HSpool     *spool;      /* NULL if not using spooling */

  double    indtuples;    /* # tuples accepted into index */

  Relation  heapRel;    /* heap relation descriptor */

} HashBuildState;

static void hashbuildCallback(Relation index,

                ItemPointer tid,

                Datum *values,

                bool *isnull,

                bool tupleIsAlive,

                void *state);

/*

 * Hash handler function: return IndexAmRoutine with access method parameters

 * and callbacks.

 */

Datum

hashhandler(PG_FUNCTION_ARGS)

{

  IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);

  amroutine->amstrategies = HTMaxStrategyNumber;

  amroutine->amsupport = HASHNProcs;

  amroutine->amoptsprocnum = HASHOPTIONS_PROC;

  amroutine->amcanorder = false;

  amroutine->amcanorderbyop = false;

  amroutine->amcanhash = true;

  amroutine->amconsistentequality = true;

  amroutine->amconsistentordering = false;

  amroutine->amcanbackward = true;

  amroutine->amcanunique = false;

  amroutine->amcanmulticol = false;

  amroutine->amoptionalkey = false;

  amroutine->amsearcharray = false;

  amroutine->amsearchnulls = false;

  amroutine->amstorage = false;

  amroutine->amclusterable = false;

  amroutine->ampredlocks = true;

  amroutine->amcanparallel = false;

  amroutine->amcanbuildparallel = false;

  amroutine->amcaninclude = false;

  amroutine->amusemaintenanceworkmem = false;

  amroutine->amsummarizing = false;

  amroutine->amparallelvacuumoptions =

    VACUUM_OPTION_PARALLEL_BULKDEL;

  amroutine->amkeytype = INT4OID;

  amroutine->ambuild = hashbuild;

  amroutine->ambuildempty = hashbuildempty;

  amroutine->aminsert = hashinsert;

  amroutine->aminsertcleanup = NULL;

  amroutine->ambulkdelete = hashbulkdelete;

  amroutine->amvacuumcleanup = hashvacuumcleanup;

  amroutine->amcanreturn = NULL;

  amroutine->amcostestimate = hashcostestimate;

  amroutine->amgettreeheight = NULL;

  amroutine->amoptions = hashoptions;

  amroutine->amproperty = NULL;

  amroutine->ambuildphasename = NULL;

  amroutine->amvalidate = hashvalidate;

  amroutine->amadjustmembers = hashadjustmembers;

  amroutine->ambeginscan = hashbeginscan;

  amroutine->amrescan = hashrescan;

  amroutine->amgettuple = hashgettuple;

  amroutine->amgetbitmap = hashgetbitmap;

  amroutine->amendscan = hashendscan;

  amroutine->ammarkpos = NULL;

  amroutine->amrestrpos = NULL;

  amroutine->amestimateparallelscan = NULL;

  amroutine->aminitparallelscan = NULL;

  amroutine->amparallelrescan = NULL;

  amroutine->amtranslatestrategy = hashtranslatestrategy;

  amroutine->amtranslatecmptype = hashtranslatecmptype;

  PG_RETURN_POINTER(amroutine);

}

/*

 *  hashbuild() -- build a new hash index.

 */

IndexBuildResult *

hashbuild(Relation heap, Relation index, IndexInfo *indexInfo)

{

  IndexBuildResult *result;

  BlockNumber relpages;

  double    reltuples;

  double    allvisfrac;

  uint32    num_buckets;

  Size    sort_threshold;

  HashBuildState buildstate;

  /*

   * We expect to be called exactly once for any index relation. If that's

   * not the case, big trouble's what we have.

   */

  if (RelationGetNumberOfBlocks(index) != 0)

    elog(ERROR, "index \"%s\" already contains data",

       RelationGetRelationName(index));

  /* Estimate the number of rows currently present in the table */

  estimate_rel_size(heap, NULL, &relpages, &reltuples, &allvisfrac);

  /* Initialize the hash index metadata page and initial buckets */

  num_buckets = _hash_init(index, reltuples, MAIN_FORKNUM);

  /*

   * If we just insert the tuples into the index in scan order, then

   * (assuming their hash codes are pretty random) there will be no locality

   * of access to the index, and if the index is bigger than available RAM

   * then we'll thrash horribly.  To prevent that scenario, we can sort the

   * tuples by (expected) bucket number.  However, such a sort is useless

   * overhead when the index does fit in RAM.  We choose to sort if the

   * initial index size exceeds maintenance_work_mem, or the number of

   * buffers usable for the index, whichever is less.  (Limiting by the

   * number of buffers should reduce thrashing between PG buffers and kernel

   * buffers, which seems useful even if no physical I/O results.  Limiting

   * by maintenance_work_mem is useful to allow easy testing of the sort

   * code path, and may be useful to DBAs as an additional control knob.)

   *

   * NOTE: this test will need adjustment if a bucket is ever different from

   * one page.  Also, "initial index size" accounting does not include the

   * metapage, nor the first bitmap page.

   */

  sort_threshold = (maintenance_work_mem * (Size) 1024) / BLCKSZ;

  if (index->rd_rel->relpersistence != RELPERSISTENCE_TEMP)

    sort_threshold = Min(sort_threshold, NBuffers);

  else

    sort_threshold = Min(sort_threshold, NLocBuffer);

  if (num_buckets >= sort_threshold)

    buildstate.spool = _h_spoolinit(heap, index, num_buckets);

  else

    buildstate.spool = NULL;

  /* prepare to build the index */

  buildstate.indtuples = 0;

  buildstate.heapRel = heap;

  /* do the heap scan */

  reltuples = table_index_build_scan(heap, index, indexInfo, true, true,

                     hashbuildCallback,

                     &buildstate, NULL);

  pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_TOTAL,

                 buildstate.indtuples);

  if (buildstate.spool)

  {

    /* sort the tuples and insert them into the index */

    _h_indexbuild(buildstate.spool, buildstate.heapRel);

    _h_spooldestroy(buildstate.spool);

  }

  /*

   * Return statistics

   */

  result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));

  result->heap_tuples = reltuples;

  result->index_tuples = buildstate.indtuples;

  return result;

}

/*

 *  hashbuildempty() -- build an empty hash index in the initialization fork

 */

void

hashbuildempty(Relation index)

{

  _hash_init(index, 0, INIT_FORKNUM);

}

/*

 * Per-tuple callback for table_index_build_scan

 */

static void

hashbuildCallback(Relation index,

          ItemPointer tid,

          Datum *values,

          bool *isnull,

          bool tupleIsAlive,

          void *state)

{

  HashBuildState *buildstate = (HashBuildState *) state;

  Datum   index_values[1];

  bool    index_isnull[1];

  IndexTuple  itup;

  /* convert data to a hash key; on failure, do not insert anything */

  if (!_hash_convert_tuple(index,

               values, isnull,

               index_values, index_isnull))

    return;

  /* Either spool the tuple for sorting, or just put it into the index */

  if (buildstate->spool)

    _h_spool(buildstate->spool, tid, index_values, index_isnull);

  else

  {

    /* form an index tuple and point it at the heap tuple */

    itup = index_form_tuple(RelationGetDescr(index),

                index_values, index_isnull);

    itup->t_tid = *tid;

    _hash_doinsert(index, itup, buildstate->heapRel, false);

    pfree(itup);

  }

  buildstate->indtuples += 1;

}

/*

 *  hashinsert() -- insert an index tuple into a hash table.

 *

 *  Hash on the heap tuple's key, form an index tuple with hash code.

 *  Find the appropriate location for the new tuple, and put it there.

 */

bool

hashinsert(Relation rel, Datum *values, bool *isnull,

       ItemPointer ht_ctid, Relation heapRel,

       IndexUniqueCheck checkUnique,

       bool indexUnchanged,

       IndexInfo *indexInfo)

{

  Datum   index_values[1];

  bool    index_isnull[1];

  IndexTuple  itup;

  /* convert data to a hash key; on failure, do not insert anything */

  if (!_hash_convert_tuple(rel,

               values, isnull,

               index_values, index_isnull))

    return false;

  /* form an index tuple and point it at the heap tuple */

  itup = index_form_tuple(RelationGetDescr(rel), index_values, index_isnull);

  itup->t_tid = *ht_ctid;

  _hash_doinsert(rel, itup, heapRel, false);

  pfree(itup);

  return false;

}

/*

 *  hashgettuple() -- Get the next tuple in the scan.

 */

bool

hashgettuple(IndexScanDesc scan, ScanDirection dir)

{

  HashScanOpaque so = (HashScanOpaque) scan->opaque;

  bool    res;

  /* Hash indexes are always lossy since we store only the hash code */

  scan->xs_recheck = true;

  /*

   * If we've already initialized this scan, we can just advance it in the

   * appropriate direction.  If we haven't done so yet, we call a routine to

   * get the first item in the scan.

   */

  if (!HashScanPosIsValid(so->currPos))

    res = _hash_first(scan, dir);

  else

  {

    /*

     * Check to see if we should kill the previously-fetched tuple.

     */

    if (scan->kill_prior_tuple)

    {

      /*

       * Yes, so remember it for later. (We'll deal with all such tuples

       * at once right after leaving the index page or at end of scan.)

       * In case if caller reverses the indexscan direction it is quite

       * possible that the same item might get entered multiple times.

       * But, we don't detect that; instead, we just forget any excess

       * entries.

       */

      if (so->killedItems == NULL)

        so->killedItems = (int *)

          palloc(MaxIndexTuplesPerPage * sizeof(int));

      if (so->numKilled < MaxIndexTuplesPerPage)

        so->killedItems[so->numKilled++] = so->currPos.itemIndex;

    }

    /*

     * Now continue the scan.

     */

    res = _hash_next(scan, dir);

  }

  return res;

}

/*

 *  hashgetbitmap() -- get all tuples at once

 */

int64

hashgetbitmap(IndexScanDesc scan, TIDBitmap *tbm)

{

  HashScanOpaque so = (HashScanOpaque) scan->opaque;

  bool    res;

  int64   ntids = 0;

  HashScanPosItem *currItem;

  res = _hash_first(scan, ForwardScanDirection);

  while (res)

  {

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

    /*

     * _hash_first and _hash_next handle eliminate dead index entries

     * whenever scan->ignore_killed_tuples is true.  Therefore, there's

     * nothing to do here except add the results to the TIDBitmap.

     */

    tbm_add_tuples(tbm, &(currItem->heapTid), 1, true);

    ntids++;

    res = _hash_next(scan, ForwardScanDirection);

  }

  return ntids;

}

/*

 *  hashbeginscan() -- start a scan on a hash index

 */

IndexScanDesc

hashbeginscan(Relation rel, int nkeys, int norderbys)

{

  IndexScanDesc scan;

  HashScanOpaque so;

  /* no order by operators allowed */

  Assert(norderbys == 0);

  scan = RelationGetIndexScan(rel, nkeys, norderbys);

  so = (HashScanOpaque) palloc(sizeof(HashScanOpaqueData));

  HashScanPosInvalidate(so->currPos);

  so->hashso_bucket_buf = InvalidBuffer;

  so->hashso_split_bucket_buf = InvalidBuffer;

  so->hashso_buc_populated = false;

  so->hashso_buc_split = false;

  so->killedItems = NULL;

  so->numKilled = 0;

  scan->opaque = so;

  return scan;

}

/*

 *  hashrescan() -- rescan an index relation

 */

void

hashrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,

       ScanKey orderbys, int norderbys)

{

  HashScanOpaque so = (HashScanOpaque) scan->opaque;

  Relation  rel = scan->indexRelation;

  if (HashScanPosIsValid(so->currPos))

  {

    /* Before leaving current page, deal with any killed items */

    if (so->numKilled > 0)

      _hash_kill_items(scan);

  }

  _hash_dropscanbuf(rel, so);

  /* set position invalid (this will cause _hash_first call) */

  HashScanPosInvalidate(so->currPos);

  /* Update scan key, if a new one is given */

  if (scankey && scan->numberOfKeys > 0)

    memcpy(scan->keyData, scankey, scan->numberOfKeys * sizeof(ScanKeyData));

  so->hashso_buc_populated = false;

  so->hashso_buc_split = false;

}

/*

 *  hashendscan() -- close down a scan

 */

void

hashendscan(IndexScanDesc scan)

{

  HashScanOpaque so = (HashScanOpaque) scan->opaque;

  Relation  rel = scan->indexRelation;

  if (HashScanPosIsValid(so->currPos))

  {

    /* Before leaving current page, deal with any killed items */

    if (so->numKilled > 0)

      _hash_kill_items(scan);

  }

  _hash_dropscanbuf(rel, so);

  if (so->killedItems != NULL)

    pfree(so->killedItems);

  pfree(so);

  scan->opaque = NULL;

}

/*

 * Bulk deletion of all index entries pointing to a set of heap tuples.

 * The set of target tuples is specified via a callback routine that tells

 * whether any given heap tuple (identified by ItemPointer) is being deleted.

 *

 * This function also deletes the tuples that are moved by split to other

 * bucket.

 *

 * Result: a palloc'd struct containing statistical info for VACUUM displays.

 */

IndexBulkDeleteResult *

hashbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,

         IndexBulkDeleteCallback callback, void *callback_state)

{

  Relation  rel = info->index;

  double    tuples_removed;

  double    num_index_tuples;

  double    orig_ntuples;

  Bucket    orig_maxbucket;

  Bucket    cur_maxbucket;

  Bucket    cur_bucket;

  Buffer    metabuf = InvalidBuffer;

  HashMetaPage metap;

  HashMetaPage cachedmetap;

  tuples_removed = 0;

  num_index_tuples = 0;

  /*

   * We need a copy of the metapage so that we can use its hashm_spares[]

   * values to compute bucket page addresses, but a cached copy should be

   * good enough.  (If not, we'll detect that further down and refresh the

   * cache as necessary.)

   */

  cachedmetap = _hash_getcachedmetap(rel, &metabuf, false);

  Assert(cachedmetap != NULL);

  orig_maxbucket = cachedmetap->hashm_maxbucket;

  orig_ntuples = cachedmetap->hashm_ntuples;

  /* Scan the buckets that we know exist */

  cur_bucket = 0;

  cur_maxbucket = orig_maxbucket;

loop_top:

  while (cur_bucket <= cur_maxbucket)

  {

    BlockNumber bucket_blkno;

    BlockNumber blkno;

    Buffer    bucket_buf;

    Buffer    buf;

    HashPageOpaque bucket_opaque;

    Page    page;

    bool    split_cleanup = false;

    /* Get address of bucket's start page */

    bucket_blkno = BUCKET_TO_BLKNO(cachedmetap, cur_bucket);

    blkno = bucket_blkno;

    /*

     * We need to acquire a cleanup lock on the primary bucket page to out

     * wait concurrent scans before deleting the dead tuples.

     */

    buf = ReadBufferExtended(rel, MAIN_FORKNUM, blkno, RBM_NORMAL, info->strategy);

    LockBufferForCleanup(buf);

    _hash_checkpage(rel, buf, LH_BUCKET_PAGE);

    page = BufferGetPage(buf);

    bucket_opaque = HashPageGetOpaque(page);

    /*

     * If the bucket contains tuples that are moved by split, then we need

     * to delete such tuples.  We can't delete such tuples if the split

     * operation on bucket is not finished as those are needed by scans.

     */

    if (!H_BUCKET_BEING_SPLIT(bucket_opaque) &&

      H_NEEDS_SPLIT_CLEANUP(bucket_opaque))

    {

      split_cleanup = true;

      /*

       * This bucket might have been split since we last held a lock on

       * the metapage.  If so, hashm_maxbucket, hashm_highmask and

       * hashm_lowmask might be old enough to cause us to fail to remove

       * tuples left behind by the most recent split.  To prevent that,

       * now that the primary page of the target bucket has been locked

       * (and thus can't be further split), check whether we need to

       * update our cached metapage data.

       */

      Assert(bucket_opaque->hasho_prevblkno != InvalidBlockNumber);

      if (bucket_opaque->hasho_prevblkno > cachedmetap->hashm_maxbucket)

      {

        cachedmetap = _hash_getcachedmetap(rel, &metabuf, true);

        Assert(cachedmetap != NULL);

      }

    }

    bucket_buf = buf;

    hashbucketcleanup(rel, cur_bucket, bucket_buf, blkno, info->strategy,

              cachedmetap->hashm_maxbucket,

              cachedmetap->hashm_highmask,

              cachedmetap->hashm_lowmask, &tuples_removed,

              &num_index_tuples, split_cleanup,

              callback, callback_state);

    _hash_dropbuf(rel, bucket_buf);

    /* Advance to next bucket */

    cur_bucket++;

  }

  if (BufferIsInvalid(metabuf))

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

  /* Write-lock metapage and check for split since we started */

  LockBuffer(metabuf, BUFFER_LOCK_EXCLUSIVE);

  metap = HashPageGetMeta(BufferGetPage(metabuf));

  if (cur_maxbucket != metap->hashm_maxbucket)

  {

    /* There's been a split, so process the additional bucket(s) */

    LockBuffer(metabuf, BUFFER_LOCK_UNLOCK);

    cachedmetap = _hash_getcachedmetap(rel, &metabuf, true);

    Assert(cachedmetap != NULL);

    cur_maxbucket = cachedmetap->hashm_maxbucket;

    goto loop_top;

  }

  /* Okay, we're really done.  Update tuple count in metapage. */

  START_CRIT_SECTION();

  if (orig_maxbucket == metap->hashm_maxbucket &&

    orig_ntuples == metap->hashm_ntuples)

  {

    /*

     * No one has split or inserted anything since start of scan, so

     * believe our count as gospel.

     */

    metap->hashm_ntuples = num_index_tuples;

  }

  else

  {

    /*

     * Otherwise, our count is untrustworthy since we may have

     * double-scanned tuples in split buckets.  Proceed by dead-reckoning.

     * (Note: we still return estimated_count = false, because using this

     * count is better than not updating reltuples at all.)

     */

    if (metap->hashm_ntuples > tuples_removed)

      metap->hashm_ntuples -= tuples_removed;

    else

      metap->hashm_ntuples = 0;

    num_index_tuples = metap->hashm_ntuples;

  }

  MarkBufferDirty(metabuf);

  /* XLOG stuff */

  if (RelationNeedsWAL(rel))

  {

    xl_hash_update_meta_page xlrec;

    XLogRecPtr  recptr;

    xlrec.ntuples = metap->hashm_ntuples;

    XLogBeginInsert();

    XLogRegisterData(&xlrec, SizeOfHashUpdateMetaPage);

    XLogRegisterBuffer(0, metabuf, REGBUF_STANDARD);

    recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_UPDATE_META_PAGE);

    PageSetLSN(BufferGetPage(metabuf), recptr);

  }

  END_CRIT_SECTION();

  _hash_relbuf(rel, metabuf);

  /* return statistics */

  if (stats == NULL)

    stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));

  stats->estimated_count = false;

  stats->num_index_tuples = num_index_tuples;

  stats->tuples_removed += tuples_removed;

  /* hashvacuumcleanup will fill in num_pages */

  return stats;

}

/*

 * Post-VACUUM cleanup.

 *

 * Result: a palloc'd struct containing statistical info for VACUUM displays.

 */

IndexBulkDeleteResult *

hashvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats)

{

  Relation  rel = info->index;

  BlockNumber num_pages;

  /* If hashbulkdelete wasn't called, return NULL signifying no change */

  /* Note: this covers the analyze_only case too */

  if (stats == NULL)

    return NULL;

  /* update statistics */

  num_pages = RelationGetNumberOfBlocks(rel);

  stats->num_pages = num_pages;

  return stats;

}

/*

 * Helper function to perform deletion of index entries from a bucket.

 *

 * This function expects that the caller has acquired a cleanup lock on the

 * primary bucket page, and will return with a write lock again held on the

 * primary bucket page.  The lock won't necessarily be held continuously,

 * though, because we'll release it when visiting overflow pages.

 *

 * There can't be any concurrent scans in progress when we first enter this

 * function because of the cleanup lock we hold on the primary bucket page,

 * but as soon as we release that lock, there might be.  If those scans got

 * ahead of our cleanup scan, they might see a tuple before we kill it and

 * wake up only after VACUUM has completed and the TID has been recycled for

 * an unrelated tuple.  To avoid that calamity, we prevent scans from passing

 * our cleanup scan by locking the next page in the bucket chain before

 * releasing the lock on the previous page.  (This type of lock chaining is not

 * ideal, so we might want to look for a better solution at some point.)

 *

 * We need to retain a pin on the primary bucket to ensure that no concurrent

 * split can start.

 */

void

hashbucketcleanup(Relation rel, Bucket cur_bucket, Buffer bucket_buf,

          BlockNumber bucket_blkno, BufferAccessStrategy bstrategy,

          uint32 maxbucket, uint32 highmask, uint32 lowmask,

          double *tuples_removed, double *num_index_tuples,

          bool split_cleanup,

          IndexBulkDeleteCallback callback, void *callback_state)

{

  BlockNumber blkno;

  Buffer    buf;

  Bucket    new_bucket PG_USED_FOR_ASSERTS_ONLY = InvalidBucket;

  bool    bucket_dirty = false;

  blkno = bucket_blkno;

  buf = bucket_buf;

  if (split_cleanup)

    new_bucket = _hash_get_newbucket_from_oldbucket(rel, cur_bucket,

                            lowmask, maxbucket);

  /* Scan each page in bucket */

  for (;;)

  {

    HashPageOpaque opaque;

    OffsetNumber offno;

    OffsetNumber maxoffno;

    Buffer    next_buf;

    Page    page;

    OffsetNumber deletable[MaxOffsetNumber];

    int     ndeletable = 0;

    bool    retain_pin = false;

    bool    clear_dead_marking = false;

    vacuum_delay_point(false);

    page = BufferGetPage(buf);

    opaque = HashPageGetOpaque(page);

    /* Scan each tuple in page */

    maxoffno = PageGetMaxOffsetNumber(page);

    for (offno = FirstOffsetNumber;

       offno <= maxoffno;

       offno = OffsetNumberNext(offno))

    {

      ItemPointer htup;

      IndexTuple  itup;

      Bucket    bucket;

      bool    kill_tuple = false;

      itup = (IndexTuple) PageGetItem(page,

                      PageGetItemId(page, offno));

      htup = &(itup->t_tid);

      /*

       * To remove the dead tuples, we strictly want to rely on results

       * of callback function.  refer btvacuumpage for detailed reason.

       */

      if (callback && callback(htup, callback_state))

      {

        kill_tuple = true;

        if (tuples_removed)

          *tuples_removed += 1;

      }

      else if (split_cleanup)

      {

        /* delete the tuples that are moved by split. */

        bucket = _hash_hashkey2bucket(_hash_get_indextuple_hashkey(itup),

                        maxbucket,

                        highmask,

                        lowmask);

        /* mark the item for deletion */

        if (bucket != cur_bucket)

        {

          /*

           * We expect tuples to either belong to current bucket or

           * new_bucket.  This is ensured because we don't allow

           * further splits from bucket that contains garbage. See

           * comments in _hash_expandtable.

           */

          Assert(bucket == new_bucket);

          kill_tuple = true;

        }

      }

      if (kill_tuple)

      {

        /* mark the item for deletion */

        deletable[ndeletable++] = offno;

      }

      else

      {

        /* we're keeping it, so count it */

        if (num_index_tuples)

          *num_index_tuples += 1;

      }

    }

    /* retain the pin on primary bucket page till end of bucket scan */

    if (blkno == bucket_blkno)

      retain_pin = true;

    else

      retain_pin = false;

    blkno = opaque->hasho_nextblkno;

    /*

     * Apply deletions, advance to next page and write page if needed.

     */

    if (ndeletable > 0)

    {

      /* No ereport(ERROR) until changes are logged */

      START_CRIT_SECTION();

      PageIndexMultiDelete(page, deletable, ndeletable);

      bucket_dirty = true;

      /*

       * Let us mark the page as clean if vacuum removes the DEAD tuples

       * from an index page. We do this by clearing

       * LH_PAGE_HAS_DEAD_TUPLES flag.

       */

      if (tuples_removed && *tuples_removed > 0 &&

        H_HAS_DEAD_TUPLES(opaque))

      {

        opaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;

        clear_dead_marking = true;

      }

      MarkBufferDirty(buf);

      /* XLOG stuff */

      if (RelationNeedsWAL(rel))

      {

        xl_hash_delete xlrec;

        XLogRecPtr  recptr;

        xlrec.clear_dead_marking = clear_dead_marking;

        xlrec.is_primary_bucket_page = (buf == bucket_buf);

        XLogBeginInsert();

        XLogRegisterData(&xlrec, SizeOfHashDelete);

        /*

         * bucket buffer was not changed, but still needs to be

         * registered to ensure that we can acquire a cleanup lock on

         * it during replay.

         */

        if (!xlrec.is_primary_bucket_page)

        {

          uint8   flags = REGBUF_STANDARD | REGBUF_NO_IMAGE | REGBUF_NO_CHANGE;

          XLogRegisterBuffer(0, bucket_buf, flags);

        }

        XLogRegisterBuffer(1, buf, REGBUF_STANDARD);

        XLogRegisterBufData(1, deletable,

                  ndeletable * sizeof(OffsetNumber));

        recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_DELETE);

        PageSetLSN(BufferGetPage(buf), recptr);

      }

      END_CRIT_SECTION();

    }

    /* bail out if there are no more pages to scan. */

    if (!BlockNumberIsValid(blkno))

      break;

    next_buf = _hash_getbuf_with_strategy(rel, blkno, HASH_WRITE,

                        LH_OVERFLOW_PAGE,

                        bstrategy);

    /*

     * release the lock on previous page after acquiring the lock on next

     * page

     */

    if (retain_pin)

      LockBuffer(buf, BUFFER_LOCK_UNLOCK);

    else

      _hash_relbuf(rel, buf);

    buf = next_buf;

  }

  /*

   * lock the bucket page to clear the garbage flag and squeeze the bucket.

   * if the current buffer is same as bucket buffer, then we already have

   * lock on bucket page.

   */

  if (buf != bucket_buf)

  {

    _hash_relbuf(rel, buf);

    LockBuffer(bucket_buf, BUFFER_LOCK_EXCLUSIVE);

  }

  /*

   * Clear the garbage flag from bucket after deleting the tuples that are

   * moved by split.  We purposefully clear the flag before squeeze bucket,

   * so that after restart, vacuum shouldn't again try to delete the moved

   * by split tuples.

   */

  if (split_cleanup)

  {

    HashPageOpaque bucket_opaque;

    Page    page;

    page = BufferGetPage(bucket_buf);

    bucket_opaque = HashPageGetOpaque(page);

    /* No ereport(ERROR) until changes are logged */

    START_CRIT_SECTION();

    bucket_opaque->hasho_flag &= ~LH_BUCKET_NEEDS_SPLIT_CLEANUP;

    MarkBufferDirty(bucket_buf);

    /* XLOG stuff */

    if (RelationNeedsWAL(rel))

    {

      XLogRecPtr  recptr;

      XLogBeginInsert();

      XLogRegisterBuffer(0, bucket_buf, REGBUF_STANDARD);

      recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_SPLIT_CLEANUP);

      PageSetLSN(page, recptr);

    }

    END_CRIT_SECTION();

  }

  /*

   * If we have deleted anything, try to compact free space.  For squeezing

   * the bucket, we must have a cleanup lock, else it can impact the

   * ordering of tuples for a scan that has started before it.

   */

  if (bucket_dirty && IsBufferCleanupOK(bucket_buf))

    _hash_squeezebucket(rel, cur_bucket, bucket_blkno, bucket_buf,

              bstrategy);

  else

    LockBuffer(bucket_buf, BUFFER_LOCK_UNLOCK);

}

CompareType

hashtranslatestrategy(StrategyNumber strategy, Oid opfamily)

{

  if (strategy == HTEqualStrategyNumber)

    return COMPARE_EQ;

  return COMPARE_INVALID;

}

StrategyNumber