/src/postgres/src/backend/access/gin/ginfast.c

Source
/*-------------------------------------------------------------------------
 *
 * ginfast.c
 *    Fast insert routines for the Postgres inverted index access method.
 *    Pending entries are stored in linear list of pages.  Later on
 *    (typically during VACUUM), ginInsertCleanup() will be invoked to
 *    transfer pending entries into the regular index structure.  This
 *    wins because bulk insertion is much more efficient than retail.
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *      src/backend/access/gin/ginfast.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/gin_private.h"
#include "access/ginxlog.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/pg_am.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "port/pg_bitutils.h"
#include "postmaster/autovacuum.h"
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "storage/predicate.h"
#include "utils/acl.h"
#include "utils/fmgrprotos.h"
#include "utils/memutils.h"
#include "utils/rel.h"

/* GUC parameter */
int     gin_pending_list_limit = 0;

#define GIN_PAGE_FREESIZE \
  ( (Size) BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(GinPageOpaqueData)) )

typedef struct KeyArray
{
  Datum    *keys;     /* expansible array */
  GinNullCategory *categories;  /* another expansible array */
  int32   nvalues;    /* current number of valid entries */
  int32   maxvalues;    /* allocated size of arrays */
} KeyArray;


/*
 * Build a pending-list page from the given array of tuples, and write it out.
 *
 * Returns amount of free space left on the page.
 */
static int32
writeListPage(Relation index, Buffer buffer,
        const IndexTuple *tuples, int32 ntuples, BlockNumber rightlink)
{
  Page    page = BufferGetPage(buffer);
  int32   i,
        freesize,
        size = 0;
  OffsetNumber l,
        off;
  PGAlignedBlock workspace;
  char     *ptr;

  START_CRIT_SECTION();

  GinInitBuffer(buffer, GIN_LIST);

  off = FirstOffsetNumber;
  ptr = workspace.data;

  for (i = 0; i < ntuples; i++)
  {
    int     this_size = IndexTupleSize(tuples[i]);

    memcpy(ptr, tuples[i], this_size);
    ptr += this_size;
    size += this_size;

    l = PageAddItem(page, tuples[i], this_size, off, false, false);

    if (l == InvalidOffsetNumber)
      elog(ERROR, "failed to add item to index page in \"%s\"",
         RelationGetRelationName(index));

    off++;
  }

  Assert(size <= BLCKSZ);   /* else we overran workspace */

  GinPageGetOpaque(page)->rightlink = rightlink;

  /*
   * tail page may contain only whole row(s) or final part of row placed on
   * previous pages (a "row" here meaning all the index tuples generated for
   * one heap tuple)
   */
  if (rightlink == InvalidBlockNumber)
  {
    GinPageSetFullRow(page);
    GinPageGetOpaque(page)->maxoff = 1;
  }
  else
  {
    GinPageGetOpaque(page)->maxoff = 0;
  }

  MarkBufferDirty(buffer);

  if (RelationNeedsWAL(index))
  {
    ginxlogInsertListPage data;
    XLogRecPtr  recptr;

    data.rightlink = rightlink;
    data.ntuples = ntuples;

    XLogBeginInsert();
    XLogRegisterData(&data, sizeof(ginxlogInsertListPage));

    XLogRegisterBuffer(0, buffer, REGBUF_WILL_INIT);
    XLogRegisterBufData(0, workspace.data, size);

    recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_INSERT_LISTPAGE);
    PageSetLSN(page, recptr);
  }

  /* get free space before releasing buffer */
  freesize = PageGetExactFreeSpace(page);

  END_CRIT_SECTION();

  UnlockReleaseBuffer(buffer);

  return freesize;
}

static void
makeSublist(Relation index, IndexTuple *tuples, int32 ntuples,
      GinMetaPageData *res)
{
  Buffer    curBuffer = InvalidBuffer;
  Buffer    prevBuffer = InvalidBuffer;
  int     i,
        size = 0,
        tupsize;
  int     startTuple = 0;

  Assert(ntuples > 0);

  /*
   * Split tuples into pages
   */
  for (i = 0; i < ntuples; i++)
  {
    if (curBuffer == InvalidBuffer)
    {
      curBuffer = GinNewBuffer(index);

      if (prevBuffer != InvalidBuffer)
      {
        res->nPendingPages++;
        writeListPage(index, prevBuffer,
                tuples + startTuple,
                i - startTuple,
                BufferGetBlockNumber(curBuffer));
      }
      else
      {
        res->head = BufferGetBlockNumber(curBuffer);
      }

      prevBuffer = curBuffer;
      startTuple = i;
      size = 0;
    }

    tupsize = MAXALIGN(IndexTupleSize(tuples[i])) + sizeof(ItemIdData);

    if (size + tupsize > GinListPageSize)
    {
      /* won't fit, force a new page and reprocess */
      i--;
      curBuffer = InvalidBuffer;
    }
    else
    {
      size += tupsize;
    }
  }

  /*
   * Write last page
   */
  res->tail = BufferGetBlockNumber(curBuffer);
  res->tailFreeSize = writeListPage(index, curBuffer,
                    tuples + startTuple,
                    ntuples - startTuple,
                    InvalidBlockNumber);
  res->nPendingPages++;
  /* that was only one heap tuple */
  res->nPendingHeapTuples = 1;
}

/*
 * Write the index tuples contained in *collector into the index's
 * pending list.
 *
 * Function guarantees that all these tuples will be inserted consecutively,
 * preserving order
 */
void
ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector)
{
  Relation  index = ginstate->index;
  Buffer    metabuffer;
  Page    metapage;
  GinMetaPageData *metadata = NULL;
  Buffer    buffer = InvalidBuffer;
  Page    page = NULL;
  ginxlogUpdateMeta data;
  bool    separateList = false;
  bool    needCleanup = false;
  int     cleanupSize;
  bool    needWal;

  if (collector->ntuples == 0)
    return;

  needWal = RelationNeedsWAL(index);

  data.locator = index->rd_locator;
  data.ntuples = 0;
  data.newRightlink = data.prevTail = InvalidBlockNumber;

  metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
  metapage = BufferGetPage(metabuffer);

  /*
   * An insertion to the pending list could logically belong anywhere in the
   * tree, so it conflicts with all serializable scans.  All scans acquire a
   * predicate lock on the metabuffer to represent that.  Therefore we'll
   * check for conflicts in, but not until we have the page locked and are
   * ready to modify the page.
   */

  if (collector->sumsize + collector->ntuples * sizeof(ItemIdData) > GinListPageSize)
  {
    /*
     * Total size is greater than one page => make sublist
     */
    separateList = true;
  }
  else
  {
    LockBuffer(metabuffer, GIN_EXCLUSIVE);
    metadata = GinPageGetMeta(metapage);

    if (metadata->head == InvalidBlockNumber ||
      collector->sumsize + collector->ntuples * sizeof(ItemIdData) > metadata->tailFreeSize)
    {
      /*
       * Pending list is empty or total size is greater than freespace
       * on tail page => make sublist
       *
       * We unlock metabuffer to keep high concurrency
       */
      separateList = true;
      LockBuffer(metabuffer, GIN_UNLOCK);
    }
  }

  if (separateList)
  {
    /*
     * We should make sublist separately and append it to the tail
     */
    GinMetaPageData sublist;

    memset(&sublist, 0, sizeof(GinMetaPageData));
    makeSublist(index, collector->tuples, collector->ntuples, &sublist);

    /*
     * metapage was unlocked, see above
     */
    LockBuffer(metabuffer, GIN_EXCLUSIVE);
    metadata = GinPageGetMeta(metapage);

    CheckForSerializableConflictIn(index, NULL, GIN_METAPAGE_BLKNO);

    if (metadata->head == InvalidBlockNumber)
    {
      /*
       * Main list is empty, so just insert sublist as main list
       */
      START_CRIT_SECTION();

      metadata->head = sublist.head;
      metadata->tail = sublist.tail;
      metadata->tailFreeSize = sublist.tailFreeSize;

      metadata->nPendingPages = sublist.nPendingPages;
      metadata->nPendingHeapTuples = sublist.nPendingHeapTuples;

      if (needWal)
        XLogBeginInsert();
    }
    else
    {
      /*
       * Merge lists
       */
      data.prevTail = metadata->tail;
      data.newRightlink = sublist.head;

      buffer = ReadBuffer(index, metadata->tail);
      LockBuffer(buffer, GIN_EXCLUSIVE);
      page = BufferGetPage(buffer);

      Assert(GinPageGetOpaque(page)->rightlink == InvalidBlockNumber);

      START_CRIT_SECTION();

      GinPageGetOpaque(page)->rightlink = sublist.head;

      MarkBufferDirty(buffer);

      metadata->tail = sublist.tail;
      metadata->tailFreeSize = sublist.tailFreeSize;

      metadata->nPendingPages += sublist.nPendingPages;
      metadata->nPendingHeapTuples += sublist.nPendingHeapTuples;

      if (needWal)
      {
        XLogBeginInsert();
        XLogRegisterBuffer(1, buffer, REGBUF_STANDARD);
      }
    }
  }
  else
  {
    /*
     * Insert into tail page.  Metapage is already locked
     */
    OffsetNumber l,
          off;
    int     i,
          tupsize;
    char     *ptr;
    char     *collectordata;

    CheckForSerializableConflictIn(index, NULL, GIN_METAPAGE_BLKNO);

    buffer = ReadBuffer(index, metadata->tail);
    LockBuffer(buffer, GIN_EXCLUSIVE);
    page = BufferGetPage(buffer);

    off = (PageIsEmpty(page)) ? FirstOffsetNumber :
      OffsetNumberNext(PageGetMaxOffsetNumber(page));

    collectordata = ptr = (char *) palloc(collector->sumsize);

    data.ntuples = collector->ntuples;

    START_CRIT_SECTION();

    if (needWal)
      XLogBeginInsert();

    /*
     * Increase counter of heap tuples
     */
    Assert(GinPageGetOpaque(page)->maxoff <= metadata->nPendingHeapTuples);
    GinPageGetOpaque(page)->maxoff++;
    metadata->nPendingHeapTuples++;

    for (i = 0; i < collector->ntuples; i++)
    {
      tupsize = IndexTupleSize(collector->tuples[i]);
      l = PageAddItem(page, collector->tuples[i], tupsize, off, false, false);

      if (l == InvalidOffsetNumber)
        elog(ERROR, "failed to add item to index page in \"%s\"",
           RelationGetRelationName(index));

      memcpy(ptr, collector->tuples[i], tupsize);
      ptr += tupsize;

      off++;
    }

    Assert((ptr - collectordata) <= collector->sumsize);

    MarkBufferDirty(buffer);

    if (needWal)
    {
      XLogRegisterBuffer(1, buffer, REGBUF_STANDARD);
      XLogRegisterBufData(1, collectordata, collector->sumsize);
    }

    metadata->tailFreeSize = PageGetExactFreeSpace(page);
  }

  /*
   * Set pd_lower just past the end of the metadata.  This is essential,
   * because without doing so, metadata will be lost if xlog.c compresses
   * the page.  (We must do this here because pre-v11 versions of PG did not
   * set the metapage's pd_lower correctly, so a pg_upgraded index might
   * contain the wrong value.)
   */
  ((PageHeader) metapage)->pd_lower =
    ((char *) metadata + sizeof(GinMetaPageData)) - (char *) metapage;

  /*
   * Write metabuffer, make xlog entry
   */
  MarkBufferDirty(metabuffer);

  if (needWal)
  {
    XLogRecPtr  recptr;

    memcpy(&data.metadata, metadata, sizeof(GinMetaPageData));

    XLogRegisterBuffer(0, metabuffer, REGBUF_WILL_INIT | REGBUF_STANDARD);
    XLogRegisterData(&data, sizeof(ginxlogUpdateMeta));

    recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_UPDATE_META_PAGE);
    PageSetLSN(metapage, recptr);

    if (buffer != InvalidBuffer)
    {
      PageSetLSN(page, recptr);
    }
  }

  if (buffer != InvalidBuffer)
    UnlockReleaseBuffer(buffer);

  /*
   * Force pending list cleanup when it becomes too long. And,
   * ginInsertCleanup could take significant amount of time, so we prefer to
   * call it when it can do all the work in a single collection cycle. In
   * non-vacuum mode, it shouldn't require maintenance_work_mem, so fire it
   * while pending list is still small enough to fit into
   * gin_pending_list_limit.
   *
   * ginInsertCleanup() should not be called inside our CRIT_SECTION.
   */
  cleanupSize = GinGetPendingListCleanupSize(index);
  if (metadata->nPendingPages * GIN_PAGE_FREESIZE > cleanupSize * (Size) 1024)
    needCleanup = true;

  END_CRIT_SECTION();

  UnlockReleaseBuffer(metabuffer);

  /*
   * Since it could contend with concurrent cleanup process we cleanup
   * pending list not forcibly.
   */
  if (needCleanup)
    ginInsertCleanup(ginstate, false, true, false, NULL);
}

/*
 * Create temporary index tuples for a single indexable item (one index column
 * for the heap tuple specified by ht_ctid), and append them to the array
 * in *collector.  They will subsequently be written out using
 * ginHeapTupleFastInsert.  Note that to guarantee consistent state, all
 * temp tuples for a given heap tuple must be written in one call to
 * ginHeapTupleFastInsert.
 */
void
ginHeapTupleFastCollect(GinState *ginstate,
            GinTupleCollector *collector,
            OffsetNumber attnum, Datum value, bool isNull,
            ItemPointer ht_ctid)
{
  Datum    *entries;
  GinNullCategory *categories;
  int32   i,
        nentries;

  /*
   * Extract the key values that need to be inserted in the index
   */
  entries = ginExtractEntries(ginstate, attnum, value, isNull,
                &nentries, &categories);

  /*
   * Protect against integer overflow in allocation calculations
   */
  if (nentries < 0 ||
    collector->ntuples + nentries > MaxAllocSize / sizeof(IndexTuple))
    elog(ERROR, "too many entries for GIN index");

  /*
   * Allocate/reallocate memory for storing collected tuples
   */
  if (collector->tuples == NULL)
  {
    /*
     * Determine the number of elements to allocate in the tuples array
     * initially.  Make it a power of 2 to avoid wasting memory when
     * resizing (since palloc likes powers of 2).
     */
    collector->lentuples = pg_nextpower2_32(Max(16, nentries));
    collector->tuples = palloc_array(IndexTuple, collector->lentuples);
  }
  else if (collector->lentuples < collector->ntuples + nentries)
  {
    /*
     * Advance lentuples to the next suitable power of 2.  This won't
     * overflow, though we could get to a value that exceeds
     * MaxAllocSize/sizeof(IndexTuple), causing an error in repalloc.
     */
    collector->lentuples = pg_nextpower2_32(collector->ntuples + nentries);
    collector->tuples = repalloc_array(collector->tuples,
                       IndexTuple, collector->lentuples);
  }

  /*
   * Build an index tuple for each key value, and add to array.  In pending
   * tuples we just stick the heap TID into t_tid.
   */
  for (i = 0; i < nentries; i++)
  {
    IndexTuple  itup;

    itup = GinFormTuple(ginstate, attnum, entries[i], categories[i],
              NULL, 0, 0, true);
    itup->t_tid = *ht_ctid;
    collector->tuples[collector->ntuples++] = itup;
    collector->sumsize += IndexTupleSize(itup);
  }
}

/*
 * Deletes pending list pages up to (not including) newHead page.
 * If newHead == InvalidBlockNumber then function drops the whole list.
 *
 * metapage is pinned and exclusive-locked throughout this function.
 */
static void
shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
      bool fill_fsm, IndexBulkDeleteResult *stats)
{
  Page    metapage;
  GinMetaPageData *metadata;
  BlockNumber blknoToDelete;

  metapage = BufferGetPage(metabuffer);
  metadata = GinPageGetMeta(metapage);
  blknoToDelete = metadata->head;

  do
  {
    Page    page;
    int     i;
    int64   nDeletedHeapTuples = 0;
    ginxlogDeleteListPages data;
    Buffer    buffers[GIN_NDELETE_AT_ONCE];
    BlockNumber freespace[GIN_NDELETE_AT_ONCE];

    data.ndeleted = 0;
    while (data.ndeleted < GIN_NDELETE_AT_ONCE && blknoToDelete != newHead)
    {
      freespace[data.ndeleted] = blknoToDelete;
      buffers[data.ndeleted] = ReadBuffer(index, blknoToDelete);
      LockBuffer(buffers[data.ndeleted], GIN_EXCLUSIVE);
      page = BufferGetPage(buffers[data.ndeleted]);

      data.ndeleted++;

      Assert(!GinPageIsDeleted(page));

      nDeletedHeapTuples += GinPageGetOpaque(page)->maxoff;
      blknoToDelete = GinPageGetOpaque(page)->rightlink;
    }

    if (stats)
      stats->pages_deleted += data.ndeleted;

    /*
     * This operation touches an unusually large number of pages, so
     * prepare the XLogInsert machinery for that before entering the
     * critical section.
     */
    if (RelationNeedsWAL(index))
      XLogEnsureRecordSpace(data.ndeleted, 0);

    START_CRIT_SECTION();

    metadata->head = blknoToDelete;

    Assert(metadata->nPendingPages >= data.ndeleted);
    metadata->nPendingPages -= data.ndeleted;
    Assert(metadata->nPendingHeapTuples >= nDeletedHeapTuples);
    metadata->nPendingHeapTuples -= nDeletedHeapTuples;

    if (blknoToDelete == InvalidBlockNumber)
    {
      metadata->tail = InvalidBlockNumber;
      metadata->tailFreeSize = 0;
      metadata->nPendingPages = 0;
      metadata->nPendingHeapTuples = 0;
    }

    /*
     * Set pd_lower just past the end of the metadata.  This is essential,
     * because without doing so, metadata will be lost if xlog.c
     * compresses the page.  (We must do this here because pre-v11
     * versions of PG did not set the metapage's pd_lower correctly, so a
     * pg_upgraded index might contain the wrong value.)
     */
    ((PageHeader) metapage)->pd_lower =
      ((char *) metadata + sizeof(GinMetaPageData)) - (char *) metapage;

    MarkBufferDirty(metabuffer);

    for (i = 0; i < data.ndeleted; i++)
    {
      page = BufferGetPage(buffers[i]);
      GinPageGetOpaque(page)->flags = GIN_DELETED;
      MarkBufferDirty(buffers[i]);
    }

    if (RelationNeedsWAL(index))
    {
      XLogRecPtr  recptr;

      XLogBeginInsert();
      XLogRegisterBuffer(0, metabuffer,
                 REGBUF_WILL_INIT | REGBUF_STANDARD);
      for (i = 0; i < data.ndeleted; i++)
        XLogRegisterBuffer(i + 1, buffers[i], REGBUF_WILL_INIT);

      memcpy(&data.metadata, metadata, sizeof(GinMetaPageData));

      XLogRegisterData(&data,
               sizeof(ginxlogDeleteListPages));

      recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_DELETE_LISTPAGE);
      PageSetLSN(metapage, recptr);

      for (i = 0; i < data.ndeleted; i++)
      {
        page = BufferGetPage(buffers[i]);
        PageSetLSN(page, recptr);
      }
    }

    END_CRIT_SECTION();

    for (i = 0; i < data.ndeleted; i++)
      UnlockReleaseBuffer(buffers[i]);

    for (i = 0; fill_fsm && i < data.ndeleted; i++)
      RecordFreeIndexPage(index, freespace[i]);

  } while (blknoToDelete != newHead);
}

/* Initialize empty KeyArray */
static void
initKeyArray(KeyArray *keys, int32 maxvalues)
{
  keys->keys = palloc_array(Datum, maxvalues);
  keys->categories = palloc_array(GinNullCategory, maxvalues);
  keys->nvalues = 0;
  keys->maxvalues = maxvalues;
}

/* Add datum to KeyArray, resizing if needed */
static void
addDatum(KeyArray *keys, Datum datum, GinNullCategory category)
{
  if (keys->nvalues >= keys->maxvalues)
  {
    keys->maxvalues *= 2;
    keys->keys = repalloc_array(keys->keys, Datum, keys->maxvalues);
    keys->categories = repalloc_array(keys->categories, GinNullCategory, keys->maxvalues);
  }

  keys->keys[keys->nvalues] = datum;
  keys->categories[keys->nvalues] = category;
  keys->nvalues++;
}

/*
 * Collect data from a pending-list page in preparation for insertion into
 * the main index.
 *
 * Go through all tuples >= startoff on page and collect values in accum
 *
 * Note that ka is just workspace --- it does not carry any state across
 * calls.
 */
static void
processPendingPage(BuildAccumulator *accum, KeyArray *ka,
           Page page, OffsetNumber startoff)
{
  ItemPointerData heapptr;
  OffsetNumber i,
        maxoff;
  OffsetNumber attrnum;

  /* reset *ka to empty */
  ka->nvalues = 0;

  maxoff = PageGetMaxOffsetNumber(page);
  Assert(maxoff >= FirstOffsetNumber);
  ItemPointerSetInvalid(&heapptr);
  attrnum = 0;

  for (i = startoff; i <= maxoff; i = OffsetNumberNext(i))
  {
    IndexTuple  itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
    OffsetNumber curattnum;
    Datum   curkey;
    GinNullCategory curcategory;

    /* Check for change of heap TID or attnum */
    curattnum = gintuple_get_attrnum(accum->ginstate, itup);

    if (!ItemPointerIsValid(&heapptr))
    {
      heapptr = itup->t_tid;
      attrnum = curattnum;
    }
    else if (!(ItemPointerEquals(&heapptr, &itup->t_tid) &&
           curattnum == attrnum))
    {
      /*
       * ginInsertBAEntries can insert several datums per call, but only
       * for one heap tuple and one column.  So call it at a boundary,
       * and reset ka.
       */
      ginInsertBAEntries(accum, &heapptr, attrnum,
                 ka->keys, ka->categories, ka->nvalues);
      ka->nvalues = 0;
      heapptr = itup->t_tid;
      attrnum = curattnum;
    }

    /* Add key to KeyArray */
    curkey = gintuple_get_key(accum->ginstate, itup, &curcategory);
    addDatum(ka, curkey, curcategory);
  }

  /* Dump out all remaining keys */
  ginInsertBAEntries(accum, &heapptr, attrnum,
             ka->keys, ka->categories, ka->nvalues);
}

/*
 * Move tuples from pending pages into regular GIN structure.
 *
 * On first glance it looks completely not crash-safe. But if we crash
 * after posting entries to the main index and before removing them from the
 * pending list, it's okay because when we redo the posting later on, nothing
 * bad will happen.
 *
 * fill_fsm indicates that ginInsertCleanup should add deleted pages
 * to FSM otherwise caller is responsible to put deleted pages into
 * FSM.
 *
 * If stats isn't null, we count deleted pending pages into the counts.
 */
void
ginInsertCleanup(GinState *ginstate, bool full_clean,
         bool fill_fsm, bool forceCleanup,
         IndexBulkDeleteResult *stats)
{
  Relation  index = ginstate->index;
  Buffer    metabuffer,
        buffer;
  Page    metapage,
        page;
  GinMetaPageData *metadata;
  MemoryContext opCtx,
        oldCtx;
  BuildAccumulator accum;
  KeyArray  datums;
  BlockNumber blkno,
        blknoFinish;
  bool    cleanupFinish = false;
  bool    fsm_vac = false;
  int     workMemory;

  /*
   * We would like to prevent concurrent cleanup process. For that we will
   * lock metapage in exclusive mode using LockPage() call. Nobody other
   * will use that lock for metapage, so we keep possibility of concurrent
   * insertion into pending list
   */

  if (forceCleanup)
  {
    /*
     * We are called from [auto]vacuum/analyze or gin_clean_pending_list()
     * and we would like to wait concurrent cleanup to finish.
     */
    LockPage(index, GIN_METAPAGE_BLKNO, ExclusiveLock);
    workMemory =
      (AmAutoVacuumWorkerProcess() && autovacuum_work_mem != -1) ?
      autovacuum_work_mem : maintenance_work_mem;
  }
  else
  {
    /*
     * We are called from regular insert and if we see concurrent cleanup
     * just exit in hope that concurrent process will clean up pending
     * list.
     */
    if (!ConditionalLockPage(index, GIN_METAPAGE_BLKNO, ExclusiveLock))
      return;
    workMemory = work_mem;
  }

  metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
  LockBuffer(metabuffer, GIN_SHARE);
  metapage = BufferGetPage(metabuffer);
  metadata = GinPageGetMeta(metapage);

  if (metadata->head == InvalidBlockNumber)
  {
    /* Nothing to do */
    UnlockReleaseBuffer(metabuffer);
    UnlockPage(index, GIN_METAPAGE_BLKNO, ExclusiveLock);
    return;
  }

  /*
   * Remember a tail page to prevent infinite cleanup if other backends add
   * new tuples faster than we can cleanup.
   */
  blknoFinish = metadata->tail;

  /*
   * Read and lock head of pending list
   */
  blkno = metadata->head;
  buffer = ReadBuffer(index, blkno);
  LockBuffer(buffer, GIN_SHARE);
  page = BufferGetPage(buffer);

  LockBuffer(metabuffer, GIN_UNLOCK);

  /*
   * Initialize.  All temporary space will be in opCtx
   */
  opCtx = AllocSetContextCreate(CurrentMemoryContext,
                  "GIN insert cleanup temporary context",
                  ALLOCSET_DEFAULT_SIZES);

  oldCtx = MemoryContextSwitchTo(opCtx);

  initKeyArray(&datums, 128);
  ginInitBA(&accum);
  accum.ginstate = ginstate;

  /*
   * At the top of this loop, we have pin and lock on the current page of
   * the pending list.  However, we'll release that before exiting the loop.
   * Note we also have pin but not lock on the metapage.
   */
  for (;;)
  {
    Assert(!GinPageIsDeleted(page));

    /*
     * Are we walk through the page which as we remember was a tail when
     * we start our cleanup?  But if caller asks us to clean up whole
     * pending list then ignore old tail, we will work until list becomes
     * empty.
     */
    if (blkno == blknoFinish && full_clean == false)
      cleanupFinish = true;

    /*
     * read page's datums into accum
     */
    processPendingPage(&accum, &datums, page, FirstOffsetNumber);

    vacuum_delay_point(false);

    /*
     * Is it time to flush memory to disk?  Flush if we are at the end of
     * the pending list, or if we have a full row and memory is getting
     * full.
     */
    if (GinPageGetOpaque(page)->rightlink == InvalidBlockNumber ||
      (GinPageHasFullRow(page) &&
       accum.allocatedMemory >= workMemory * (Size) 1024))
    {
      ItemPointerData *list;
      uint32    nlist;
      Datum   key;
      GinNullCategory category;
      OffsetNumber maxoff,
            attnum;

      /*
       * Unlock current page to increase performance. Changes of page
       * will be checked later by comparing maxoff after completion of
       * memory flush.
       */
      maxoff = PageGetMaxOffsetNumber(page);
      LockBuffer(buffer, GIN_UNLOCK);

      /*
       * Moving collected data into regular structure can take
       * significant amount of time - so, run it without locking pending
       * list.
       */
      ginBeginBAScan(&accum);
      while ((list = ginGetBAEntry(&accum,
                     &attnum, &key, &category, &nlist)) != NULL)
      {
        ginEntryInsert(ginstate, attnum, key, category,
                 list, nlist, NULL);
        vacuum_delay_point(false);
      }

      /*
       * Lock the whole list to remove pages
       */
      LockBuffer(metabuffer, GIN_EXCLUSIVE);
      LockBuffer(buffer, GIN_SHARE);

      Assert(!GinPageIsDeleted(page));

      /*
       * While we left the page unlocked, more stuff might have gotten
       * added to it.  If so, process those entries immediately.  There
       * shouldn't be very many, so we don't worry about the fact that
       * we're doing this with exclusive lock. Insertion algorithm
       * guarantees that inserted row(s) will not continue on next page.
       * NOTE: intentionally no vacuum_delay_point in this loop.
       */
      if (PageGetMaxOffsetNumber(page) != maxoff)
      {
        ginInitBA(&accum);
        processPendingPage(&accum, &datums, page, maxoff + 1);

        ginBeginBAScan(&accum);
        while ((list = ginGetBAEntry(&accum,
                       &attnum, &key, &category, &nlist)) != NULL)
          ginEntryInsert(ginstate, attnum, key, category,
                   list, nlist, NULL);
      }

      /*
       * Remember next page - it will become the new list head
       */
      blkno = GinPageGetOpaque(page)->rightlink;
      UnlockReleaseBuffer(buffer);  /* shiftList will do exclusive
                       * locking */

      /*
       * remove read pages from pending list, at this point all content
       * of read pages is in regular structure
       */
      shiftList(index, metabuffer, blkno, fill_fsm, stats);

      /* At this point, some pending pages have been freed up */
      fsm_vac = true;

      Assert(blkno == metadata->head);
      LockBuffer(metabuffer, GIN_UNLOCK);

      /*
       * if we removed the whole pending list or we cleanup tail (which
       * we remembered on start our cleanup process) then just exit
       */
      if (blkno == InvalidBlockNumber || cleanupFinish)
        break;

      /*
       * release memory used so far and reinit state
       */
      MemoryContextReset(opCtx);
      initKeyArray(&datums, datums.maxvalues);
      ginInitBA(&accum);
    }
    else
    {
      blkno = GinPageGetOpaque(page)->rightlink;
      UnlockReleaseBuffer(buffer);
    }

    /*
     * Read next page in pending list
     */
    vacuum_delay_point(false);
    buffer = ReadBuffer(index, blkno);
    LockBuffer(buffer, GIN_SHARE);
    page = BufferGetPage(buffer);
  }

  UnlockPage(index, GIN_METAPAGE_BLKNO, ExclusiveLock);
  ReleaseBuffer(metabuffer);

  /*
   * As pending list pages can have a high churn rate, it is desirable to
   * recycle them immediately to the FreeSpaceMap when ordinary backends
   * clean the list.
   */
  if (fsm_vac && fill_fsm)
    IndexFreeSpaceMapVacuum(index);

  /* Clean up temporary space */
  MemoryContextSwitchTo(oldCtx);
  MemoryContextDelete(opCtx);
}

/*
 * SQL-callable function to clean the insert pending list
 */
Datum
gin_clean_pending_list(PG_FUNCTION_ARGS)
{
  Oid     indexoid = PG_GETARG_OID(0);
  Relation  indexRel = index_open(indexoid, RowExclusiveLock);
  IndexBulkDeleteResult stats;

  if (RecoveryInProgress())
    ereport(ERROR,
        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
         errmsg("recovery is in progress"),
         errhint("GIN pending list cannot be cleaned up during recovery.")));

  /* Must be a GIN index */
  if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
    indexRel->rd_rel->relam != GIN_AM_OID)
    ereport(ERROR,
        (errcode(ERRCODE_WRONG_OBJECT_TYPE),
         errmsg("\"%s\" is not a GIN index",
            RelationGetRelationName(indexRel))));

  /*
   * Reject attempts to read non-local temporary relations; we would be
   * likely to get wrong data since we have no visibility into the owning
   * session's local buffers.
   */
  if (RELATION_IS_OTHER_TEMP(indexRel))
    ereport(ERROR,
        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
         errmsg("cannot access temporary indexes of other sessions")));

  /* User must own the index (comparable to privileges needed for VACUUM) */
  if (!object_ownercheck(RelationRelationId, indexoid, GetUserId()))
    aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
             RelationGetRelationName(indexRel));

  memset(&stats, 0, sizeof(stats));

  /*
   * Can't assume anything about the content of an !indisready index.  Make
   * those a no-op, not an error, so users can just run this function on all
   * indexes of the access method.  Since an indisready&&!indisvalid index
   * is merely awaiting missed aminsert calls, we're capable of processing
   * it.  Decline to do so, out of an abundance of caution.
   */
  if (indexRel->rd_index->indisvalid)
  {
    GinState  ginstate;

    initGinState(&ginstate, indexRel);
    ginInsertCleanup(&ginstate, true, true, true, &stats);
  }
  else
    ereport(DEBUG1,
        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
         errmsg("index \"%s\" is not valid",
            RelationGetRelationName(indexRel))));

  index_close(indexRel, RowExclusiveLock);

  PG_RETURN_INT64((int64) stats.pages_deleted);
}

Coverage Report

Created: 2026-08-13 07:12