Skip to content

P
Postgres FD Implementation
Commit 4016bdef authored by Tom Lane's avatar Tom Lane
Browse files
Options

Fix assorted bugs in GIN's WAL replay logic. 

The original coding was quite sloppy about handling the case where
XLogReadBuffer fails (because the page has since been deleted).  This
would result in either "bad buffer id: 0" or an Assert failure during
replay, if indeed the page were no longer there.  In a couple of places
it also neglected to check whether the change had already been applied,
which would probably result in corrupted index contents.  I believe that
bug #5703 is an instance of the first problem.  These issues could show up
without replication, but only if you were unfortunate enough to crash
between modification of a GIN index and the next checkpoint.

Back-patch to 8.2, which is as far back as GIN has WAL support.
parent 220e45bf
Show whitespace changes
Inline Side-by-side
Showing with 181 additions and 127 deletions
src/backend/access/gin/ginxlog.c
View file @ 4016bdef
......@@ -77,11 +77,13 @@ ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
MetaBuffer = XLogReadBuffer(*node, GIN_METAPAGE_BLKNO, true);
Assert(BufferIsValid(MetaBuffer));
page = (Page) BufferGetPage(MetaBuffer);
GinInitMetabuffer(MetaBuffer);
page = (Page) BufferGetPage(MetaBuffer);
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(MetaBuffer);
RootBuffer = XLogReadBuffer(*node, GIN_ROOT_BLKNO, true);
Assert(BufferIsValid(RootBuffer));
......@@ -91,11 +93,10 @@ ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(MetaBuffer);
UnlockReleaseBuffer(MetaBuffer);
MarkBufferDirty(RootBuffer);
UnlockReleaseBuffer(RootBuffer);
UnlockReleaseBuffer(MetaBuffer);
}
static void
......@@ -128,21 +129,49 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer;
Page page;
/* first, forget any incomplete split this insertion completes */
if (data->isData)
{
Assert(data->isDelete == FALSE);
if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
{
PostingItem *pitem;
pitem = (PostingItem *) (XLogRecGetData(record) + sizeof(ginxlogInsert));
forgetIncompleteSplit(data->node,
PostingItemGetBlockNumber(pitem),
data->updateBlkno);
}
}
else
{
if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
{
IndexTuple itup;
itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsert));
forgetIncompleteSplit(data->node,
GinItemPointerGetBlockNumber(&itup->t_tid),
data->updateBlkno);
}
}
/* nothing else to do if page was backed up */
if (record->xl_info & XLR_BKP_BLOCK_1)
return;
buffer = XLogReadBuffer(data->node, data->blkno, false);
Assert(BufferIsValid(buffer));
if (!BufferIsValid(buffer))
return; /* page was deleted, nothing to do */
page = (Page) BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
if (data->isData)
{
Assert(data->isDelete == FALSE);
Assert(GinPageIsData(page));
if (!XLByteLE(lsn, PageGetLSN(page)))
{
if (data->isLeaf)
{
OffsetNumber i;
......@@ -172,23 +201,12 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
GinDataPageAddItem(page, pitem, data->offset);
}
}
if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
{
PostingItem *pitem = (PostingItem *) (XLogRecGetData(record) + sizeof(ginxlogInsert));
forgetIncompleteSplit(data->node, PostingItemGetBlockNumber(pitem), data->updateBlkno);
}
}
else
{
IndexTuple itup;
Assert(!GinPageIsData(page));
if (!XLByteLE(lsn, PageGetLSN(page)))
{
if (data->updateBlkno != InvalidBlockNumber)
{
/* update link to right page after split */
......@@ -212,20 +230,12 @@ ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
data->node.spcNode, data->node.dbNode, data->node.relNode);
}
if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
{
itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsert));
forgetIncompleteSplit(data->node, GinItemPointerGetBlockNumber(&itup->t_tid), data->updateBlkno);
}
}
if (!XLByteLE(lsn, PageGetLSN(page)))
{
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
}
UnlockReleaseBuffer(buffer);
}
......@@ -244,7 +254,7 @@ ginRedoSplit(XLogRecPtr lsn, XLogRecord *record)
if (data->isData)
flags |= GIN_DATA;
lbuffer = XLogReadBuffer(data->node, data->lblkno, data->isRootSplit);
lbuffer = XLogReadBuffer(data->node, data->lblkno, true);
Assert(BufferIsValid(lbuffer));
lpage = (Page) BufferGetPage(lbuffer);
GinInitBuffer(lbuffer, flags);
......@@ -321,7 +331,7 @@ ginRedoSplit(XLogRecPtr lsn, XLogRecord *record)
if (data->isRootSplit)
{
Buffer rootBuf = XLogReadBuffer(data->node, data->rootBlkno, false);
Buffer rootBuf = XLogReadBuffer(data->node, data->rootBlkno, true);
Page rootPage = BufferGetPage(rootBuf);
GinInitBuffer(rootBuf, flags & ~GIN_LEAF);
......@@ -357,14 +367,17 @@ ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record)
Buffer buffer;
Page page;
/* nothing else to do if page was backed up (and no info to do it with) */
/* nothing 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(data->node, data->blkno, false);
Assert(BufferIsValid(buffer));
if (!BufferIsValid(buffer))
return;
page = (Page) BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
if (GinPageIsData(page))
{
memcpy(GinDataPageGetData(page), XLogRecGetData(record) + sizeof(ginxlogVacuumPage),
......@@ -394,8 +407,9 @@ ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record)
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
}
UnlockReleaseBuffer(buffer);
}
......@@ -409,39 +423,57 @@ ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record)
if (!(record->xl_info & XLR_BKP_BLOCK_1))
{
buffer = XLogReadBuffer(data->node, data->blkno, false);
if (BufferIsValid(buffer))
{
page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
Assert(GinPageIsData(page));
GinPageGetOpaque(page)->flags = GIN_DELETED;
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
}
UnlockReleaseBuffer(buffer);
}
}
if (!(record->xl_info & XLR_BKP_BLOCK_2))
{
buffer = XLogReadBuffer(data->node, data->parentBlkno, false);
if (BufferIsValid(buffer))
{
page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
Assert(GinPageIsData(page));
Assert(!GinPageIsLeaf(page));
PageDeletePostingItem(page, data->parentOffset);
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
}
UnlockReleaseBuffer(buffer);
}
}
if (!(record->xl_info & XLR_BKP_BLOCK_3) && data->leftBlkno != InvalidBlockNumber)
{
buffer = XLogReadBuffer(data->node, data->leftBlkno, false);
if (BufferIsValid(buffer))
{
page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
{
Assert(GinPageIsData(page));
GinPageGetOpaque(page)->rightlink = data->rightLink;
PageSetLSN(page, lsn);
PageSetTLI(page, ThisTimeLineID);
MarkBufferDirty(buffer);
}
UnlockReleaseBuffer(buffer);
}
}
}
static void
......@@ -450,8 +482,11 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
ginxlogUpdateMeta *data = (ginxlogUpdateMeta *) XLogRecGetData(record);
Buffer metabuffer;
Page metapage;
Buffer buffer;
metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
if (!BufferIsValid(metabuffer))
elog(PANIC, "GIN metapage disappeared");
metapage = BufferGetPage(metabuffer);
if (!XLByteLE(lsn, PageGetLSN(metapage)))
......@@ -469,7 +504,9 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
*/
if (!(record->xl_info & XLR_BKP_BLOCK_1))
{
Buffer buffer = XLogReadBuffer(data->node, data->metadata.tail, false);
buffer = XLogReadBuffer(data->node, data->metadata.tail, false);
if (BufferIsValid(buffer))
{
Page page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
......@@ -505,13 +542,15 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
UnlockReleaseBuffer(buffer);
}
}
}
else if (data->prevTail != InvalidBlockNumber)
{
/*
* New tail
*/
Buffer buffer = XLogReadBuffer(data->node, data->prevTail, false);
buffer = XLogReadBuffer(data->node, data->prevTail, false);
if (BufferIsValid(buffer))
{
Page page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
......@@ -524,6 +563,7 @@ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
}
UnlockReleaseBuffer(buffer);
}
}
UnlockReleaseBuffer(metabuffer);
}
......@@ -544,6 +584,7 @@ ginRedoInsertListPage(XLogRecPtr lsn, XLogRecord *record)
return;
buffer = XLogReadBuffer(data->node, data->blkno, true);
Assert(BufferIsValid(buffer));
page = BufferGetPage(buffer);
GinInitBuffer(buffer, GIN_LIST);
......@@ -587,6 +628,8 @@ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
int i;
metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
if (!BufferIsValid(metabuffer))
elog(PANIC, "GIN metapage disappeared");
metapage = BufferGetPage(metabuffer);
if (!XLByteLE(lsn, PageGetLSN(metapage)))
......@@ -600,6 +643,9 @@ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
for (i = 0; i < data->ndeleted; i++)
{
Buffer buffer = XLogReadBuffer(data->node, data->toDelete[i], false);
if (BufferIsValid(buffer))
{
Page page = BufferGetPage(buffer);
if (!XLByteLE(lsn, PageGetLSN(page)))
......@@ -613,6 +659,7 @@ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
UnlockReleaseBuffer(buffer);
}
}
UnlockReleaseBuffer(metabuffer);
}
......@@ -755,6 +802,13 @@ ginContinueSplit(ginIncompleteSplit *split)
*/
buffer = XLogReadBuffer(split->node, split->leftBlkno, false);
/*
* Failure should be impossible here, because we wrote the page earlier.
*/
if (!BufferIsValid(buffer))
elog(PANIC, "ginContinueSplit: left block %u not found",
split->leftBlkno);
reln = CreateFakeRelcacheEntry(split->node);
if (split->rootBlkno == GIN_ROOT_BLKNO)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment