external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

wal without requiring FPI

custom-rmgr-20220714
Jeff Davis 2022-04-07 09:25:13 -07:00
parent 95762054a5
commit 46f4aebd42
1 changed files with 27 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/backend/columnar/columnar_wal.c
View File

@ -98,20 +98,24 @@ columnar_wal_page_overwrite(Relation rel, BlockNumber blockno, char *buf,
Page page = BufferGetPage(buffer); Page page = BufferGetPage(buffer);
PageHeader phdr = (PageHeader) page; PageHeader phdr = (PageHeader) page;
PageInit(page, BLCKSZ, 0);
if (len > phdr->pd_upper - phdr->pd_lower) if (PageIsNew(page))
PageInit(page, BLCKSZ, 0);
if (len > phdr->pd_upper - SizeOfPageHeaderData)
{ {
elog(ERROR, elog(ERROR,
"write of columnar data of length %d exceeds page size on block %d of relation %d", "write of columnar data of length %d exceeds page size on block %d of relation %d",
len, blockno, rel->rd_id); len, blockno, rel->rd_id);
} }
memcpy_s(page + phdr->pd_lower, phdr->pd_upper - phdr->pd_lower, buf, len); memcpy_s(page + SizeOfPageHeaderData, phdr->pd_upper - SizeOfPageHeaderData, buf, len);
phdr->pd_lower += len; phdr->pd_lower = SizeOfPageHeaderData + len;
MarkBufferDirty(buffer);
XLogBeginInsert(); XLogBeginInsert();
XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | REGBUF_FORCE_IMAGE); XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
XLogRegisterBufData(0, buf, len);
recptr = XLogInsert(RM_COLUMNAR_ID, XLOG_COLUMNAR_PAGE_OVERWRITE); recptr = XLogInsert(RM_COLUMNAR_ID, XLOG_COLUMNAR_PAGE_OVERWRITE);
PageSetLSN(page, recptr); PageSetLSN(page, recptr);
@ -228,8 +232,24 @@ columnar_redo_page_overwrite(XLogReaderState *record)
XLogRedoAction action = XLogReadBufferForRedo(record, 0, &buffer); XLogRedoAction action = XLogReadBufferForRedo(record, 0, &buffer);
/* full page image is forced, no redo necessary */ if (action == BLK_NEEDS_REDO)
Assert(action != BLK_NEEDS_REDO); {
XLogRecPtr lsn = record->EndRecPtr;
Page page = BufferGetPage(buffer);
PageHeader phdr = (PageHeader) page;
Size datalen;
char *data = XLogRecGetBlockData(record, 0, &datalen);
PageInit(page, BLCKSZ, 0);
memcpy_s(page + phdr->pd_lower, phdr->pd_upper - phdr->pd_lower,
data, datalen);
phdr->pd_lower += datalen;
PageSetLSN(page, lsn);
MarkBufferDirty(buffer);
}
if (BufferIsValid(buffer)) if (BufferIsValid(buffer))
UnlockReleaseBuffer(buffer); UnlockReleaseBuffer(buffer);