open-dsi
/
postgres
mirror of https://github.com/postgres/postgres
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix multiple problems in WAL replay.

Browse Source 
Most of the replay functions for WAL record types that modify more than
one page failed to ensure that those pages were locked correctly to ensure
that concurrent queries could not see inconsistent page states.  This is
a hangover from coding decisions made long before Hot Standby was added,
when it was hardly necessary to acquire buffer locks during WAL replay
at all, let alone hold them for carefully-chosen periods.

The key problem was that RestoreBkpBlocks was written to hold lock on each
page restored from a full-page image for only as long as it took to update
that page.  This was guaranteed to break any WAL replay function in which
there was any update-ordering constraint between pages, because even if the
nominal order of the pages is the right one, any mixture of full-page and
non-full-page updates in the same record would result in out-of-order
updates.  Moreover, it wouldn't work for situations where there's a
requirement to maintain lock on one page while updating another.  Failure
to honor an update ordering constraint in this way is thought to be the
cause of bug #7648 from Daniel Farina: what seems to have happened there
is that a btree page being split was rewritten from a full-page image
before the new right sibling page was written, and because lock on the
original page was not maintained it was possible for hot standby queries to
try to traverse the page's right-link to the not-yet-existing sibling page.

To fix, get rid of RestoreBkpBlocks as such, and instead create a new
function RestoreBackupBlock that restores just one full-page image at a
time.  This function can be invoked by WAL replay functions at the points
where they would otherwise perform non-full-page updates; in this way, the
physical order of page updates remains the same no matter which pages are
replaced by full-page images.  We can then further adjust the logic in
individual replay functions if it is necessary to hold buffer locks
for overlapping periods.  A side benefit is that we can simplify the
handling of concurrency conflict resolution by moving that code into the
record-type-specfic functions; there's no more need to contort the code
layout to keep conflict resolution in front of the RestoreBkpBlocks call.

In connection with that, standardize on zero-based numbering rather than
one-based numbering for referencing the full-page images.  In HEAD, I
removed the macros XLR_BKP_BLOCK_1 through XLR_BKP_BLOCK_4.  They are
still there in the header files in previous branches, but are no longer
used by the code.

In addition, fix some other bugs identified in the course of making these
changes:

spgRedoAddNode could fail to update the parent downlink at all, if the
parent tuple is in the same page as either the old or new split tuple and
we're not doing a full-page image: it would get fooled by the LSN having
been advanced already.  This would result in permanent index corruption,
not just transient failure of concurrent queries.

Also, ginHeapTupleFastInsert's "merge lists" case failed to mark the old
tail page as a candidate for a full-page image; in the worst case this
could result in torn-page corruption.

heap_xlog_freeze() was inconsistent about using a cleanup lock or plain
exclusive lock: it did the former in the normal path but the latter for a
full-page image.  A plain exclusive lock seems sufficient, so change to
that.

Also, remove gistRedoPageDeleteRecord(), which has been dead code since
VACUUM FULL was rewritten.

Back-patch to 9.0, where hot standby was introduced.  Note however that 9.0
had a significantly different WAL-logging scheme for GIST index updates,
and it doesn't appear possible to make that scheme safe for concurrent hot
standby queries, because it can leave inconsistent states in the index even
between WAL records.  Given the lack of complaints from the field, we won't
work too hard on fixing that branch.
pull/3/head
Tom Lane 13 years ago
parent 9b3ac49e5a
commit 3bbf668de9
10 changed files with 753 additions and 464 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 10
      src/backend/access/gin/ginfast.c
  2. 101
      src/backend/access/gin/ginxlog.c
  3. 169
      src/backend/access/gist/gistxlog.c
  4. 198
      src/backend/access/heap/heapam.c
  5. 190
      src/backend/access/nbtree/nbtxlog.c
  6. 259
      src/backend/access/spgist/spgxlog.c
  7. 59
      src/backend/access/transam/README
  8. 67
      src/backend/access/transam/xlog.c
  9. 8
      src/include/access/gist_private.h
  10. 16
      src/include/access/xlog.h

10
src/backend/access/gin/ginfast.c
Unescape View File

@ -290,7 +290,7 @@ ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector)
if (metadata->head == InvalidBlockNumber) if (metadata->head == InvalidBlockNumber)
{ {
/* /*
* Main list is empty, so just copy sublist into main list * Main list is empty, so just insert sublist as main list
*/ */
START_CRIT_SECTION(); START_CRIT_SECTION();
@ -313,6 +313,14 @@ ginHeapTupleFastInsert(GinState *ginstate, GinTupleCollector *collector)
LockBuffer(buffer, GIN_EXCLUSIVE); LockBuffer(buffer, GIN_EXCLUSIVE);
page = BufferGetPage(buffer); page = BufferGetPage(buffer);
rdata[0].next = rdata + 1;
rdata[1].buffer = buffer;
rdata[1].buffer_std = true;
rdata[1].data = NULL;
rdata[1].len = 0;
rdata[1].next = NULL;
Assert(GinPageGetOpaque(page)->rightlink == InvalidBlockNumber); Assert(GinPageGetOpaque(page)->rightlink == InvalidBlockNumber);
START_CRIT_SECTION(); START_CRIT_SECTION();

101
src/backend/access/gin/ginxlog.c
Unescape View File

@ -77,6 +77,9 @@ ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
MetaBuffer; MetaBuffer;
Page page; Page page;
/* Backup blocks are not used in create_index records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
MetaBuffer = XLogReadBuffer(*node, GIN_METAPAGE_BLKNO, true); MetaBuffer = XLogReadBuffer(*node, GIN_METAPAGE_BLKNO, true);
Assert(BufferIsValid(MetaBuffer)); Assert(BufferIsValid(MetaBuffer));
page = (Page) BufferGetPage(MetaBuffer); page = (Page) BufferGetPage(MetaBuffer);
@ -109,6 +112,9 @@ ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/* Backup blocks are not used in create_ptree records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
buffer = XLogReadBuffer(data->node, data->blkno, true); buffer = XLogReadBuffer(data->node, data->blkno, true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
@ -159,9 +165,12 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
} }
} }
/* nothing else to do if page was backed up */ /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK_1) if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(data->node, data->blkno, false); buffer = XLogReadBuffer(data->node, data->blkno, false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
@ -256,6 +265,9 @@ ginRedoSplit(XLogRecPtr lsn, XLogRecord *record)
if (data->isData) if (data->isData)
flags |= GIN_DATA; flags |= GIN_DATA;
/* Backup blocks are not used in split records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
lbuffer = XLogReadBuffer(data->node, data->lblkno, true); lbuffer = XLogReadBuffer(data->node, data->lblkno, true);
Assert(BufferIsValid(lbuffer)); Assert(BufferIsValid(lbuffer));
lpage = (Page) BufferGetPage(lbuffer); lpage = (Page) BufferGetPage(lbuffer);
@ -369,9 +381,12 @@ ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/* nothing to do if page was backed up (and no info to do it with) */ /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK_1) if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(data->node, data->blkno, false); buffer = XLogReadBuffer(data->node, data->blkno, false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
@ -420,33 +435,38 @@ static void
ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record) ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record)
{ {
ginxlogDeletePage *data = (ginxlogDeletePage *) XLogRecGetData(record); ginxlogDeletePage *data = (ginxlogDeletePage *) XLogRecGetData(record);
Buffer buffer; Buffer dbuffer;
Buffer pbuffer;
Buffer lbuffer;
Page page; Page page;
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
dbuffer = RestoreBackupBlock(lsn, record, 0, false, true);
else
{ {
buffer = XLogReadBuffer(data->node, data->blkno, false); dbuffer = XLogReadBuffer(data->node, data->blkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(dbuffer))
{ {
page = BufferGetPage(buffer); page = BufferGetPage(dbuffer);
if (!XLByteLE(lsn, PageGetLSN(page))) if (!XLByteLE(lsn, PageGetLSN(page)))
{ {
Assert(GinPageIsData(page)); Assert(GinPageIsData(page));
GinPageGetOpaque(page)->flags = GIN_DELETED; GinPageGetOpaque(page)->flags = GIN_DELETED;
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(dbuffer);
} }
UnlockReleaseBuffer(buffer);
} }
} }
if (!(record->xl_info & XLR_BKP_BLOCK_2)) if (record->xl_info & XLR_BKP_BLOCK(1))
pbuffer = RestoreBackupBlock(lsn, record, 1, false, true);
else
{ {
buffer = XLogReadBuffer(data->node, data->parentBlkno, false); pbuffer = XLogReadBuffer(data->node, data->parentBlkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(pbuffer))
{ {
page = BufferGetPage(buffer); page = BufferGetPage(pbuffer);
if (!XLByteLE(lsn, PageGetLSN(page))) if (!XLByteLE(lsn, PageGetLSN(page)))
{ {
Assert(GinPageIsData(page)); Assert(GinPageIsData(page));
@ -454,29 +474,35 @@ ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record)
GinPageDeletePostingItem(page, data->parentOffset); GinPageDeletePostingItem(page, data->parentOffset);
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(pbuffer);
} }
UnlockReleaseBuffer(buffer);
} }
} }
if (!(record->xl_info & XLR_BKP_BLOCK_3) && data->leftBlkno != InvalidBlockNumber) if (record->xl_info & XLR_BKP_BLOCK(2))
(void) RestoreBackupBlock(lsn, record, 2, false, false);
else if (data->leftBlkno != InvalidBlockNumber)
{ {
buffer = XLogReadBuffer(data->node, data->leftBlkno, false); lbuffer = XLogReadBuffer(data->node, data->leftBlkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(lbuffer))
{ {
page = BufferGetPage(buffer); page = BufferGetPage(lbuffer);
if (!XLByteLE(lsn, PageGetLSN(page))) if (!XLByteLE(lsn, PageGetLSN(page)))
{ {
Assert(GinPageIsData(page)); Assert(GinPageIsData(page));
GinPageGetOpaque(page)->rightlink = data->rightLink; GinPageGetOpaque(page)->rightlink = data->rightLink;
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(lbuffer);
} }
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(lbuffer);
} }
} }
if (BufferIsValid(pbuffer))
UnlockReleaseBuffer(pbuffer);
if (BufferIsValid(dbuffer))
UnlockReleaseBuffer(dbuffer);
} }
static void static void
@ -505,7 +531,9 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
/* /*
* insert into tail page * insert into tail page
*/ */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(data->node, data->metadata.tail, false); buffer = XLogReadBuffer(data->node, data->metadata.tail, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -553,6 +581,10 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
/* /*
* New tail * New tail
*/ */
if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{
buffer = XLogReadBuffer(data->node, data->prevTail, false); buffer = XLogReadBuffer(data->node, data->prevTail, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
{ {
@ -569,6 +601,7 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
} }
} }
}
UnlockReleaseBuffer(metabuffer); UnlockReleaseBuffer(metabuffer);
} }
@ -585,8 +618,12 @@ ginRedoInsertListPage(XLogRecPtr lsn, XLogRecord *record)
tupsize; tupsize;
IndexTuple tuples = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsertListPage)); IndexTuple tuples = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsertListPage));
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(data->node, data->blkno, true); buffer = XLogReadBuffer(data->node, data->blkno, true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
@ -632,6 +669,9 @@ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
Page metapage; Page metapage;
int i; int i;
/* Backup blocks are not used in delete_listpage records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false); metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
if (!BufferIsValid(metabuffer)) if (!BufferIsValid(metabuffer))
return; /* assume index was deleted, nothing to do */ return; /* assume index was deleted, nothing to do */
@ -645,6 +685,16 @@ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
MarkBufferDirty(metabuffer); MarkBufferDirty(metabuffer);
} }
/*
* In normal operation, shiftList() takes exclusive lock on all the
* pages-to-be-deleted simultaneously. During replay, however, it should
* be all right to lock them one at a time. This is dependent on the fact
* that we are deleting pages from the head of the list, and that readers
* share-lock the next page before releasing the one they are on. So we
* cannot get past a reader that is on, or due to visit, any page we are
* going to delete. New incoming readers will block behind our metapage
* lock and then see a fully updated page list.
*/
for (i = 0; i < data->ndeleted; i++) for (i = 0; i < data->ndeleted; i++)
{ {
Buffer buffer = XLogReadBuffer(data->node, data->toDelete[i], false); Buffer buffer = XLogReadBuffer(data->node, data->toDelete[i], false);
@ -678,7 +728,6 @@ gin_redo(XLogRecPtr lsn, XLogRecord *record)
* implement a similar optimization as we have in b-tree, and remove * implement a similar optimization as we have in b-tree, and remove
* killed tuples outside VACUUM, we'll need to handle that here. * killed tuples outside VACUUM, we'll need to handle that here.
*/ */
RestoreBkpBlocks(lsn, record, false);
topCtx = MemoryContextSwitchTo(opCtx); topCtx = MemoryContextSwitchTo(opCtx);
switch (info) switch (info)

169
src/backend/access/gist/gistxlog.c
Unescape View File

@ -32,23 +32,37 @@ typedef struct
static MemoryContext opCtx; /* working memory for operations */ static MemoryContext opCtx; /* working memory for operations */
/* /*
* Replay the clearing of F_FOLLOW_RIGHT flag. * 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 static void
gistRedoClearFollowRight(RelFileNode node, XLogRecPtr lsn, gistRedoClearFollowRight(XLogRecPtr lsn, XLogRecord *record, int block_index,
BlockNumber leftblkno) RelFileNode node, BlockNumber childblkno)
{ {
Buffer buffer; Buffer buffer;
Page page;
buffer = XLogReadBuffer(node, leftblkno, false); if (record->xl_info & XLR_BKP_BLOCK(block_index))
if (BufferIsValid(buffer)) buffer = RestoreBackupBlock(lsn, record, block_index, false, true);
else
{ {
Page page = (Page) BufferGetPage(buffer); buffer = XLogReadBuffer(node, childblkno, false);
if (!BufferIsValid(buffer))
return; /* page was deleted, nothing to do */
}
page = (Page) BufferGetPage(buffer);
/* /*
* Note that we still update the page even if page LSN is equal to the * Note that we still update the page even if page LSN is equal to the LSN
* LSN of this record, because the updated NSN is not included in the * of this record, because the updated NSN is not included in the full
* full page image. * page image.
*/ */
if (!XLByteLT(lsn, PageGetLSN(page))) if (!XLByteLT(lsn, PageGetLSN(page)))
{ {
@ -61,7 +75,6 @@ gistRedoClearFollowRight(RelFileNode node, XLogRecPtr lsn,
} }
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
} }
}
/* /*
* redo any page update (except page split) * redo any page update (except page split)
@ -75,18 +88,37 @@ gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record)
Page page; Page page;
char *data; char *data;
/*
* We need to acquire and hold lock on target page while updating the left
* child page. If we have a full-page image of target page, getting the
* lock is a side-effect of restoring that image. Note that even if the
* target page no longer exists, we'll still attempt to replay the change
* on the child page.
*/
if (record->xl_info & XLR_BKP_BLOCK(0))
buffer = RestoreBackupBlock(lsn, record, 0, false, true);
else
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
/* Fix follow-right data on left child page */
if (BlockNumberIsValid(xldata->leftchild)) if (BlockNumberIsValid(xldata->leftchild))
gistRedoClearFollowRight(xldata->node, lsn, xldata->leftchild); gistRedoClearFollowRight(lsn, record, 1,
xldata->node, xldata->leftchild);
/* nothing more to do if page was backed up (and no info to do it with) */ /* Done if target page no longer exists */
if (record->xl_info & XLR_BKP_BLOCK_1) if (!BufferIsValid(buffer))
return; return;
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); /* nothing more to do if page was backed up (and no info to do it with) */
if (!BufferIsValid(buffer)) if (record->xl_info & XLR_BKP_BLOCK(0))
{
UnlockReleaseBuffer(buffer);
return; return;
}
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
/* nothing more to do if change already applied */
if (XLByteLE(lsn, PageGetLSN(page))) if (XLByteLE(lsn, PageGetLSN(page)))
{ {
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
@ -140,39 +172,18 @@ gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record)
GistClearTuplesDeleted(page); GistClearTuplesDeleted(page);
} }
if (!GistPageIsLeaf(page) && PageGetMaxOffsetNumber(page) == InvalidOffsetNumber && xldata->blkno == GIST_ROOT_BLKNO) if (!GistPageIsLeaf(page) &&
PageGetMaxOffsetNumber(page) == InvalidOffsetNumber &&
xldata->blkno == GIST_ROOT_BLKNO)
{
/* /*
* all links on non-leaf root page was deleted by vacuum full, so root * all links on non-leaf root page was deleted by vacuum full, so root
* page becomes a leaf * page becomes a leaf
*/ */
GistPageSetLeaf(page); GistPageSetLeaf(page);
GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
UnlockReleaseBuffer(buffer);
} }
static void GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
gistRedoPageDeleteRecord(XLogRecPtr lsn, XLogRecord *record)
{
gistxlogPageDelete *xldata = (gistxlogPageDelete *) XLogRecGetData(record);
Buffer buffer;
Page page;
/* nothing else to do if page was backed up (and no info to do it with) */
if (record->xl_info & XLR_BKP_BLOCK_1)
return;
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
if (!BufferIsValid(buffer))
return;
page = (Page) BufferGetPage(buffer);
GistPageSetDeleted(page);
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(buffer);
@ -215,15 +226,22 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
{ {
gistxlogPageSplit *xldata = (gistxlogPageSplit *) XLogRecGetData(record); gistxlogPageSplit *xldata = (gistxlogPageSplit *) XLogRecGetData(record);
PageSplitRecord xlrec; PageSplitRecord xlrec;
Buffer firstbuffer = InvalidBuffer;
Buffer buffer; Buffer buffer;
Page page; Page page;
int i; int i;
bool isrootsplit = false; bool isrootsplit = false;
if (BlockNumberIsValid(xldata->leftchild))
gistRedoClearFollowRight(xldata->node, lsn, xldata->leftchild);
decodePageSplitRecord(&xlrec, record); decodePageSplitRecord(&xlrec, record);
/*
* 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 */ /* loop around all pages */
for (i = 0; i < xlrec.data->npage; i++) for (i = 0; i < xlrec.data->npage; i++)
{ {
@ -273,8 +291,20 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(buffer);
if (i == 0)
firstbuffer = buffer;
else
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
} }
/* Fix follow-right data on left child page, if any */
if (BlockNumberIsValid(xldata->leftchild))
gistRedoClearFollowRight(lsn, record, 0,
xldata->node, xldata->leftchild);
/* Finally, release lock on the first page */
UnlockReleaseBuffer(firstbuffer);
} }
static void static void
@ -284,6 +314,9 @@ gistRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/* Backup blocks are not used in create_index records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
buffer = XLogReadBuffer(*node, GIST_ROOT_BLKNO, true); buffer = XLogReadBuffer(*node, GIST_ROOT_BLKNO, true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
@ -308,7 +341,6 @@ gist_redo(XLogRecPtr lsn, XLogRecord *record)
* implement a similar optimization we have in b-tree, and remove killed * implement a similar optimization we have in b-tree, and remove killed
* tuples outside VACUUM, we'll need to handle that here. * tuples outside VACUUM, we'll need to handle that here.
*/ */
RestoreBkpBlocks(lsn, record, false);
oldCxt = MemoryContextSwitchTo(opCtx); oldCxt = MemoryContextSwitchTo(opCtx);
switch (info) switch (info)
@ -316,9 +348,6 @@ gist_redo(XLogRecPtr lsn, XLogRecord *record)
case XLOG_GIST_PAGE_UPDATE: case XLOG_GIST_PAGE_UPDATE:
gistRedoPageUpdateRecord(lsn, record); gistRedoPageUpdateRecord(lsn, record);
break; break;
case XLOG_GIST_PAGE_DELETE:
gistRedoPageDeleteRecord(lsn, record);
break;
case XLOG_GIST_PAGE_SPLIT: case XLOG_GIST_PAGE_SPLIT:
gistRedoPageSplitRecord(lsn, record); gistRedoPageSplitRecord(lsn, record);
break; break;
@ -347,14 +376,6 @@ out_gistxlogPageUpdate(StringInfo buf, gistxlogPageUpdate *xlrec)
appendStringInfo(buf, "; block number %u", xlrec->blkno); appendStringInfo(buf, "; block number %u", xlrec->blkno);
} }
static void
out_gistxlogPageDelete(StringInfo buf, gistxlogPageDelete *xlrec)
{
appendStringInfo(buf, "page_delete: rel %u/%u/%u; blkno %u",
xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode,
xlrec->blkno);
}
static void static void
out_gistxlogPageSplit(StringInfo buf, gistxlogPageSplit *xlrec) out_gistxlogPageSplit(StringInfo buf, gistxlogPageSplit *xlrec)
{ {
@ -375,9 +396,6 @@ gist_desc(StringInfo buf, uint8 xl_info, char *rec)
appendStringInfo(buf, "page_update: "); appendStringInfo(buf, "page_update: ");
out_gistxlogPageUpdate(buf, (gistxlogPageUpdate *) rec); out_gistxlogPageUpdate(buf, (gistxlogPageUpdate *) rec);
break; break;
case XLOG_GIST_PAGE_DELETE:
out_gistxlogPageDelete(buf, (gistxlogPageDelete *) rec);
break;
case XLOG_GIST_PAGE_SPLIT: case XLOG_GIST_PAGE_SPLIT:
out_gistxlogPageSplit(buf, (gistxlogPageSplit *) rec); out_gistxlogPageSplit(buf, (gistxlogPageSplit *) rec);
break; break;
@ -498,37 +516,30 @@ gistXLogUpdate(RelFileNode node, Buffer buffer,
Buffer leftchildbuf) Buffer leftchildbuf)
{ {
XLogRecData *rdata; XLogRecData *rdata;
gistxlogPageUpdate *xlrec; gistxlogPageUpdate xlrec;
int cur, int cur,
i; i;
XLogRecPtr recptr; XLogRecPtr recptr;
rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (4 + ituplen)); rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (3 + ituplen));
xlrec = (gistxlogPageUpdate *) palloc(sizeof(gistxlogPageUpdate));
xlrec->node = node; xlrec.node = node;
xlrec->blkno = BufferGetBlockNumber(buffer); xlrec.blkno = BufferGetBlockNumber(buffer);
xlrec->ntodelete = ntodelete; xlrec.ntodelete = ntodelete;
xlrec->leftchild = xlrec.leftchild =
BufferIsValid(leftchildbuf) ? BufferGetBlockNumber(leftchildbuf) : InvalidBlockNumber; BufferIsValid(leftchildbuf) ? BufferGetBlockNumber(leftchildbuf) : InvalidBlockNumber;
rdata[0].buffer = buffer; rdata[0].data = (char *) &xlrec;
rdata[0].buffer_std = true; rdata[0].len = sizeof(gistxlogPageUpdate);
rdata[0].data = NULL; rdata[0].buffer = InvalidBuffer;
rdata[0].len = 0;
rdata[0].next = &(rdata[1]); rdata[0].next = &(rdata[1]);
rdata[1].data = (char *) xlrec; rdata[1].data = (char *) todelete;
rdata[1].len = sizeof(gistxlogPageUpdate); rdata[1].len = sizeof(OffsetNumber) * ntodelete;
rdata[1].buffer = InvalidBuffer; rdata[1].buffer = buffer;
rdata[1].next = &(rdata[2]); rdata[1].buffer_std = true;
rdata[2].data = (char *) todelete;
rdata[2].len = sizeof(OffsetNumber) * ntodelete;
rdata[2].buffer = buffer;
rdata[2].buffer_std = true;
cur = 3; cur = 2;
/* new tuples */ /* new tuples */
for (i = 0; i < ituplen; i++) for (i = 0; i < ituplen; i++)

198
src/backend/access/heap/heapam.c
Unescape View File

@ -4620,6 +4620,9 @@ heap_xlog_cleanup_info(XLogRecPtr lsn, XLogRecord *record)
* conflict processing to occur before we begin index vacuum actions. see * conflict processing to occur before we begin index vacuum actions. see
* vacuumlazy.c and also comments in btvacuumpage() * vacuumlazy.c and also comments in btvacuumpage()
*/ */
/* Backup blocks are not used in cleanup_info records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
} }
/* /*
@ -4652,10 +4655,15 @@ heap_xlog_clean(XLogRecPtr lsn, XLogRecord *record)
ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
xlrec->node); xlrec->node);
RestoreBkpBlocks(lsn, record, true); /*
* If we have a full-page image, restore it (using a cleanup lock) and
if (record->xl_info & XLR_BKP_BLOCK_1) * we're done.
*/
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, true, false);
return; return;
}
buffer = XLogReadBufferExtended(xlrec->node, MAIN_FORKNUM, xlrec->block, RBM_NORMAL); buffer = XLogReadBufferExtended(xlrec->node, MAIN_FORKNUM, xlrec->block, RBM_NORMAL);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
@ -4721,15 +4729,16 @@ heap_xlog_freeze(XLogRecPtr lsn, XLogRecord *record)
if (InHotStandby) if (InHotStandby)
ResolveRecoveryConflictWithSnapshot(cutoff_xid, xlrec->node); ResolveRecoveryConflictWithSnapshot(cutoff_xid, xlrec->node);
RestoreBkpBlocks(lsn, record, false); /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
if (record->xl_info & XLR_BKP_BLOCK_1) {
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBufferExtended(xlrec->node, MAIN_FORKNUM, xlrec->block, RBM_NORMAL); buffer = XLogReadBuffer(xlrec->node, xlrec->block, false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
return; return;
LockBufferForCleanup(buffer);
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
if (XLByteLE(lsn, PageGetLSN(page))) if (XLByteLE(lsn, PageGetLSN(page)))
@ -4778,18 +4787,6 @@ heap_xlog_visible(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/*
* Read the heap page, if it still exists. If the heap file has been
* dropped or truncated later in recovery, this might fail. In that case,
* there's no point in doing anything further, since the visibility map
* will have to be cleared out at the same time.
*/
buffer = XLogReadBufferExtended(xlrec->node, MAIN_FORKNUM, xlrec->block,
RBM_NORMAL);
if (!BufferIsValid(buffer))
return;
page = (Page) BufferGetPage(buffer);
/* /*
* If there are any Hot Standby transactions running that have an xmin * If there are any Hot Standby transactions running that have an xmin
* horizon old enough that this page isn't all-visible for them, they * horizon old enough that this page isn't all-visible for them, they
@ -4802,19 +4799,31 @@ heap_xlog_visible(XLogRecPtr lsn, XLogRecord *record)
if (InHotStandby) if (InHotStandby)
ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, xlrec->node); ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, xlrec->node);
/*
* Read the heap page, if it still exists. If the heap file has been
* dropped or truncated later in recovery, we don't need to update the
* page, but we'd better still update the visibility map.
*/
buffer = XLogReadBufferExtended(xlrec->node, MAIN_FORKNUM, xlrec->block,
RBM_NORMAL);
if (BufferIsValid(buffer))
{
LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE); LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
page = (Page) BufferGetPage(buffer);
/* /*
* We don't bump the LSN of the heap page when setting the visibility map * We don't bump the LSN of the heap page when setting the visibility
* bit, because that would generate an unworkable volume of full-page * map bit, because that would generate an unworkable volume of
* writes. This exposes us to torn page hazards, but since we're not * full-page writes. This exposes us to torn page hazards, but since
* inspecting the existing page contents in any way, we don't care. * we're not inspecting the existing page contents in any way, we
* don't care.
* *
* However, all operations that clear the visibility map bit *do* bump the * However, all operations that clear the visibility map bit *do* bump
* LSN, and those operations will only be replayed if the XLOG LSN follows * the LSN, and those operations will only be replayed if the XLOG LSN
* the page LSN. Thus, if the page LSN has advanced past our XLOG * follows the page LSN. Thus, if the page LSN has advanced past our
* record's LSN, we mustn't mark the page all-visible, because the * XLOG record's LSN, we mustn't mark the page all-visible, because
* subsequent update won't be replayed to clear the flag. * the subsequent update won't be replayed to clear the flag.
*/ */
if (!XLByteLE(lsn, PageGetLSN(page))) if (!XLByteLE(lsn, PageGetLSN(page)))
{ {
@ -4824,15 +4833,16 @@ heap_xlog_visible(XLogRecPtr lsn, XLogRecord *record)
/* Done with heap page. */ /* Done with heap page. */
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
}
/* /*
* Even we skipped the heap page update due to the LSN interlock, it's * Even if we skipped the heap page update due to the LSN interlock, it's
* still safe to update the visibility map. Any WAL record that clears * still safe to update the visibility map. Any WAL record that clears
* the visibility map bit does so before checking the page LSN, so any * the visibility map bit does so before checking the page LSN, so any
* bits that need to be cleared will still be cleared. * bits that need to be cleared will still be cleared.
*/ */
if (record->xl_info & XLR_BKP_BLOCK_1) if (record->xl_info & XLR_BKP_BLOCK(0))
RestoreBkpBlocks(lsn, record, false); (void) RestoreBackupBlock(lsn, record, 0, false, false);
else else
{ {
Relation reln; Relation reln;
@ -4844,13 +4854,13 @@ heap_xlog_visible(XLogRecPtr lsn, XLogRecord *record)
/* /*
* Don't set the bit if replay has already passed this point. * Don't set the bit if replay has already passed this point.
* *
* It might be safe to do this unconditionally; if replay has past * It might be safe to do this unconditionally; if replay has passed
* this point, we'll replay at least as far this time as we did * this point, we'll replay at least as far this time as we did
* before, and if this bit needs to be cleared, the record responsible * before, and if this bit needs to be cleared, the record responsible
* for doing so should be again replayed, and clear it. For right * for doing so should be again replayed, and clear it. For right
* now, out of an abundance of conservatism, we use the same test here * now, out of an abundance of conservatism, we use the same test here
* we did for the heap page; if this results in a dropped bit, no real * we did for the heap page. If this results in a dropped bit, no
* harm is done; and the next VACUUM will fix it. * real harm is done; and the next VACUUM will fix it.
*/ */
if (!XLByteLE(lsn, PageGetLSN(BufferGetPage(vmbuffer)))) if (!XLByteLE(lsn, PageGetLSN(BufferGetPage(vmbuffer))))
visibilitymap_set(reln, xlrec->block, lsn, vmbuffer, visibilitymap_set(reln, xlrec->block, lsn, vmbuffer,
@ -4868,6 +4878,9 @@ heap_xlog_newpage(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/* Backup blocks are not used in newpage records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
/* /*
* Note: the NEWPAGE log record is used for both heaps and indexes, so do * Note: the NEWPAGE log record is used for both heaps and indexes, so do
* not do anything that assumes we are touching a heap. * not do anything that assumes we are touching a heap.
@ -4923,8 +4936,12 @@ heap_xlog_delete(XLogRecPtr lsn, XLogRecord *record)
FreeFakeRelcacheEntry(reln); FreeFakeRelcacheEntry(reln);
} }
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(xlrec->target.node, blkno, false); buffer = XLogReadBuffer(xlrec->target.node, blkno, false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
@ -5004,8 +5021,12 @@ heap_xlog_insert(XLogRecPtr lsn, XLogRecord *record)
FreeFakeRelcacheEntry(reln); FreeFakeRelcacheEntry(reln);
} }
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
if (record->xl_info & XLOG_HEAP_INIT_PAGE) if (record->xl_info & XLOG_HEAP_INIT_PAGE)
{ {
@ -5107,8 +5128,6 @@ heap_xlog_multi_insert(XLogRecPtr lsn, XLogRecord *record)
* required. * required.
*/ */
RestoreBkpBlocks(lsn, record, false);
xlrec = (xl_heap_multi_insert *) recdata; xlrec = (xl_heap_multi_insert *) recdata;
recdata += SizeOfHeapMultiInsert; recdata += SizeOfHeapMultiInsert;
@ -5137,8 +5156,12 @@ heap_xlog_multi_insert(XLogRecPtr lsn, XLogRecord *record)
FreeFakeRelcacheEntry(reln); FreeFakeRelcacheEntry(reln);
} }
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
if (isinit) if (isinit)
{ {
@ -5232,9 +5255,10 @@ static void
heap_xlog_update(XLogRecPtr lsn, XLogRecord *record, bool hot_update) heap_xlog_update(XLogRecPtr lsn, XLogRecord *record, bool hot_update)
{ {
xl_heap_update *xlrec = (xl_heap_update *) XLogRecGetData(record); xl_heap_update *xlrec = (xl_heap_update *) XLogRecGetData(record);
Buffer buffer;
bool samepage = (ItemPointerGetBlockNumber(&(xlrec->newtid)) == bool samepage = (ItemPointerGetBlockNumber(&(xlrec->newtid)) ==
ItemPointerGetBlockNumber(&(xlrec->target.tid))); ItemPointerGetBlockNumber(&(xlrec->target.tid)));
Buffer obuffer,
nbuffer;
Page page; Page page;
OffsetNumber offnum; OffsetNumber offnum;
ItemId lp = NULL; ItemId lp = NULL;
@ -5265,27 +5289,44 @@ heap_xlog_update(XLogRecPtr lsn, XLogRecord *record, bool hot_update)
FreeFakeRelcacheEntry(reln); FreeFakeRelcacheEntry(reln);
} }
if (record->xl_info & XLR_BKP_BLOCK_1) /*
* In normal operation, it is important to lock the two pages in
* page-number order, to avoid possible deadlocks against other update
* operations going the other way. However, during WAL replay there can
* be no other update happening, so we don't need to worry about that. But
* we *do* need to worry that we don't expose an inconsistent state to Hot
* Standby queries --- so the original page can't be unlocked before we've
* added the new tuple to the new page.
*/
if (record->xl_info & XLR_BKP_BLOCK(0))
{ {
obuffer = RestoreBackupBlock(lsn, record, 0, false, true);
if (samepage) if (samepage)
return; /* backup block covered both changes */ {
/* backup block covered both changes, so we're done */
UnlockReleaseBuffer(obuffer);
return;
}
goto newt; goto newt;
} }
/* Deal with old tuple version */ /* Deal with old tuple version */
buffer = XLogReadBuffer(xlrec->target.node, obuffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->target.tid)), ItemPointerGetBlockNumber(&(xlrec->target.tid)),
false); false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(obuffer))
goto newt; goto newt;
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(obuffer);
if (XLByteLE(lsn, PageGetLSN(page))) /* changes are applied */ if (XLByteLE(lsn, PageGetLSN(page))) /* changes are applied */
{ {
UnlockReleaseBuffer(buffer);
if (samepage) if (samepage)
{
UnlockReleaseBuffer(obuffer);
return; return;
}
goto newt; goto newt;
} }
@ -5323,11 +5364,14 @@ heap_xlog_update(XLogRecPtr lsn, XLogRecord *record, bool hot_update)
* is already applied * is already applied
*/ */
if (samepage) if (samepage)
{
nbuffer = obuffer;
goto newsame; goto newsame;
}
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(obuffer);
UnlockReleaseBuffer(buffer);
/* Deal with new tuple */ /* Deal with new tuple */
@ -5349,31 +5393,38 @@ newt:;
FreeFakeRelcacheEntry(reln); FreeFakeRelcacheEntry(reln);
} }
if (record->xl_info & XLR_BKP_BLOCK_2) if (record->xl_info & XLR_BKP_BLOCK(1))
{
(void) RestoreBackupBlock(lsn, record, 1, false, false);
if (BufferIsValid(obuffer))
UnlockReleaseBuffer(obuffer);
return; return;
}
if (record->xl_info & XLOG_HEAP_INIT_PAGE) if (record->xl_info & XLOG_HEAP_INIT_PAGE)
{ {
buffer = XLogReadBuffer(xlrec->target.node, nbuffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->newtid)), ItemPointerGetBlockNumber(&(xlrec->newtid)),
true); true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(nbuffer));
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(nbuffer);
PageInit(page, BufferGetPageSize(buffer), 0); PageInit(page, BufferGetPageSize(nbuffer), 0);
} }
else else
{ {
buffer = XLogReadBuffer(xlrec->target.node, nbuffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->newtid)), ItemPointerGetBlockNumber(&(xlrec->newtid)),
false); false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(nbuffer))
return; return;
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(nbuffer);
if (XLByteLE(lsn, PageGetLSN(page))) /* changes are applied */ if (XLByteLE(lsn, PageGetLSN(page))) /* changes are applied */
{ {
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(nbuffer);
if (BufferIsValid(obuffer))
UnlockReleaseBuffer(obuffer);
return; return;
} }
} }
@ -5418,11 +5469,14 @@ newsame:;
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(nbuffer);
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(nbuffer);
if (BufferIsValid(obuffer) && obuffer != nbuffer)
UnlockReleaseBuffer(obuffer);
/* /*
* If the page is running low on free space, update the FSM as well. * If the new page is running low on free space, update the FSM as well.
* Arbitrarily, our definition of "low" is less than 20%. We can't do much * Arbitrarily, our definition of "low" is less than 20%. We can't do much
* better than that without knowing the fill-factor for the table. * better than that without knowing the fill-factor for the table.
* *
@ -5438,7 +5492,8 @@ newsame:;
*/ */
if (!hot_update && freespace < BLCKSZ / 5) if (!hot_update && freespace < BLCKSZ / 5)
XLogRecordPageWithFreeSpace(xlrec->target.node, XLogRecordPageWithFreeSpace(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->newtid)), freespace); ItemPointerGetBlockNumber(&(xlrec->newtid)),
freespace);
} }
static void static void
@ -5451,8 +5506,12 @@ heap_xlog_lock(XLogRecPtr lsn, XLogRecord *record)
ItemId lp = NULL; ItemId lp = NULL;
HeapTupleHeader htup; HeapTupleHeader htup;
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(xlrec->target.node, buffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->target.tid)), ItemPointerGetBlockNumber(&(xlrec->target.tid)),
@ -5510,8 +5569,12 @@ heap_xlog_inplace(XLogRecPtr lsn, XLogRecord *record)
uint32 oldlen; uint32 oldlen;
uint32 newlen; uint32 newlen;
if (record->xl_info & XLR_BKP_BLOCK_1) /* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return; return;
}
buffer = XLogReadBuffer(xlrec->target.node, buffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->target.tid)), ItemPointerGetBlockNumber(&(xlrec->target.tid)),
@ -5560,8 +5623,6 @@ heap_redo(XLogRecPtr lsn, XLogRecord *record)
* required. The ones in heap2 rmgr do. * required. The ones in heap2 rmgr do.
*/ */
RestoreBkpBlocks(lsn, record, false);
switch (info & XLOG_HEAP_OPMASK) switch (info & XLOG_HEAP_OPMASK)
{ {
case XLOG_HEAP_INSERT: case XLOG_HEAP_INSERT:
@ -5595,11 +5656,6 @@ heap2_redo(XLogRecPtr lsn, XLogRecord *record)
{ {
uint8 info = record->xl_info & ~XLR_INFO_MASK; uint8 info = record->xl_info & ~XLR_INFO_MASK;
/*
* Note that RestoreBkpBlocks() is called after conflict processing within
* each record type handling function.
*/
switch (info & XLOG_HEAP_OPMASK) switch (info & XLOG_HEAP_OPMASK)
{ {
case XLOG_HEAP2_FREEZE: case XLOG_HEAP2_FREEZE:

190
src/backend/access/nbtree/nbtxlog.c
Unescape View File

@ -218,10 +218,9 @@ btree_xlog_insert(bool isleaf, bool ismeta,
datalen -= sizeof(xl_btree_metadata); datalen -= sizeof(xl_btree_metadata);
} }
if ((record->xl_info & XLR_BKP_BLOCK_1) && !ismeta && isleaf) if (record->xl_info & XLR_BKP_BLOCK(0))
return; /* nothing to do */ (void) RestoreBackupBlock(lsn, record, 0, false, false);
else
if (!(record->xl_info & XLR_BKP_BLOCK_1))
{ {
buffer = XLogReadBuffer(xlrec->target.node, buffer = XLogReadBuffer(xlrec->target.node,
ItemPointerGetBlockNumber(&(xlrec->target.tid)), ItemPointerGetBlockNumber(&(xlrec->target.tid)),
@ -249,6 +248,13 @@ btree_xlog_insert(bool isleaf, bool ismeta,
} }
} }
/*
* Note: in normal operation, we'd update the metapage while still holding
* lock on the page we inserted into. But during replay it's not
* necessary to hold that lock, since no other index updates can be
* happening concurrently, and readers will cope fine with following an
* obsolete link from the metapage.
*/
if (ismeta) if (ismeta)
_bt_restore_meta(xlrec->target.node, lsn, _bt_restore_meta(xlrec->target.node, lsn,
md.root, md.level, md.root, md.level,
@ -290,7 +296,7 @@ btree_xlog_split(bool onleft, bool isroot,
forget_matching_split(xlrec->node, downlink, false); forget_matching_split(xlrec->node, downlink, false);
/* Extract left hikey and its size (still assuming 16-bit alignment) */ /* Extract left hikey and its size (still assuming 16-bit alignment) */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (!(record->xl_info & XLR_BKP_BLOCK(0)))
{ {
/* We assume 16-bit alignment is enough for IndexTupleSize */ /* We assume 16-bit alignment is enough for IndexTupleSize */
left_hikey = (Item) datapos; left_hikey = (Item) datapos;
@ -310,7 +316,7 @@ btree_xlog_split(bool onleft, bool isroot,
datalen -= sizeof(OffsetNumber); datalen -= sizeof(OffsetNumber);
} }
if (onleft && !(record->xl_info & XLR_BKP_BLOCK_1)) if (onleft && !(record->xl_info & XLR_BKP_BLOCK(0)))
{ {
/* /*
* We assume that 16-bit alignment is enough to apply IndexTupleSize * We assume that 16-bit alignment is enough to apply IndexTupleSize
@ -323,7 +329,7 @@ btree_xlog_split(bool onleft, bool isroot,
datalen -= newitemsz; datalen -= newitemsz;
} }
/* Reconstruct right (new) sibling from scratch */ /* Reconstruct right (new) sibling page from scratch */
rbuf = XLogReadBuffer(xlrec->node, xlrec->rightsib, true); rbuf = XLogReadBuffer(xlrec->node, xlrec->rightsib, true);
Assert(BufferIsValid(rbuf)); Assert(BufferIsValid(rbuf));
rpage = (Page) BufferGetPage(rbuf); rpage = (Page) BufferGetPage(rbuf);
@ -357,18 +363,21 @@ btree_xlog_split(bool onleft, bool isroot,
/* don't release the buffer yet; we touch right page's first item below */ /* don't release the buffer yet; we touch right page's first item below */
/* /* Now reconstruct left (original) sibling page */
* Reconstruct left (original) sibling if needed. Note that this code if (record->xl_info & XLR_BKP_BLOCK(0))
* ensures that the items remaining on the left page are in the correct (void) RestoreBackupBlock(lsn, record, 0, false, false);
* item number order, but it does not reproduce the physical order they else
* would have had. Is this worth changing? See also _bt_restore_page().
*/
if (!(record->xl_info & XLR_BKP_BLOCK_1))
{ {
Buffer lbuf = XLogReadBuffer(xlrec->node, xlrec->leftsib, false); Buffer lbuf = XLogReadBuffer(xlrec->node, xlrec->leftsib, false);
if (BufferIsValid(lbuf)) if (BufferIsValid(lbuf))
{ {
/*
* Note that this code ensures that the items remaining on the
* left page are in the correct item number order, but it does not
* reproduce the physical order they would have had. Is this
* worth changing? See also _bt_restore_page().
*/
Page lpage = (Page) BufferGetPage(lbuf); Page lpage = (Page) BufferGetPage(lbuf);
BTPageOpaque lopaque = (BTPageOpaque) PageGetSpecialPointer(lpage); BTPageOpaque lopaque = (BTPageOpaque) PageGetSpecialPointer(lpage);
@ -432,8 +441,17 @@ btree_xlog_split(bool onleft, bool isroot,
/* We no longer need the right buffer */ /* We no longer need the right buffer */
UnlockReleaseBuffer(rbuf); UnlockReleaseBuffer(rbuf);
/* Fix left-link of the page to the right of the new right sibling */ /*
if (xlrec->rnext != P_NONE && !(record->xl_info & XLR_BKP_BLOCK_2)) * Fix left-link of the page to the right of the new right sibling.
*
* Note: in normal operation, we do this while still holding lock on the
* two split pages. However, that's not necessary for correctness in WAL
* replay, because no other index update can be in progress, and readers
* will cope properly when following an obsolete left-link.
*/
if (record->xl_info & XLR_BKP_BLOCK(1))
(void) RestoreBackupBlock(lsn, record, 1, false, false);
else if (xlrec->rnext != P_NONE)
{ {
Buffer buffer = XLogReadBuffer(xlrec->node, xlrec->rnext, false); Buffer buffer = XLogReadBuffer(xlrec->node, xlrec->rnext, false);
@ -463,13 +481,11 @@ btree_xlog_split(bool onleft, bool isroot,
static void static void
btree_xlog_vacuum(XLogRecPtr lsn, XLogRecord *record) btree_xlog_vacuum(XLogRecPtr lsn, XLogRecord *record)
{ {
xl_btree_vacuum *xlrec; xl_btree_vacuum *xlrec = (xl_btree_vacuum *) XLogRecGetData(record);
Buffer buffer; Buffer buffer;
Page page; Page page;
BTPageOpaque opaque; BTPageOpaque opaque;
xlrec = (xl_btree_vacuum *) XLogRecGetData(record);
/* /*
* If queries might be active then we need to ensure every block is * If queries might be active then we need to ensure every block is
* unpinned between the lastBlockVacuumed and the current block, if there * unpinned between the lastBlockVacuumed and the current block, if there
@ -502,13 +518,14 @@ btree_xlog_vacuum(XLogRecPtr lsn, XLogRecord *record)
} }
/* /*
* If the block was restored from a full page image, nothing more to do. * If we have a full-page image, restore it (using a cleanup lock) and
* The RestoreBkpBlocks() call already pinned and took cleanup lock on it. * we're done.
* XXX: Perhaps we should call RestoreBkpBlocks() *after* the loop above,
* to make the disk access more sequential.
*/ */
if (record->xl_info & XLR_BKP_BLOCK_1) if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, true, false);
return; return;
}
/* /*
* Like in btvacuumpage(), we need to take a cleanup lock on every leaf * Like in btvacuumpage(), we need to take a cleanup lock on every leaf
@ -563,9 +580,8 @@ btree_xlog_vacuum(XLogRecPtr lsn, XLogRecord *record)
* XXX optimise later with something like XLogPrefetchBuffer() * XXX optimise later with something like XLogPrefetchBuffer()
*/ */
static TransactionId static TransactionId
btree_xlog_delete_get_latestRemovedXid(XLogRecord *record) btree_xlog_delete_get_latestRemovedXid(xl_btree_delete *xlrec)
{ {
xl_btree_delete *xlrec = (xl_btree_delete *) XLogRecGetData(record);
OffsetNumber *unused; OffsetNumber *unused;
Buffer ibuffer, Buffer ibuffer,
hbuffer; hbuffer;
@ -702,15 +718,35 @@ btree_xlog_delete_get_latestRemovedXid(XLogRecord *record)
static void static void
btree_xlog_delete(XLogRecPtr lsn, XLogRecord *record) btree_xlog_delete(XLogRecPtr lsn, XLogRecord *record)
{ {
xl_btree_delete *xlrec; xl_btree_delete *xlrec = (xl_btree_delete *) XLogRecGetData(record);
Buffer buffer; Buffer buffer;
Page page; Page page;
BTPageOpaque opaque; BTPageOpaque opaque;
if (record->xl_info & XLR_BKP_BLOCK_1) /*
return; * If we have any conflict processing to do, it must happen before we
* update the page.
*
* Btree 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_CLEANUP_INFO 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 btree vacuum records on that index.
*/
if (InHotStandby)
{
TransactionId latestRemovedXid = btree_xlog_delete_get_latestRemovedXid(xlrec);
xlrec = (xl_btree_delete *) XLogRecGetData(record); ResolveRecoveryConflictWithSnapshot(latestRemovedXid, xlrec->node);
}
/* If we have a full-page image, restore it and we're done */
if (record->xl_info & XLR_BKP_BLOCK(0))
{
(void) RestoreBackupBlock(lsn, record, 0, false, false);
return;
}
/* /*
* We don't need to take a cleanup lock to apply these changes. See * We don't need to take a cleanup lock to apply these changes. See
@ -766,8 +802,18 @@ btree_xlog_delete_page(uint8 info, XLogRecPtr lsn, XLogRecord *record)
leftsib = xlrec->leftblk; leftsib = xlrec->leftblk;
rightsib = xlrec->rightblk; rightsib = xlrec->rightblk;
/*
* In normal operation, we would lock all the pages this WAL record
* touches before changing any of them. In WAL replay, it should be okay
* to lock just one page at a time, since no concurrent index updates can
* be happening, and readers should not care whether they arrive at the
* target page or not (since it's surely empty).
*/
/* parent page */ /* parent page */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xlrec->target.node, parent, false); buffer = XLogReadBuffer(xlrec->target.node, parent, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -813,7 +859,9 @@ btree_xlog_delete_page(uint8 info, XLogRecPtr lsn, XLogRecord *record)
} }
/* Fix left-link of right sibling */ /* Fix left-link of right sibling */
if (!(record->xl_info & XLR_BKP_BLOCK_2)) if (record->xl_info & XLR_BKP_BLOCK(1))
(void) RestoreBackupBlock(lsn, record, 1, false, false);
else
{ {
buffer = XLogReadBuffer(xlrec->target.node, rightsib, false); buffer = XLogReadBuffer(xlrec->target.node, rightsib, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -837,7 +885,9 @@ btree_xlog_delete_page(uint8 info, XLogRecPtr lsn, XLogRecord *record)
} }
/* Fix right-link of left sibling, if any */ /* Fix right-link of left sibling, if any */
if (!(record->xl_info & XLR_BKP_BLOCK_3)) if (record->xl_info & XLR_BKP_BLOCK(2))
(void) RestoreBackupBlock(lsn, record, 2, false, false);
else
{ {
if (leftsib != P_NONE) if (leftsib != P_NONE)
{ {
@ -911,6 +961,9 @@ btree_xlog_newroot(XLogRecPtr lsn, XLogRecord *record)
BTPageOpaque pageop; BTPageOpaque pageop;
BlockNumber downlink = 0; BlockNumber downlink = 0;
/* Backup blocks are not used in newroot records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
buffer = XLogReadBuffer(xlrec->node, xlrec->rootblk, true); buffer = XLogReadBuffer(xlrec->node, xlrec->rootblk, true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
@ -952,67 +1005,36 @@ btree_xlog_newroot(XLogRecPtr lsn, XLogRecord *record)
forget_matching_split(xlrec->node, downlink, true); forget_matching_split(xlrec->node, downlink, true);
} }
static void
void btree_xlog_reuse_page(XLogRecPtr lsn, XLogRecord *record)
btree_redo(XLogRecPtr lsn, XLogRecord *record)
{ {
uint8 info = record->xl_info & ~XLR_INFO_MASK; xl_btree_reuse_page *xlrec = (xl_btree_reuse_page *) XLogRecGetData(record);
/* /*
* If we have any conflict processing to do, it must happen before we * Btree reuse_page records exist to provide a conflict point when we
* update the page. * reuse pages in the index via the FSM. That's all they do though.
*
* latestRemovedXid was the page's btpo.xact. The btpo.xact <
* RecentGlobalXmin test in _bt_page_recyclable() conceptually mirrors the
* pgxact->xmin > limitXmin test in GetConflictingVirtualXIDs().
* Consequently, one XID value achieves the same exclusion effect on
* master and standby.
*/ */
if (InHotStandby) if (InHotStandby)
{ {
switch (info) ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
{ xlrec->node);
case XLOG_BTREE_DELETE:
/*
* Btree 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_CLEANUP_INFO 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 any we know that no conflicts exist
* from individual btree vacuum records on that index.
*/
{
TransactionId latestRemovedXid = btree_xlog_delete_get_latestRemovedXid(record);
xl_btree_delete *xlrec = (xl_btree_delete *) XLogRecGetData(record);
ResolveRecoveryConflictWithSnapshot(latestRemovedXid, xlrec->node);
} }
break;
case XLOG_BTREE_REUSE_PAGE:
/*
* Btree reuse page records exist to provide a conflict point
* when we reuse pages in the index via the FSM. That's all it
* does though. latestRemovedXid was the page's btpo.xact. The
* btpo.xact < RecentGlobalXmin test in _bt_page_recyclable()
* conceptually mirrors the pgxact->xmin > limitXmin test in
* GetConflictingVirtualXIDs(). Consequently, one XID value
* achieves the same exclusion effect on master and standby.
*/
{
xl_btree_reuse_page *xlrec = (xl_btree_reuse_page *) XLogRecGetData(record);
ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, xlrec->node); /* Backup blocks are not used in reuse_page records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
} }
return;
default:
break;
}
}
/* void
* Vacuum needs to pin and take cleanup lock on every leaf page, a regular btree_redo(XLogRecPtr lsn, XLogRecord *record)
* exclusive lock is enough for all other purposes. {
*/ uint8 info = record->xl_info & ~XLR_INFO_MASK;
RestoreBkpBlocks(lsn, record, (info == XLOG_BTREE_VACUUM));
switch (info) switch (info)
{ {
@ -1052,7 +1074,7 @@ btree_redo(XLogRecPtr lsn, XLogRecord *record)
btree_xlog_newroot(lsn, record); btree_xlog_newroot(lsn, record);
break; break;
case XLOG_BTREE_REUSE_PAGE: case XLOG_BTREE_REUSE_PAGE:
/* Handled above before restoring bkp block */ btree_xlog_reuse_page(lsn, record);
break; break;
default: default:
elog(PANIC, "btree_redo: unknown op code %u", info); elog(PANIC, "btree_redo: unknown op code %u", info);

259
src/backend/access/spgist/spgxlog.c
Unescape View File

@ -76,6 +76,9 @@ spgRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer; Buffer buffer;
Page page; Page page;
/* Backup blocks are not used in create_index records */
Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
buffer = XLogReadBuffer(*node, SPGIST_METAPAGE_BLKNO, true); buffer = XLogReadBuffer(*node, SPGIST_METAPAGE_BLKNO, true);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
@ -117,7 +120,14 @@ spgRedoAddLeaf(XLogRecPtr lsn, XLogRecord *record)
ptr += sizeof(spgxlogAddLeaf); ptr += sizeof(spgxlogAddLeaf);
leafTuple = (SpGistLeafTuple) ptr; leafTuple = (SpGistLeafTuple) ptr;
if (!(record->xl_info & XLR_BKP_BLOCK_1)) /*
* In normal operation we would have both current and parent pages locked
* simultaneously; but in WAL replay it should be safe to update the leaf
* page before updating the parent.
*/
if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoLeaf, buffer = XLogReadBuffer(xldata->node, xldata->blknoLeaf,
xldata->newPage); xldata->newPage);
@ -169,8 +179,9 @@ spgRedoAddLeaf(XLogRecPtr lsn, XLogRecord *record)
} }
/* update parent downlink if necessary */ /* update parent downlink if necessary */
if (xldata->blknoParent != InvalidBlockNumber && if (record->xl_info & XLR_BKP_BLOCK(1))
!(record->xl_info & XLR_BKP_BLOCK_2)) (void) RestoreBackupBlock(lsn, record, 1, false, false);
else if (xldata->blknoParent != InvalidBlockNumber)
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false); buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -219,8 +230,16 @@ spgRedoMoveLeafs(XLogRecPtr lsn, XLogRecord *record)
/* now ptr points to the list of leaf tuples */ /* now ptr points to the list of leaf tuples */
/*
* In normal operation we would have all three pages (source, dest, and
* parent) locked simultaneously; but in WAL replay it should be safe to
* update them one at a time, as long as we do it in the right order.
*/
/* Insert tuples on the dest page (do first, so redirect is valid) */ /* Insert tuples on the dest page (do first, so redirect is valid) */
if (!(record->xl_info & XLR_BKP_BLOCK_2)) if (record->xl_info & XLR_BKP_BLOCK(1))
(void) RestoreBackupBlock(lsn, record, 1, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoDst, buffer = XLogReadBuffer(xldata->node, xldata->blknoDst,
xldata->newPage); xldata->newPage);
@ -253,7 +272,9 @@ spgRedoMoveLeafs(XLogRecPtr lsn, XLogRecord *record)
} }
/* Delete tuples from the source page, inserting a redirection pointer */ /* Delete tuples from the source page, inserting a redirection pointer */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false); buffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -276,7 +297,9 @@ spgRedoMoveLeafs(XLogRecPtr lsn, XLogRecord *record)
} }
/* And update the parent downlink */ /* And update the parent downlink */
if (!(record->xl_info & XLR_BKP_BLOCK_3)) if (record->xl_info & XLR_BKP_BLOCK(2))
(void) RestoreBackupBlock(lsn, record, 2, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false); buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -322,7 +345,9 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
{ {
/* update in place */ /* update in place */
Assert(xldata->blknoParent == InvalidBlockNumber); Assert(xldata->blknoParent == InvalidBlockNumber);
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -347,8 +372,22 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
} }
else else
{ {
/*
* In normal operation we would have all three pages (source, dest,
* and parent) locked simultaneously; but in WAL replay it should be
* safe to update them one at a time, as long as we do it in the right
* order.
*
* The logic here depends on the assumption that blkno != blknoNew,
* else we can't tell which BKP bit goes with which page, and the LSN
* checks could go wrong too.
*/
Assert(xldata->blkno != xldata->blknoNew);
/* Install new tuple first so redirect is valid */ /* Install new tuple first so redirect is valid */
if (!(record->xl_info & XLR_BKP_BLOCK_2)) if (record->xl_info & XLR_BKP_BLOCK(1))
(void) RestoreBackupBlock(lsn, record, 1, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoNew, buffer = XLogReadBuffer(xldata->node, xldata->blknoNew,
xldata->newPage); xldata->newPage);
@ -365,8 +404,17 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
addOrReplaceTuple(page, (Item) innerTuple, addOrReplaceTuple(page, (Item) innerTuple,
innerTuple->size, xldata->offnumNew); innerTuple->size, xldata->offnumNew);
/*
* If parent is in this same page, don't advance LSN;
* doing so would fool us into not applying the parent
* downlink update below. We'll update the LSN when we
* fix the parent downlink.
*/
if (xldata->blknoParent != xldata->blknoNew)
{
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
}
MarkBufferDirty(buffer); MarkBufferDirty(buffer);
} }
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
@ -374,7 +422,9 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
} }
/* Delete old tuple, replacing it with redirect or placeholder tuple */ /* Delete old tuple, replacing it with redirect or placeholder tuple */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -405,8 +455,17 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
else else
SpGistPageGetOpaque(page)->nRedirection++; SpGistPageGetOpaque(page)->nRedirection++;
/*
* If parent is in this same page, don't advance LSN;
* doing so would fool us into not applying the parent
* downlink update below. We'll update the LSN when we
* fix the parent downlink.
*/
if (xldata->blknoParent != xldata->blkno)
{
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
}
MarkBufferDirty(buffer); MarkBufferDirty(buffer);
} }
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
@ -425,7 +484,12 @@ spgRedoAddNode(XLogRecPtr lsn, XLogRecord *record)
else else
bbi = 2; bbi = 2;
if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi))) if (record->xl_info & XLR_BKP_BLOCK(bbi))
{
if (bbi == 2) /* else we already did it */
(void) RestoreBackupBlock(lsn, record, bbi, false, false);
}
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false); buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -467,9 +531,16 @@ spgRedoSplitTuple(XLogRecPtr lsn, XLogRecord *record)
ptr += prefixTuple->size; ptr += prefixTuple->size;
postfixTuple = (SpGistInnerTuple) ptr; postfixTuple = (SpGistInnerTuple) ptr;
/*
* In normal operation we would have both pages locked simultaneously; but
* in WAL replay it should be safe to update them one at a time, as long
* as we do it in the right order.
*/
/* insert postfix tuple first to avoid dangling link */ /* insert postfix tuple first to avoid dangling link */
if (xldata->blknoPostfix != xldata->blknoPrefix && if (record->xl_info & XLR_BKP_BLOCK(1))
!(record->xl_info & XLR_BKP_BLOCK_2)) (void) RestoreBackupBlock(lsn, record, 1, false, false);
else if (xldata->blknoPostfix != xldata->blknoPrefix)
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoPostfix, buffer = XLogReadBuffer(xldata->node, xldata->blknoPostfix,
xldata->newPage); xldata->newPage);
@ -495,7 +566,9 @@ spgRedoSplitTuple(XLogRecPtr lsn, XLogRecord *record)
} }
/* now handle the original page */ /* now handle the original page */
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blknoPrefix, false); buffer = XLogReadBuffer(xldata->node, xldata->blknoPrefix, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -535,6 +608,8 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
uint8 *leafPageSelect; uint8 *leafPageSelect;
Buffer srcBuffer; Buffer srcBuffer;
Buffer destBuffer; Buffer destBuffer;
Page srcPage;
Page destPage;
Page page; Page page;
int bbi; int bbi;
int i; int i;
@ -563,13 +638,14 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
{ {
/* when splitting root, we touch it only in the guise of new inner */ /* when splitting root, we touch it only in the guise of new inner */
srcBuffer = InvalidBuffer; srcBuffer = InvalidBuffer;
srcPage = NULL;
} }
else if (xldata->initSrc) else if (xldata->initSrc)
{ {
/* just re-init the source page */ /* just re-init the source page */
srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, true); srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, true);
Assert(BufferIsValid(srcBuffer)); Assert(BufferIsValid(srcBuffer));
page = (Page) BufferGetPage(srcBuffer); srcPage = (Page) BufferGetPage(srcBuffer);
SpGistInitBuffer(srcBuffer, SpGistInitBuffer(srcBuffer,
SPGIST_LEAF | (xldata->storesNulls ? SPGIST_NULLS : 0)); SPGIST_LEAF | (xldata->storesNulls ? SPGIST_NULLS : 0));
@ -577,14 +653,24 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
} }
else else
{ {
/* delete the specified tuples from source page */ /*
if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi))) * Delete the specified tuples from source page. (In case we're in
* Hot Standby, we need to hold lock on the page till we're done
* inserting leaf tuples and the new inner tuple, else the added
* redirect tuple will be a dangling link.)
*/
if (record->xl_info & XLR_BKP_BLOCK(bbi))
{
srcBuffer = RestoreBackupBlock(lsn, record, bbi, false, true);
srcPage = NULL; /* don't need to do any page updates */
}
else
{ {
srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false); srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false);
if (BufferIsValid(srcBuffer)) if (BufferIsValid(srcBuffer))
{ {
page = BufferGetPage(srcBuffer); srcPage = BufferGetPage(srcBuffer);
if (!XLByteLE(lsn, PageGetLSN(page))) if (!XLByteLE(lsn, PageGetLSN(srcPage)))
{ {
/* /*
* We have it a bit easier here than in doPickSplit(), * We have it a bit easier here than in doPickSplit(),
@ -592,14 +678,14 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
* we can inject the correct redirection tuple now. * we can inject the correct redirection tuple now.
*/ */
if (!state.isBuild) if (!state.isBuild)
spgPageIndexMultiDelete(&state, page, spgPageIndexMultiDelete(&state, srcPage,
toDelete, xldata->nDelete, toDelete, xldata->nDelete,
SPGIST_REDIRECT, SPGIST_REDIRECT,
SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,
xldata->blknoInner, xldata->blknoInner,
xldata->offnumInner); xldata->offnumInner);
else else
spgPageIndexMultiDelete(&state, page, spgPageIndexMultiDelete(&state, srcPage,
toDelete, xldata->nDelete, toDelete, xldata->nDelete,
SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,
SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,
@ -608,10 +694,12 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
/* don't update LSN etc till we're done with it */ /* don't update LSN etc till we're done with it */
} }
} else
srcPage = NULL; /* don't do any page updates */
} }
else else
srcBuffer = InvalidBuffer; srcPage = NULL;
}
bbi++; bbi++;
} }
@ -619,13 +707,14 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
if (xldata->blknoDest == InvalidBlockNumber) if (xldata->blknoDest == InvalidBlockNumber)
{ {
destBuffer = InvalidBuffer; destBuffer = InvalidBuffer;
destPage = NULL;
} }
else if (xldata->initDest) else if (xldata->initDest)
{ {
/* just re-init the dest page */ /* just re-init the dest page */
destBuffer = XLogReadBuffer(xldata->node, xldata->blknoDest, true); destBuffer = XLogReadBuffer(xldata->node, xldata->blknoDest, true);
Assert(BufferIsValid(destBuffer)); Assert(BufferIsValid(destBuffer));
page = (Page) BufferGetPage(destBuffer); destPage = (Page) BufferGetPage(destBuffer);
SpGistInitBuffer(destBuffer, SpGistInitBuffer(destBuffer,
SPGIST_LEAF | (xldata->storesNulls ? SPGIST_NULLS : 0)); SPGIST_LEAF | (xldata->storesNulls ? SPGIST_NULLS : 0));
@ -633,10 +722,27 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
} }
else else
{ {
if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi))) /*
* We could probably release the page lock immediately in the
* full-page-image case, but for safety let's hold it till later.
*/
if (record->xl_info & XLR_BKP_BLOCK(bbi))
{
destBuffer = RestoreBackupBlock(lsn, record, bbi, false, true);
destPage = NULL; /* don't need to do any page updates */
}
else
{
destBuffer = XLogReadBuffer(xldata->node, xldata->blknoDest, false); destBuffer = XLogReadBuffer(xldata->node, xldata->blknoDest, false);
if (BufferIsValid(destBuffer))
{
destPage = (Page) BufferGetPage(destBuffer);
if (XLByteLE(lsn, PageGetLSN(destPage)))
destPage = NULL; /* don't do any page updates */
}
else else
destBuffer = InvalidBuffer; destPage = NULL;
}
bbi++; bbi++;
} }
@ -644,47 +750,34 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
for (i = 0; i < xldata->nInsert; i++) for (i = 0; i < xldata->nInsert; i++)
{ {
SpGistLeafTuple lt = (SpGistLeafTuple) ptr; SpGistLeafTuple lt = (SpGistLeafTuple) ptr;
Buffer leafBuffer;
ptr += lt->size; ptr += lt->size;
leafBuffer = leafPageSelect[i] ? destBuffer : srcBuffer; page = leafPageSelect[i] ? destPage : srcPage;
if (!BufferIsValid(leafBuffer)) if (page == NULL)
continue; /* no need to touch this page */ continue; /* no need to touch this page */
page = BufferGetPage(leafBuffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
addOrReplaceTuple(page, (Item) lt, lt->size, toInsert[i]); addOrReplaceTuple(page, (Item) lt, lt->size, toInsert[i]);
} }
}
/* Now update src and dest page LSNs */ /* Now update src and dest page LSNs if needed */
if (BufferIsValid(srcBuffer)) if (srcPage != NULL)
{
page = BufferGetPage(srcBuffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{ {
PageSetLSN(page, lsn); PageSetLSN(srcPage, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(srcPage, ThisTimeLineID);
MarkBufferDirty(srcBuffer); MarkBufferDirty(srcBuffer);
} }
UnlockReleaseBuffer(srcBuffer); if (destPage != NULL)
}
if (BufferIsValid(destBuffer))
{ {
page = BufferGetPage(destBuffer); PageSetLSN(destPage, lsn);
if (!XLByteLE(lsn, PageGetLSN(page))) PageSetTLI(destPage, ThisTimeLineID);
{
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(destBuffer); MarkBufferDirty(destBuffer);
} }
UnlockReleaseBuffer(destBuffer);
}
/* restore new inner tuple */ /* restore new inner tuple */
if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi))) if (record->xl_info & XLR_BKP_BLOCK(bbi))
(void) RestoreBackupBlock(lsn, record, bbi, false, false);
else
{ {
Buffer buffer = XLogReadBuffer(xldata->node, xldata->blknoInner, Buffer buffer = XLogReadBuffer(xldata->node, xldata->blknoInner,
xldata->initInner); xldata->initInner);
@ -722,6 +815,15 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
} }
bbi++; bbi++;
/*
* Now we can release the leaf-page locks. It's okay to do this before
* updating the parent downlink.
*/
if (BufferIsValid(srcBuffer))
UnlockReleaseBuffer(srcBuffer);
if (BufferIsValid(destBuffer))
UnlockReleaseBuffer(destBuffer);
/* update parent downlink, unless we did it above */ /* update parent downlink, unless we did it above */
if (xldata->blknoParent == InvalidBlockNumber) if (xldata->blknoParent == InvalidBlockNumber)
{ {
@ -730,7 +832,9 @@ spgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record)
} }
else if (xldata->blknoInner != xldata->blknoParent) else if (xldata->blknoInner != xldata->blknoParent)
{ {
if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi))) if (record->xl_info & XLR_BKP_BLOCK(bbi))
(void) RestoreBackupBlock(lsn, record, bbi, false, false);
else
{ {
Buffer buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false); Buffer buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false);
@ -788,7 +892,9 @@ spgRedoVacuumLeaf(XLogRecPtr lsn, XLogRecord *record)
ptr += sizeof(OffsetNumber) * xldata->nChain; ptr += sizeof(OffsetNumber) * xldata->nChain;
chainDest = (OffsetNumber *) ptr; chainDest = (OffsetNumber *) ptr;
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -857,7 +963,9 @@ spgRedoVacuumRoot(XLogRecPtr lsn, XLogRecord *record)
ptr += sizeof(spgxlogVacuumRoot); ptr += sizeof(spgxlogVacuumRoot);
toDelete = (OffsetNumber *) ptr; toDelete = (OffsetNumber *) ptr;
if (!(record->xl_info & XLR_BKP_BLOCK_1)) if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
@ -889,7 +997,20 @@ spgRedoVacuumRedirect(XLogRecPtr lsn, XLogRecord *record)
ptr += sizeof(spgxlogVacuumRedirect); ptr += sizeof(spgxlogVacuumRedirect);
itemToPlaceholder = (OffsetNumber *) ptr; itemToPlaceholder = (OffsetNumber *) ptr;
if (!(record->xl_info & XLR_BKP_BLOCK_1)) /*
* If any redirection tuples are being removed, make sure there are no
* live Hot Standby transactions that might need to see them.
*/
if (InHotStandby)
{
if (TransactionIdIsValid(xldata->newestRedirectXid))
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
xldata->node);
}
if (record->xl_info & XLR_BKP_BLOCK(0))
(void) RestoreBackupBlock(lsn, record, 0, false, false);
else
{ {
buffer = XLogReadBuffer(xldata->node, xldata->blkno, false); buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);
@ -954,36 +1075,6 @@ spg_redo(XLogRecPtr lsn, XLogRecord *record)
uint8 info = record->xl_info & ~XLR_INFO_MASK; uint8 info = record->xl_info & ~XLR_INFO_MASK;
MemoryContext oldCxt; MemoryContext oldCxt;
/*
* If we have any conflict processing to do, it must happen before we
* update the page.
*/
if (InHotStandby)
{
switch (info)
{
case XLOG_SPGIST_VACUUM_REDIRECT:
{
spgxlogVacuumRedirect *xldata =
(spgxlogVacuumRedirect *) XLogRecGetData(record);
/*
* If any redirection tuples are being removed, make sure
* there are no live Hot Standby transactions that might
* need to see them.
*/
if (TransactionIdIsValid(xldata->newestRedirectXid))
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
xldata->node);
break;
}
default:
break;
}
}
RestoreBkpBlocks(lsn, record, false);
oldCxt = MemoryContextSwitchTo(opCtx); oldCxt = MemoryContextSwitchTo(opCtx);
switch (info) switch (info)
{ {

59
src/backend/access/transam/README
Unescape View File

@ -438,8 +438,9 @@ critical section.)
4. Mark the shared buffer(s) as dirty with MarkBufferDirty(). (This must 4. Mark the shared buffer(s) as dirty with MarkBufferDirty(). (This must
happen before the WAL record is inserted; see notes in SyncOneBuffer().) happen before the WAL record is inserted; see notes in SyncOneBuffer().)
5. Build a WAL log record and pass it to XLogInsert(); then update the page's 5. If the relation requires WAL-logging, build a WAL log record and pass it
LSN and TLI using the returned XLOG location. For instance, to XLogInsert(); then update the page's LSN and TLI using the returned XLOG
location. For instance,
recptr = XLogInsert(rmgr_id, info, rdata); recptr = XLogInsert(rmgr_id, info, rdata);
@ -466,9 +467,9 @@ which buffers were handled that way --- otherwise they may be misled about
what the XLOG record actually contains. XLOG records that describe multi-page what the XLOG record actually contains. XLOG records that describe multi-page
changes therefore require some care to design: you must be certain that you changes therefore require some care to design: you must be certain that you
know what data is indicated by each "BKP" bit. An example of the trickiness know what data is indicated by each "BKP" bit. An example of the trickiness
is that in a HEAP_UPDATE record, BKP(1) normally is associated with the source is that in a HEAP_UPDATE record, BKP(0) normally is associated with the source
page and BKP(2) is associated with the destination page --- but if these are page and BKP(1) is associated with the destination page --- but if these are
the same page, only BKP(1) would have been set. the same page, only BKP(0) would have been set.
For this reason as well as the risk of deadlocking on buffer locks, it's best For this reason as well as the risk of deadlocking on buffer locks, it's best
to design WAL records so that they reflect small atomic actions involving just to design WAL records so that they reflect small atomic actions involving just
@ -497,12 +498,19 @@ incrementally update the page, the rdata array *must* mention the buffer
ID at least once; otherwise there is no defense against torn-page problems. ID at least once; otherwise there is no defense against torn-page problems.
The standard replay-routine pattern for this case is The standard replay-routine pattern for this case is
if (record->xl_info & XLR_BKP_BLOCK_n) if (record->xl_info & XLR_BKP_BLOCK(N))
<< do nothing, page was rewritten from logged copy >>; {
/* apply the change from the full-page image */
(void) RestoreBackupBlock(lsn, record, N, false, false);
return;
}
buffer = XLogReadBuffer(rnode, blkno, false); buffer = XLogReadBuffer(rnode, blkno, false);
if (!BufferIsValid(buffer)) if (!BufferIsValid(buffer))
<< do nothing, page has been deleted >>; {
/* page has been deleted, so we need do nothing */
return;
}
page = (Page) BufferGetPage(buffer); page = (Page) BufferGetPage(buffer);
if (XLByteLE(lsn, PageGetLSN(page))) if (XLByteLE(lsn, PageGetLSN(page)))
@ -520,13 +528,42 @@ The standard replay-routine pattern for this case is
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
As noted above, for a multi-page update you need to be able to determine As noted above, for a multi-page update you need to be able to determine
which XLR_BKP_BLOCK_n flag applies to each page. If a WAL record reflects which XLR_BKP_BLOCK(N) flag applies to each page. If a WAL record reflects
a combination of fully-rewritable and incremental updates, then the rewritable a combination of fully-rewritable and incremental updates, then the rewritable
pages don't count for the XLR_BKP_BLOCK_n numbering. (XLR_BKP_BLOCK_n is pages don't count for the XLR_BKP_BLOCK(N) numbering. (XLR_BKP_BLOCK(N) is
associated with the n'th distinct buffer ID seen in the "rdata" array, and associated with the N'th distinct buffer ID seen in the "rdata" array, and
per the above discussion, fully-rewritable buffers shouldn't be mentioned in per the above discussion, fully-rewritable buffers shouldn't be mentioned in
"rdata".) "rdata".)
When replaying a WAL record that describes changes on multiple pages, you
must be careful to lock the pages properly to prevent concurrent Hot Standby
queries from seeing an inconsistent state. If this requires that two
or more buffer locks be held concurrently, the coding pattern shown above
is too simplistic, since it assumes the routine can exit as soon as it's
known the current page requires no modification. Instead, you might have
something like
if (record->xl_info & XLR_BKP_BLOCK(0))
{
/* apply the change from the full-page image */
buffer0 = RestoreBackupBlock(lsn, record, 0, false, true);
}
else
{
buffer0 = XLogReadBuffer(rnode, blkno, false);
if (BufferIsValid(buffer0))
{
... apply the change if not already done ...
MarkBufferDirty(buffer0);
}
}
... similarly apply the changes for remaining pages ...
/* and now we can release the lock on the first page */
if (BufferIsValid(buffer0))
UnlockReleaseBuffer(buffer0);
Due to all these constraints, complex changes (such as a multilevel index Due to all these constraints, complex changes (such as a multilevel index
insertion) normally need to be described by a series of atomic-action WAL insertion) normally need to be described by a series of atomic-action WAL
records. What do you do if the intermediate states are not self-consistent? records. What do you do if the intermediate states are not self-consistent?

67
src/backend/access/transam/xlog.c
Unescape View File

@ -835,8 +835,8 @@ begin:;
* At the exit of this loop, write_len includes the backup block data. * At the exit of this loop, write_len includes the backup block data.
* *
* Also set the appropriate info bits to show which buffers were backed * Also set the appropriate info bits to show which buffers were backed
* up. The i'th XLR_SET_BKP_BLOCK bit corresponds to the i'th distinct * up. The XLR_BKP_BLOCK(N) bit corresponds to the N'th distinct buffer
* buffer value (ignoring InvalidBuffer) appearing in the rdata chain. * value (ignoring InvalidBuffer) appearing in the rdata chain.
*/ */
rdt_lastnormal = rdt; rdt_lastnormal = rdt;
write_len = len; write_len = len;
@ -848,7 +848,7 @@ begin:;
if (!dtbuf_bkp[i]) if (!dtbuf_bkp[i])
continue; continue;
info |= XLR_SET_BKP_BLOCK(i); info |= XLR_BKP_BLOCK(i);
bkpb = &(dtbuf_xlg[i]); bkpb = &(dtbuf_xlg[i]);
page = (char *) BufferGetBlock(dtbuf[i]); page = (char *) BufferGetBlock(dtbuf[i]);
@ -3080,9 +3080,16 @@ CleanupBackupHistory(void)
} }
/* /*
* Restore the backup blocks present in an XLOG record, if any. * Restore a full-page image from a backup block attached to an XLOG record.
* *
* We assume all of the record has been read into memory at *record. * lsn: LSN of the XLOG record being replayed
* record: the complete XLOG record
* block_index: which backup block to restore (0 .. XLR_MAX_BKP_BLOCKS - 1)
* get_cleanup_lock: TRUE to get a cleanup rather than plain exclusive lock
* keep_buffer: TRUE to return the buffer still locked and pinned
*
* Returns the buffer number containing the page. Note this is not terribly
* useful unless keep_buffer is specified as TRUE.
* *
* Note: when a backup block is available in XLOG, we restore it * Note: when a backup block is available in XLOG, we restore it
* unconditionally, even if the page in the database appears newer. * unconditionally, even if the page in the database appears newer.
@ -3093,15 +3100,20 @@ CleanupBackupHistory(void)
* modifications of the page that appear in XLOG, rather than possibly * modifications of the page that appear in XLOG, rather than possibly
* ignoring them as already applied, but that's not a huge drawback. * ignoring them as already applied, but that's not a huge drawback.
* *
* If 'cleanup' is true, a cleanup lock is used when restoring blocks. * If 'get_cleanup_lock' is true, a cleanup lock is obtained on the buffer,
* Otherwise, a normal exclusive lock is used. During crash recovery, that's * else a normal exclusive lock is used. During crash recovery, that's just
* just pro forma because there can't be any regular backends in the system, * pro forma because there can't be any regular backends in the system, but
* but in hot standby mode the distinction is important. The 'cleanup' * in hot standby mode the distinction is important.
* argument applies to all backup blocks in the WAL record, that suffices for *
* now. * If 'keep_buffer' is true, return without releasing the buffer lock and pin;
* then caller is responsible for doing UnlockReleaseBuffer() later. This
* is needed in some cases when replaying XLOG records that touch multiple
* pages, to prevent inconsistent states from being visible to other backends.
* (Again, that's only important in hot standby mode.)
*/ */
void Buffer
RestoreBkpBlocks(XLogRecPtr lsn, XLogRecord *record, bool cleanup) RestoreBackupBlock(XLogRecPtr lsn, XLogRecord *record, int block_index,
bool get_cleanup_lock, bool keep_buffer)
{ {
Buffer buffer; Buffer buffer;
Page page; Page page;
@ -3109,22 +3121,23 @@ RestoreBkpBlocks(XLogRecPtr lsn, XLogRecord *record, bool cleanup)
char *blk; char *blk;
int i; int i;
if (!(record->xl_info & XLR_BKP_BLOCK_MASK)) /* Locate requested BkpBlock in the record */
return;
blk = (char *) XLogRecGetData(record) + record->xl_len; blk = (char *) XLogRecGetData(record) + record->xl_len;
for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++) for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
{ {
if (!(record->xl_info & XLR_SET_BKP_BLOCK(i))) if (!(record->xl_info & XLR_BKP_BLOCK(i)))
continue; continue;
memcpy(&bkpb, blk, sizeof(BkpBlock)); memcpy(&bkpb, blk, sizeof(BkpBlock));
blk += sizeof(BkpBlock); blk += sizeof(BkpBlock);
if (i == block_index)
{
/* Found it, apply the update */
buffer = XLogReadBufferExtended(bkpb.node, bkpb.fork, bkpb.block, buffer = XLogReadBufferExtended(bkpb.node, bkpb.fork, bkpb.block,
RBM_ZERO); RBM_ZERO);
Assert(BufferIsValid(buffer)); Assert(BufferIsValid(buffer));
if (cleanup) if (get_cleanup_lock)
LockBufferForCleanup(buffer); LockBufferForCleanup(buffer);
else else
LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE); LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
@ -3148,10 +3161,19 @@ RestoreBkpBlocks(XLogRecPtr lsn, XLogRecord *record, bool cleanup)
PageSetLSN(page, lsn); PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID); PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer); MarkBufferDirty(buffer);
if (!keep_buffer)
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);
return buffer;
}
blk += BLCKSZ - bkpb.hole_length; blk += BLCKSZ - bkpb.hole_length;
} }
/* Caller specified a bogus block_index */
elog(ERROR, "failed to restore block_index %d", block_index);
return InvalidBuffer; /* keep compiler quiet */
} }
/* /*
@ -3193,7 +3215,7 @@ RecordIsValid(XLogRecord *record, XLogRecPtr recptr, int emode)
{ {
uint32 blen; uint32 blen;
if (!(record->xl_info & XLR_SET_BKP_BLOCK(i))) if (!(record->xl_info & XLR_BKP_BLOCK(i)))
continue; continue;
if (remaining < sizeof(BkpBlock)) if (remaining < sizeof(BkpBlock))
@ -8081,7 +8103,8 @@ xlog_outrec(StringInfo buf, XLogRecord *record)
int i; int i;
appendStringInfo(buf, "prev %X/%X; xid %u", appendStringInfo(buf, "prev %X/%X; xid %u",
(uint32) (record->xl_prev >> 32), (uint32) record->xl_prev, (uint32) (record->xl_prev >> 32),
(uint32) record->xl_prev,
record->xl_xid); record->xl_xid);
appendStringInfo(buf, "; len %u", appendStringInfo(buf, "; len %u",
@ -8089,8 +8112,8 @@ xlog_outrec(StringInfo buf, XLogRecord *record)
for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++) for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
{ {
if (record->xl_info & XLR_SET_BKP_BLOCK(i)) if (record->xl_info & XLR_BKP_BLOCK(i))
appendStringInfo(buf, "; bkpb%d", i + 1); appendStringInfo(buf, "; bkpb%d", i);
} }
appendStringInfo(buf, ": %s", RmgrTable[record->xl_rmid].rm_name); appendStringInfo(buf, ": %s", RmgrTable[record->xl_rmid].rm_name);

8
src/include/access/gist_private.h
Unescape View File

@ -167,7 +167,7 @@ typedef GISTScanOpaqueData *GISTScanOpaque;
#define XLOG_GIST_PAGE_SPLIT 0x30 #define XLOG_GIST_PAGE_SPLIT 0x30
/* #define XLOG_GIST_INSERT_COMPLETE 0x40 */ /* not used anymore */ /* #define XLOG_GIST_INSERT_COMPLETE 0x40 */ /* not used anymore */
#define XLOG_GIST_CREATE_INDEX 0x50 #define XLOG_GIST_CREATE_INDEX 0x50
#define XLOG_GIST_PAGE_DELETE 0x60 /* #define XLOG_GIST_PAGE_DELETE 0x60 */ /* not used anymore */
typedef struct gistxlogPageUpdate typedef struct gistxlogPageUpdate
{ {
@ -211,12 +211,6 @@ typedef struct gistxlogPage
int num; /* number of index tuples following */ int num; /* number of index tuples following */
} gistxlogPage; } gistxlogPage;
typedef struct gistxlogPageDelete
{
RelFileNode node;
BlockNumber blkno;
} gistxlogPageDelete;
/* SplitedPageLayout - gistSplit function result */ /* SplitedPageLayout - gistSplit function result */
typedef struct SplitedPageLayout typedef struct SplitedPageLayout
{ {

16
src/include/access/xlog.h
Unescape View File

@ -71,11 +71,7 @@ typedef struct XLogRecord
*/ */
#define XLR_BKP_BLOCK_MASK 0x0F /* all info bits used for bkp blocks */ #define XLR_BKP_BLOCK_MASK 0x0F /* all info bits used for bkp blocks */
#define XLR_MAX_BKP_BLOCKS 4 #define XLR_MAX_BKP_BLOCKS 4
#define XLR_SET_BKP_BLOCK(iblk) (0x08 >> (iblk)) #define XLR_BKP_BLOCK(iblk) (0x08 >> (iblk)) /* iblk in 0..3 */
#define XLR_BKP_BLOCK_1 XLR_SET_BKP_BLOCK(0) /* 0x08 */
#define XLR_BKP_BLOCK_2 XLR_SET_BKP_BLOCK(1) /* 0x04 */
#define XLR_BKP_BLOCK_3 XLR_SET_BKP_BLOCK(2) /* 0x02 */
#define XLR_BKP_BLOCK_4 XLR_SET_BKP_BLOCK(3) /* 0x01 */
/* Sync methods */ /* Sync methods */
#define SYNC_METHOD_FSYNC 0 #define SYNC_METHOD_FSYNC 0
@ -94,13 +90,13 @@ extern int sync_method;
* If buffer is valid then XLOG will check if buffer must be backed up * If buffer is valid then XLOG will check if buffer must be backed up
* (ie, whether this is first change of that page since last checkpoint). * (ie, whether this is first change of that page since last checkpoint).
* If so, the whole page contents are attached to the XLOG record, and XLOG * If so, the whole page contents are attached to the XLOG record, and XLOG
* sets XLR_BKP_BLOCK_X bit in xl_info. Note that the buffer must be pinned * sets XLR_BKP_BLOCK(N) bit in xl_info. Note that the buffer must be pinned
* and exclusive-locked by the caller, so that it won't change under us. * and exclusive-locked by the caller, so that it won't change under us.
* NB: when the buffer is backed up, we DO NOT insert the data pointed to by * NB: when the buffer is backed up, we DO NOT insert the data pointed to by
* this XLogRecData struct into the XLOG record, since we assume it's present * this XLogRecData struct into the XLOG record, since we assume it's present
* in the buffer. Therefore, rmgr redo routines MUST pay attention to * in the buffer. Therefore, rmgr redo routines MUST pay attention to
* XLR_BKP_BLOCK_X to know what is actually stored in the XLOG record. * XLR_BKP_BLOCK(N) to know what is actually stored in the XLOG record.
* The i'th XLR_BKP_BLOCK bit corresponds to the i'th distinct buffer * The N'th XLR_BKP_BLOCK bit corresponds to the N'th distinct buffer
* value (ignoring InvalidBuffer) appearing in the rdata chain. * value (ignoring InvalidBuffer) appearing in the rdata chain.
* *
* When buffer is valid, caller must set buffer_std to indicate whether the * When buffer is valid, caller must set buffer_std to indicate whether the
@ -274,7 +270,9 @@ extern int XLogFileOpen(XLogSegNo segno);
extern void XLogGetLastRemoved(XLogSegNo *segno); extern void XLogGetLastRemoved(XLogSegNo *segno);
extern void XLogSetAsyncXactLSN(XLogRecPtr record); extern void XLogSetAsyncXactLSN(XLogRecPtr record);
extern void RestoreBkpBlocks(XLogRecPtr lsn, XLogRecord *record, bool cleanup); extern Buffer RestoreBackupBlock(XLogRecPtr lsn, XLogRecord *record,
int block_index,
bool get_cleanup_lock, bool keep_buffer);
extern void xlog_redo(XLogRecPtr lsn, XLogRecord *record); extern void xlog_redo(XLogRecPtr lsn, XLogRecord *record);
extern void xlog_desc(StringInfo buf, uint8 xl_info, char *rec); extern void xlog_desc(StringInfo buf, uint8 xl_info, char *rec);

Write Preview
Loading…
Cancel
Save