Quelle  gistxlog.c

/*-------------------------------------------------------------------------
 *
 * gistxlog.c
 *   WAL replay logic for GiST.
 *
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *    src/backend/access/gist/gistxlog.c
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/gist_private.h"
#include "access/gistxlog.h"
#include "access/transam.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "storage/standby.h"
#include "utils/memutils.h"
#include "utils/rel.h"

static MemoryContext opCtx;  /* working memory for operations */

/*
 * Replay the clearing of F_FOLLOW_RIGHT flag on a child page.
 *
 * Even if the WAL record includes a full-page image, we have to update the
 * follow-right flag, because that change is not included in the full-page
 * image.  To be sure that the intermediate state with the wrong flag value is
 * not visible to concurrent Hot Standby queries, this function handles
 * restoring the full-page image as well as updating the flag.  (Note that
 * we never need to do anything else to the child page in the current WAL
 * action.)
 */
static void
gistRedoClearFollowRight(XLogReaderState *record, uint8 block_id)
{
 XLogRecPtr lsn = record->EndRecPtr;
 Buffer  buffer;
 Page  page;
 XLogRedoAction action;

 /*
  * Note that we still update the page even if it was restored from a full
  * page image, because the updated NSN is not included in the image.
 */
 action = XLogReadBufferForRedo(record, block_id, &buffer);
 if (action == BLK_NEEDS_REDO || action == BLK_RESTORED)
 {
  page = BufferGetPage(buffer);

  GistPageSetNSN(page, lsn);
  GistClearFollowRight(page);

  PageSetLSN(page, lsn);
  MarkBufferDirty(buffer);
 }
 if (BufferIsValid(buffer))
  UnlockReleaseBuffer(buffer);
}

/*
 * redo any page update (except page split)
 */
static void
gistRedoPageUpdateRecord(XLogReaderState *record)
{
 XLogRecPtr lsn = record->EndRecPtr;
 gistxlogPageUpdate *xldata = (gistxlogPageUpdate *) XLogRecGetData(record);
 Buffer  buffer;
 Page  page;

 if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
 {
  char    *begin;
  char    *data;
  Size  datalen;
  int   ninserted PG_USED_FOR_ASSERTS_ONLY = 0;

  data = begin = XLogRecGetBlockData(record, 0, &datalen);

  page = (Page) BufferGetPage(buffer);

  if (xldata->ntodelete == 1 && xldata->ntoinsert == 1)
  {
   /*
    * When replacing one tuple with one other tuple, we must use
    * PageIndexTupleOverwrite for consistency with gistplacetopage.
 */
   OffsetNumber offnum = *((OffsetNumber *) data);
   IndexTuple itup;
   Size  itupsize;

   data += sizeof(OffsetNumber);
   itup = (IndexTuple) data;
   itupsize = IndexTupleSize(itup);
   if (!PageIndexTupleOverwrite(page, offnum, (Item) itup, itupsize))
    elog(ERROR, "failed to add item to GiST index page, size %d bytes",
      (int) itupsize);
   data += itupsize;
   /* should be nothing left after consuming 1 tuple */
   Assert(data - begin == datalen);
   /* update insertion count for assert check below */
   ninserted++;
  }
  else if (xldata->ntodelete > 0)
  {
   /* Otherwise, delete old tuples if any */
   OffsetNumber *todelete = (OffsetNumber *) data;

   data += sizeof(OffsetNumber) * xldata->ntodelete;

   PageIndexMultiDelete(page, todelete, xldata->ntodelete);
   if (GistPageIsLeaf(page))
    GistMarkTuplesDeleted(page);
  }

  /* Add new tuples if any */
  if (data - begin < datalen)
  {
   OffsetNumber off = (PageIsEmpty(page)) ? FirstOffsetNumber :
    OffsetNumberNext(PageGetMaxOffsetNumber(page));

   while (data - begin < datalen)
   {
    IndexTuple itup = (IndexTuple) data;
    Size  sz = IndexTupleSize(itup);
    OffsetNumber l;

    data += sz;

    l = PageAddItem(page, (Item) itup, sz, off, false, false);
    if (l == InvalidOffsetNumber)
     elog(ERROR, "failed to add item to GiST index page, size %d bytes",
       (int) sz);
    off++;
    ninserted++;
   }
  }

  /* Check that XLOG record contained expected number of tuples */
  Assert(ninserted == xldata->ntoinsert);

  PageSetLSN(page, lsn);
  MarkBufferDirty(buffer);
 }

 /*
  * Fix follow-right data on left child page
  *
  * This must be done while still holding the lock on the target page. Note
  * that even if the target page no longer exists, we still attempt to
  * replay the change on the child page.
 */
 if (XLogRecHasBlockRef(record, 1))
  gistRedoClearFollowRight(record, 1);

 if (BufferIsValid(buffer))
  UnlockReleaseBuffer(buffer);
}


/*
 * redo delete on gist index page to remove tuples marked as DEAD during index
 * tuple insertion
 */
static void
gistRedoDeleteRecord(XLogReaderState *record)
{
 XLogRecPtr lsn = record->EndRecPtr;
 gistxlogDelete *xldata = (gistxlogDelete *) XLogRecGetData(record);
 Buffer  buffer;
 Page  page;
 OffsetNumber *toDelete = xldata->offsets;

 /*
  * If we have any conflict processing to do, it must happen before we
  * update the page.
  *
  * GiST delete records can conflict with standby queries.  You might think
  * that vacuum records would conflict as well, but we've handled that
  * already.  XLOG_HEAP2_PRUNE_VACUUM_SCAN records provide the highest xid
  * cleaned by the vacuum of the heap and so we can resolve any conflicts
  * just once when that arrives.  After that we know that no conflicts
  * exist from individual gist vacuum records on that index.
 */
 if (InHotStandby)
 {
  RelFileLocator rlocator;

  XLogRecGetBlockTag(record, 0, &rlocator, NULL, NULL);

  ResolveRecoveryConflictWithSnapshot(xldata->snapshotConflictHorizon,
           xldata->isCatalogRel,
           rlocator);
 }

 if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
 {
  page = (Page) BufferGetPage(buffer);

  PageIndexMultiDelete(page, toDelete, xldata->ntodelete);

  GistClearPageHasGarbage(page);
  GistMarkTuplesDeleted(page);

  PageSetLSN(page, lsn);
  MarkBufferDirty(buffer);
 }

 if (BufferIsValid(buffer))
  UnlockReleaseBuffer(buffer);
}

/*
 * Returns an array of index pointers.
 */
static IndexTuple *
decodePageSplitRecord(char *begin, int len, int *n)
{
 char    *ptr;
 int   i = 0;
 IndexTuple *tuples;

 /* extract the number of tuples */
 memcpy(n, begin, sizeof(int));
 ptr = begin + sizeof(int);

 tuples = palloc(*n * sizeof(IndexTuple));

 for (i = 0; i < *n; i++)
 {
  Assert(ptr - begin < len);
  tuples[i] = (IndexTuple) ptr;
  ptr += IndexTupleSize((IndexTuple) ptr);
 }
 Assert(ptr - begin == len);

 return tuples;
}

static void
gistRedoPageSplitRecord(XLogReaderState *record)
{
 XLogRecPtr lsn = record->EndRecPtr;
 gistxlogPageSplit *xldata = (gistxlogPageSplit *) XLogRecGetData(record);
 Buffer  firstbuffer = InvalidBuffer;
 Buffer  buffer;
 Page  page;
 int   i;
 bool  isrootsplit = false;

 /*
  * We must hold lock on the first-listed page throughout the action,
  * including while updating the left child page (if any).  We can unlock
  * remaining pages in the list as soon as they've been written, because
  * there is no path for concurrent queries to reach those pages without
  * first visiting the first-listed page.
 */

 /* loop around all pages */
 for (i = 0; i < xldata->npage; i++)
 {
  int   flags;
  char    *data;
  Size  datalen;
  int   num;
  BlockNumber blkno;
  IndexTuple *tuples;

  XLogRecGetBlockTag(record, i + 1, NULL, NULL, &blkno);
  if (blkno == GIST_ROOT_BLKNO)
  {
   Assert(i == 0);
   isrootsplit = true;
  }

  buffer = XLogInitBufferForRedo(record, i + 1);
  page = (Page) BufferGetPage(buffer);
  data = XLogRecGetBlockData(record, i + 1, &datalen);

  tuples = decodePageSplitRecord(data, datalen, &num);

  /* ok, clear buffer */
  if (xldata->origleaf && blkno != GIST_ROOT_BLKNO)
   flags = F_LEAF;
  else
   flags = 0;
  GISTInitBuffer(buffer, flags);

  /* and fill it */
  gistfillbuffer(page, tuples, num, FirstOffsetNumber);

  if (blkno == GIST_ROOT_BLKNO)
  {
   GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
   GistPageSetNSN(page, xldata->orignsn);
   GistClearFollowRight(page);
  }
  else
  {
   if (i < xldata->npage - 1)
   {
    BlockNumber nextblkno;

    XLogRecGetBlockTag(record, i + 2, NULL, NULL, &nextblkno);
    GistPageGetOpaque(page)->rightlink = nextblkno;
   }
   else
    GistPageGetOpaque(page)->rightlink = xldata->origrlink;
   GistPageSetNSN(page, xldata->orignsn);
   if (i < xldata->npage - 1 && !isrootsplit &&
    xldata->markfollowright)
    GistMarkFollowRight(page);
   else
    GistClearFollowRight(page);
  }

  PageSetLSN(page, lsn);
  MarkBufferDirty(buffer);

  if (i == 0)
   firstbuffer = buffer;
  else
   UnlockReleaseBuffer(buffer);
 }

 /* Fix follow-right data on left child page, if any */
 if (XLogRecHasBlockRef(record, 0))
  gistRedoClearFollowRight(record, 0);

 /* Finally, release lock on the first page */
 UnlockReleaseBuffer(firstbuffer);
}

/* redo page deletion */
static void
gistRedoPageDelete(XLogReaderState *record)
{
 XLogRecPtr lsn = record->EndRecPtr;
 gistxlogPageDelete *xldata = (gistxlogPageDelete *) XLogRecGetData(record);
 Buffer  parentBuffer;
 Buffer  leafBuffer;

 if (XLogReadBufferForRedo(record, 0, &leafBuffer) == BLK_NEEDS_REDO)
 {
  Page  page = (Page) BufferGetPage(leafBuffer);

  GistPageSetDeleted(page, xldata->deleteXid);

  PageSetLSN(page, lsn);
  MarkBufferDirty(leafBuffer);
 }

 if (XLogReadBufferForRedo(record, 1, &parentBuffer) == BLK_NEEDS_REDO)
 {
  Page  page = (Page) BufferGetPage(parentBuffer);

  PageIndexTupleDelete(page, xldata->downlinkOffset);

  PageSetLSN(page, lsn);
  MarkBufferDirty(parentBuffer);
 }

 if (BufferIsValid(parentBuffer))
  UnlockReleaseBuffer(parentBuffer);
 if (BufferIsValid(leafBuffer))
  UnlockReleaseBuffer(leafBuffer);
}

static void
gistRedoPageReuse(XLogReaderState *record)
{
 gistxlogPageReuse *xlrec = (gistxlogPageReuse *) XLogRecGetData(record);

 /*
  * PAGE_REUSE records exist to provide a conflict point when we reuse
  * pages in the index via the FSM.  That's all they do though.
  *
  * snapshotConflictHorizon was the page's deleteXid.  The
  * GlobalVisCheckRemovableFullXid(deleteXid) test in gistPageRecyclable()
  * conceptually mirrors the PGPROC->xmin > limitXmin test in
  * GetConflictingVirtualXIDs().  Consequently, one XID value achieves the
  * same exclusion effect on primary and standby.
 */
 if (InHotStandby)
  ResolveRecoveryConflictWithSnapshotFullXid(xlrec->snapshotConflictHorizon,
               xlrec->isCatalogRel,
               xlrec->locator);
}

void
gist_redo(XLogReaderState *record)
{
 uint8  info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 MemoryContext oldCxt;

 /*
  * GiST indexes do not require any conflict processing. NB: If we ever
  * implement a similar optimization we have in b-tree, and remove killed
  * tuples outside VACUUM, we'll need to handle that here.
 */

 oldCxt = MemoryContextSwitchTo(opCtx);
 switch (info)
 {
  case XLOG_GIST_PAGE_UPDATE:
   gistRedoPageUpdateRecord(record);
   break;
  case XLOG_GIST_DELETE:
   gistRedoDeleteRecord(record);
   break;
  case XLOG_GIST_PAGE_REUSE:
   gistRedoPageReuse(record);
   break;
  case XLOG_GIST_PAGE_SPLIT:
   gistRedoPageSplitRecord(record);
   break;
  case XLOG_GIST_PAGE_DELETE:
   gistRedoPageDelete(record);
   break;
  case XLOG_GIST_ASSIGN_LSN:
   /* nop. See gistGetFakeLSN(). */
   break;
  default:
   elog(PANIC, "gist_redo: unknown op code %u", info);
 }

 MemoryContextSwitchTo(oldCxt);
 MemoryContextReset(opCtx);
}

void
gist_xlog_startup(void)
{
 opCtx = createTempGistContext();
}

void
gist_xlog_cleanup(void)
{
 MemoryContextDelete(opCtx);
}

/*
 * Mask a Gist page before running consistency checks on it.
 */
void
gist_mask(char *pagedata, BlockNumber blkno)
{
 Page  page = (Page) pagedata;

 mask_page_lsn_and_checksum(page);

 mask_page_hint_bits(page);
 mask_unused_space(page);

 /*
  * NSN is nothing but a special purpose LSN. Hence, mask it for the same
  * reason as mask_page_lsn_and_checksum.
 */
 GistPageSetNSN(page, (uint64) MASK_MARKER);

 /*
  * We update F_FOLLOW_RIGHT flag on the left child after writing WAL
  * record. Hence, mask this flag. See gistplacetopage() for details.
 */
 GistMarkFollowRight(page);

 if (GistPageIsLeaf(page))
 {
  /*
   * In gist leaf pages, it is possible to modify the LP_FLAGS without
   * emitting any WAL record. Hence, mask the line pointer flags. See
   * gistkillitems() for details.
 */
  mask_lp_flags(page);
 }

 /*
  * During gist redo, we never mark a page as garbage. Hence, mask it to
  * ignore any differences.
 */
 GistClearPageHasGarbage(page);
}

/*
 * Write WAL record of a page split.
 */
XLogRecPtr
gistXLogSplit(bool page_is_leaf,
     SplitPageLayout *dist,
     BlockNumber origrlink, GistNSN orignsn,
     Buffer leftchildbuf, bool markfollowright)
{
 gistxlogPageSplit xlrec;
 SplitPageLayout *ptr;
 int   npage = 0;
 XLogRecPtr recptr;
 int   i;

 for (ptr = dist; ptr; ptr = ptr->next)
  npage++;

 xlrec.origrlink = origrlink;
 xlrec.orignsn = orignsn;
 xlrec.origleaf = page_is_leaf;
 xlrec.npage = (uint16) npage;
 xlrec.markfollowright = markfollowright;

 XLogBeginInsert();

 /*
  * Include a full page image of the child buf. (only necessary if a
  * checkpoint happened since the child page was split)
 */
 if (BufferIsValid(leftchildbuf))
  XLogRegisterBuffer(0, leftchildbuf, REGBUF_STANDARD);

 /*
  * NOTE: We register a lot of data. The caller must've called
  * XLogEnsureRecordSpace() to prepare for that. We cannot do it here,
  * because we're already in a critical section. If you change the number
  * of buffer or data registrations here, make sure you modify the
  * XLogEnsureRecordSpace() calls accordingly!
 */
 XLogRegisterData(&xlrec, sizeof(gistxlogPageSplit));

 i = 1;
 for (ptr = dist; ptr; ptr = ptr->next)
 {
  XLogRegisterBuffer(i, ptr->buffer, REGBUF_WILL_INIT);
  XLogRegisterBufData(i, &(ptr->block.num), sizeof(int));
  XLogRegisterBufData(i, ptr->list, ptr->lenlist);
  i++;
 }

 recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT);

 return recptr;
}

/*
 * Write XLOG record describing a page deletion. This also includes removal of
 * downlink from the parent page.
 */
XLogRecPtr
gistXLogPageDelete(Buffer buffer, FullTransactionId xid,
       Buffer parentBuffer, OffsetNumber downlinkOffset)
{
 gistxlogPageDelete xlrec;
 XLogRecPtr recptr;

 xlrec.deleteXid = xid;
 xlrec.downlinkOffset = downlinkOffset;

 XLogBeginInsert();
 XLogRegisterData(&xlrec, SizeOfGistxlogPageDelete);

 XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
 XLogRegisterBuffer(1, parentBuffer, REGBUF_STANDARD);

 recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_DELETE);

 return recptr;
}

/*
 * Write an empty XLOG record to assign a distinct LSN.
 */
XLogRecPtr
gistXLogAssignLSN(void)
{
 int   dummy = 0;

 /*
  * Records other than XLOG_SWITCH must have content. We use an integer 0
  * to follow the restriction.
 */
 XLogBeginInsert();
 XLogSetRecordFlags(XLOG_MARK_UNIMPORTANT);
 XLogRegisterData(&dummy, sizeof(dummy));
 return XLogInsert(RM_GIST_ID, XLOG_GIST_ASSIGN_LSN);
}

/*
 * Write XLOG record about reuse of a deleted page.
 */
void
gistXLogPageReuse(Relation rel, Relation heaprel,
      BlockNumber blkno, FullTransactionId deleteXid)
{
 gistxlogPageReuse xlrec_reuse;

 /*
  * Note that we don't register the buffer with the record, because this
  * operation doesn't modify the page. This record only exists to provide a
  * conflict point for Hot Standby.
 */

 /* XLOG stuff */
 xlrec_reuse.isCatalogRel = RelationIsAccessibleInLogicalDecoding(heaprel);
 xlrec_reuse.locator = rel->rd_locator;
 xlrec_reuse.block = blkno;
 xlrec_reuse.snapshotConflictHorizon = deleteXid;

 XLogBeginInsert();
 XLogRegisterData(&xlrec_reuse, SizeOfGistxlogPageReuse);

 XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_REUSE);
}

/*
 * Write XLOG record describing a page update. The update can include any
 * number of deletions and/or insertions of tuples on a single index page.
 *
 * If this update inserts a downlink for a split page, also record that
 * the F_FOLLOW_RIGHT flag on the child page is cleared and NSN set.
 *
 * Note that both the todelete array and the tuples are marked as belonging
 * to the target buffer; they need not be stored in XLOG if XLogInsert decides
 * to log the whole buffer contents instead.
 */
XLogRecPtr
gistXLogUpdate(Buffer buffer,
      OffsetNumber *todelete, int ntodelete,
      IndexTuple *itup, int ituplen,
      Buffer leftchildbuf)
{
 gistxlogPageUpdate xlrec;
 int   i;
 XLogRecPtr recptr;

 xlrec.ntodelete = ntodelete;
 xlrec.ntoinsert = ituplen;

 XLogBeginInsert();
 XLogRegisterData(&xlrec, sizeof(gistxlogPageUpdate));

 XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
 XLogRegisterBufData(0, todelete, sizeof(OffsetNumber) * ntodelete);

 /* new tuples */
 for (i = 0; i < ituplen; i++)
  XLogRegisterBufData(0, itup[i], IndexTupleSize(itup[i]));

 /*
  * Include a full page image of the child buf. (only necessary if a
  * checkpoint happened since the child page was split)
 */
 if (BufferIsValid(leftchildbuf))
  XLogRegisterBuffer(1, leftchildbuf, REGBUF_STANDARD);

 recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_UPDATE);

 return recptr;
}

/*
 * Write XLOG record describing a delete of leaf index tuples marked as DEAD
 * during new tuple insertion.  One may think that this case is already covered
 * by gistXLogUpdate().  But deletion of index tuples might conflict with
 * standby queries and needs special handling.
 */
XLogRecPtr
gistXLogDelete(Buffer buffer, OffsetNumber *todelete, int ntodelete,
      TransactionId snapshotConflictHorizon, Relation heaprel)
{
 gistxlogDelete xlrec;
 XLogRecPtr recptr;

 xlrec.isCatalogRel = RelationIsAccessibleInLogicalDecoding(heaprel);
 xlrec.snapshotConflictHorizon = snapshotConflictHorizon;
 xlrec.ntodelete = ntodelete;

 XLogBeginInsert();
 XLogRegisterData(&xlrec, SizeOfGistxlogDelete);

 /*
  * 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(todelete, ntodelete * sizeof(OffsetNumber));

 XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);

 recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_DELETE);

 return recptr;
}