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

 *

 * hashinsert.c

 *    Item insertion in hash tables 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/hashinsert.c

 *

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

 */

#include "postgres.h"

#include "access/hash.h"

#include "access/hash_xlog.h"

#include "access/xloginsert.h"

#include "miscadmin.h"

#include "storage/predicate.h"

#include "utils/rel.h"

static void _hash_vacuum_one_page(Relation rel, Relation hrel,

                  Buffer metabuf, Buffer buf);

/*

 *  _hash_doinsert() -- Handle insertion of a single index tuple.

 *

 *    This routine is called by the public interface routines, hashbuild

 *    and hashinsert.  By here, itup is completely filled in.

 *

 * 'sorted' must only be passed as 'true' when inserts are done in hashkey

 * order.

 */

void

_hash_doinsert(Relation rel, IndexTuple itup, Relation heapRel, bool sorted)

{

  Buffer    buf = InvalidBuffer;

  Buffer    bucket_buf;

  Buffer    metabuf;

  HashMetaPage metap;

  HashMetaPage usedmetap = NULL;

  Page    metapage;

  Page    page;

  HashPageOpaque pageopaque;

  Size    itemsz;

  bool    do_expand;

  uint32    hashkey;

  Bucket    bucket;

  OffsetNumber itup_off;

  /*

   * Get the hash key for the item (it's stored in the index tuple itself).

   */

  hashkey = _hash_get_indextuple_hashkey(itup);

  /* compute item size too */

  itemsz = IndexTupleSize(itup);

  itemsz = MAXALIGN(itemsz); /* be safe, PageAddItem will do this but we

                 * need to be consistent */

restart_insert:

  /*

   * Read the metapage.  We don't lock it yet; HashMaxItemSize() will

   * examine pd_pagesize_version, but that can't change so we can examine it

   * without a lock.

   */

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

  metapage = BufferGetPage(metabuf);

  /*

   * Check whether the item can fit on a hash page at all. (Eventually, we

   * ought to try to apply TOAST methods if not.)  Note that at this point,

   * itemsz doesn't include the ItemId.

   *

   * XXX this is useless code if we are only storing hash keys.

   */

  if (itemsz > HashMaxItemSize(metapage))

    ereport(ERROR,

        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),

         errmsg("index row size %zu exceeds hash maximum %zu",

            itemsz, HashMaxItemSize(metapage)),

         errhint("Values larger than a buffer page cannot be indexed.")));

  /* Lock the primary bucket page for the target bucket. */

  buf = _hash_getbucketbuf_from_hashkey(rel, hashkey, HASH_WRITE,

                      &usedmetap);

  Assert(usedmetap != NULL);

  CheckForSerializableConflictIn(rel, NULL, BufferGetBlockNumber(buf));

  /* remember the primary bucket buffer to release the pin on it at end. */

  bucket_buf = buf;

  page = BufferGetPage(buf);

  pageopaque = HashPageGetOpaque(page);

  bucket = pageopaque->hasho_bucket;

  /*

   * If this bucket is in the process of being split, try to finish the

   * split before inserting, because that might create room for the

   * insertion to proceed without allocating an additional overflow page.

   * It's only interesting to finish the split if we're trying to insert

   * into the bucket from which we're removing tuples (the "old" bucket),

   * not if we're trying to insert into the bucket into which tuples are

   * being moved (the "new" bucket).

   */

  if (H_BUCKET_BEING_SPLIT(pageopaque) && IsBufferCleanupOK(buf))

  {

    /* release the lock on bucket buffer, before completing the split. */

    LockBuffer(buf, BUFFER_LOCK_UNLOCK);

    _hash_finish_split(rel, metabuf, buf, bucket,

               usedmetap->hashm_maxbucket,

               usedmetap->hashm_highmask,

               usedmetap->hashm_lowmask);

    /* release the pin on old and meta buffer.  retry for insert. */

    _hash_dropbuf(rel, buf);

    _hash_dropbuf(rel, metabuf);

    goto restart_insert;

  }

  /* Do the insertion */

  while (PageGetFreeSpace(page) < itemsz)

  {

    BlockNumber nextblkno;

    /*

     * Check if current page has any DEAD tuples. If yes, delete these

     * tuples and see if we can get a space for the new item to be

     * inserted before moving to the next page in the bucket chain.

     */

    if (H_HAS_DEAD_TUPLES(pageopaque))

    {

      if (IsBufferCleanupOK(buf))

      {

        _hash_vacuum_one_page(rel, heapRel, metabuf, buf);

        if (PageGetFreeSpace(page) >= itemsz)

          break;   /* OK, now we have enough space */

      }

    }

    /*

     * no space on this page; check for an overflow page

     */

    nextblkno = pageopaque->hasho_nextblkno;

    if (BlockNumberIsValid(nextblkno))

    {

      /*

       * ovfl page exists; go get it.  if it doesn't have room, we'll

       * find out next pass through the loop test above.  we always

       * release both the lock and pin if this is an overflow page, but

       * only the lock if this is the primary bucket page, since the pin

       * on the primary bucket must be retained throughout the scan.

       */

      if (buf != bucket_buf)

        _hash_relbuf(rel, buf);

      else

        LockBuffer(buf, BUFFER_LOCK_UNLOCK);

      buf = _hash_getbuf(rel, nextblkno, HASH_WRITE, LH_OVERFLOW_PAGE);

      page = BufferGetPage(buf);

    }

    else

    {

      /*

       * we're at the end of the bucket chain and we haven't found a

       * page with enough room.  allocate a new overflow page.

       */

      /* release our write lock without modifying buffer */

      LockBuffer(buf, BUFFER_LOCK_UNLOCK);

      /* chain to a new overflow page */

      buf = _hash_addovflpage(rel, metabuf, buf, (buf == bucket_buf));

      page = BufferGetPage(buf);

      /* should fit now, given test above */

      Assert(PageGetFreeSpace(page) >= itemsz);

    }

    pageopaque = HashPageGetOpaque(page);

    Assert((pageopaque->hasho_flag & LH_PAGE_TYPE) == LH_OVERFLOW_PAGE);

    Assert(pageopaque->hasho_bucket == bucket);

  }

  /*

   * Write-lock the metapage so we can increment the tuple count. After

   * incrementing it, check to see if it's time for a split.

   */

  LockBuffer(metabuf, BUFFER_LOCK_EXCLUSIVE);

  /* Do the update.  No ereport(ERROR) until changes are logged */

  START_CRIT_SECTION();

  /* found page with enough space, so add the item here */

  itup_off = _hash_pgaddtup(rel, buf, itemsz, itup, sorted);

  MarkBufferDirty(buf);

  /* metapage operations */

  metap = HashPageGetMeta(metapage);

  metap->hashm_ntuples += 1;

  /* Make sure this stays in sync with _hash_expandtable() */

  do_expand = metap->hashm_ntuples >

    (double) metap->hashm_ffactor * (metap->hashm_maxbucket + 1);

  MarkBufferDirty(metabuf);

  /* XLOG stuff */

  if (RelationNeedsWAL(rel))

  {

    xl_hash_insert xlrec;

    XLogRecPtr  recptr;

    xlrec.offnum = itup_off;

    XLogBeginInsert();

    XLogRegisterData(&xlrec, SizeOfHashInsert);

    XLogRegisterBuffer(1, metabuf, REGBUF_STANDARD);

    XLogRegisterBuffer(0, buf, REGBUF_STANDARD);

    XLogRegisterBufData(0, itup, IndexTupleSize(itup));

    recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_INSERT);

    PageSetLSN(BufferGetPage(buf), recptr);

    PageSetLSN(BufferGetPage(metabuf), recptr);

  }

  END_CRIT_SECTION();

  /* drop lock on metapage, but keep pin */

  LockBuffer(metabuf, BUFFER_LOCK_UNLOCK);

  /*

   * Release the modified page and ensure to release the pin on primary

   * page.

   */

  _hash_relbuf(rel, buf);

  if (buf != bucket_buf)

    _hash_dropbuf(rel, bucket_buf);

  /* Attempt to split if a split is needed */

  if (do_expand)

    _hash_expandtable(rel, metabuf);

  /* Finally drop our pin on the metapage */

  _hash_dropbuf(rel, metabuf);

}

/*

 *  _hash_pgaddtup() -- add a tuple to a particular page in the index.

 *

 * This routine adds the tuple to the page as requested; it does not write out

 * the page.  It is an error to call this function without pin and write lock

 * on the target buffer.

 *

 * Returns the offset number at which the tuple was inserted.  This function

 * is responsible for preserving the condition that tuples in a hash index

 * page are sorted by hashkey value, however, if the caller is certain that

 * the hashkey for the tuple being added is >= the hashkeys of all existing

 * tuples on the page, then the 'appendtup' flag may be passed as true.  This

 * saves from having to binary search for the correct location to insert the

 * tuple.

 */

OffsetNumber

_hash_pgaddtup(Relation rel, Buffer buf, Size itemsize, IndexTuple itup,

         bool appendtup)

{

  OffsetNumber itup_off;

  Page    page;

  _hash_checkpage(rel, buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);

  page = BufferGetPage(buf);

  /*

   * Find where to insert the tuple (preserving page's hashkey ordering). If

   * 'appendtup' is true then we just insert it at the end.

   */

  if (appendtup)

  {

    itup_off = PageGetMaxOffsetNumber(page) + 1;

#ifdef USE_ASSERT_CHECKING

    /* ensure this tuple's hashkey is >= the final existing tuple */

    if (PageGetMaxOffsetNumber(page) > 0)

    {

      IndexTuple  lasttup;

      ItemId    itemid;

      itemid = PageGetItemId(page, PageGetMaxOffsetNumber(page));

      lasttup = (IndexTuple) PageGetItem(page, itemid);

      Assert(_hash_get_indextuple_hashkey(lasttup) <=

           _hash_get_indextuple_hashkey(itup));

    }

#endif

  }

  else

  {

    uint32    hashkey = _hash_get_indextuple_hashkey(itup);

    itup_off = _hash_binsearch(page, hashkey);

  }

  if (PageAddItem(page, (Item) itup, itemsize, itup_off, false, false)

    == InvalidOffsetNumber)

    elog(ERROR, "failed to add index item to \"%s\"",

       RelationGetRelationName(rel));

  return itup_off;

}

/*

 *  _hash_pgaddmultitup() -- add a tuple vector to a particular page in the

 *               index.

 *

 * This routine has same requirements for locking and tuple ordering as

 * _hash_pgaddtup().

 *

 * Returns the offset number array at which the tuples were inserted.

 */

void

_hash_pgaddmultitup(Relation rel, Buffer buf, IndexTuple *itups,

          OffsetNumber *itup_offsets, uint16 nitups)

{

  OffsetNumber itup_off;

  Page    page;

  uint32    hashkey;

  int     i;

  _hash_checkpage(rel, buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);

  page = BufferGetPage(buf);

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

  {

    Size    itemsize;

    itemsize = IndexTupleSize(itups[i]);

    itemsize = MAXALIGN(itemsize);

    /* Find where to insert the tuple (preserving page's hashkey ordering) */

    hashkey = _hash_get_indextuple_hashkey(itups[i]);

    itup_off = _hash_binsearch(page, hashkey);

    itup_offsets[i] = itup_off;

    if (PageAddItem(page, (Item) itups[i], itemsize, itup_off, false, false)

      == InvalidOffsetNumber)

      elog(ERROR, "failed to add index item to \"%s\"",

         RelationGetRelationName(rel));

  }

}

/*

 * _hash_vacuum_one_page - vacuum just one index page.

 *

 * Try to remove LP_DEAD items from the given page. We must acquire cleanup

 * lock on the page being modified before calling this function.

 */

static void

_hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)

{

  OffsetNumber deletable[MaxOffsetNumber];

  int     ndeletable = 0;

  OffsetNumber offnum,

        maxoff;

  Page    page = BufferGetPage(buf);

  HashPageOpaque pageopaque;

  HashMetaPage metap;

  /* Scan each tuple in page to see if it is marked as LP_DEAD */

  maxoff = PageGetMaxOffsetNumber(page);

  for (offnum = FirstOffsetNumber;

     offnum <= maxoff;

     offnum = OffsetNumberNext(offnum))

  {

    ItemId    itemId = PageGetItemId(page, offnum);

    if (ItemIdIsDead(itemId))

      deletable[ndeletable++] = offnum;

  }

  if (ndeletable > 0)

  {

    TransactionId snapshotConflictHorizon;

    snapshotConflictHorizon =

      index_compute_xid_horizon_for_tuples(rel, hrel, buf,

                         deletable, ndeletable);

    /*

     * Write-lock the meta page so that we can decrement tuple count.

     */

    LockBuffer(metabuf, BUFFER_LOCK_EXCLUSIVE);

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

    START_CRIT_SECTION();

    PageIndexMultiDelete(page, deletable, ndeletable);

    /*

     * Mark the page as not containing any LP_DEAD items. This is not

     * certainly true (there might be some that have recently been marked,

     * but weren't included in our target-item list), but it will almost

     * always be true and it doesn't seem worth an additional page scan to

     * check it. Remember that LH_PAGE_HAS_DEAD_TUPLES is only a hint

     * anyway.

     */

    pageopaque = HashPageGetOpaque(page);

    pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;

    metap = HashPageGetMeta(BufferGetPage(metabuf));

    metap->hashm_ntuples -= ndeletable;

    MarkBufferDirty(buf);

    MarkBufferDirty(metabuf);

    /* XLOG stuff */

    if (RelationNeedsWAL(rel))

    {

      xl_hash_vacuum_one_page xlrec;

      XLogRecPtr  recptr;

      xlrec.isCatalogRel = RelationIsAccessibleInLogicalDecoding(hrel);

      xlrec.snapshotConflictHorizon = snapshotConflictHorizon;

      xlrec.ntuples = ndeletable;

      XLogBeginInsert();

      XLogRegisterBuffer(0, buf, REGBUF_STANDARD);

      XLogRegisterData(&xlrec, SizeOfHashVacuumOnePage);

      /*

       * We need the target-offsets array whether or not we store the

       * whole buffer, to allow us to find the snapshotConflictHorizon

       * on a standby server.

       */

      XLogRegisterData(deletable,

               ndeletable * sizeof(OffsetNumber));

      XLogRegisterBuffer(1, metabuf, REGBUF_STANDARD);

      recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_VACUUM_ONE_PAGE);

      PageSetLSN(BufferGetPage(buf), recptr);

      PageSetLSN(BufferGetPage(metabuf), recptr);

    }

    END_CRIT_SECTION();

    /*

     * Releasing write lock on meta page as we have updated the tuple

     * count.

     */

    LockBuffer(metabuf, BUFFER_LOCK_UNLOCK);

  }

}