57 int segsize, const
char *waldir);
60 #define MAX_ERRORMSG_LEN 1000
66 #define DEFAULT_DECODE_BUFFER_SIZE (64 * 1024)
83 state->errormsg_deferred =
true;
96 state->decode_buffer = buffer;
97 state->decode_buffer_size = size;
98 state->decode_buffer_tail = buffer;
99 state->decode_buffer_head = buffer;
120 state->routine = *routine;
142 state->private_data = private_data;
146 if (!
state->errormsg_buf)
152 state->errormsg_buf[0] =
'\0';
172 if (
state->seg.ws_file != -1)
175 if (
state->decode_buffer &&
state->free_decode_buffer)
179 if (
state->readRecordBuf)
199 uint32 newSize = reclength;
201 newSize += XLOG_BLCKSZ - (newSize % XLOG_BLCKSZ);
202 newSize =
Max(newSize, 5 *
Max(BLCKSZ, XLOG_BLCKSZ));
223 if (
state->readRecordBuf)
225 state->readRecordBuf =
227 if (
state->readRecordBuf == NULL)
229 state->readRecordBufSize = 0;
232 state->readRecordBufSize = newSize;
241 int segsize,
const char *waldir)
271 state->EndRecPtr = RecPtr;
272 state->NextRecPtr = RecPtr;
294 record =
state->record;
297 state->record = NULL;
301 if (
state->decode_queue_tail == record)
302 state->decode_queue_tail = NULL;
313 Assert(
state->decode_buffer_head == (
char *) record);
320 record = record->next;
321 while (
unlikely(record && record->oversized))
322 record = record->next;
327 state->decode_buffer_head = (
char *) record;
337 state->decode_buffer_head =
state->decode_buffer;
338 state->decode_buffer_tail =
state->decode_buffer;
363 if (
state->decode_queue_head == NULL)
366 if (
state->errormsg_deferred)
368 if (
state->errormsg_buf[0] !=
'\0')
369 *errormsg =
state->errormsg_buf;
370 state->errormsg_deferred =
false;
402 return state->record;
479 if (
state->decode_buffer_size == 0)
482 state->decode_buffer_head =
state->decode_buffer;
483 state->decode_buffer_tail =
state->decode_buffer;
484 state->free_decode_buffer =
true;
488 if (
state->decode_buffer_tail >=
state->decode_buffer_head)
491 if (
state->decode_buffer_tail + required_space <=
492 state->decode_buffer +
state->decode_buffer_size)
499 else if (
state->decode_buffer + required_space <
500 state->decode_buffer_head)
511 if (
state->decode_buffer_tail + required_space <
512 state->decode_buffer_head)
559 state->errormsg_buf[0] =
'\0';
565 RecPtr =
state->NextRecPtr;
592 state->nonblocking = nonblocking;
593 state->currRecPtr = RecPtr;
596 targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
597 targetRecOff = RecPtr % XLOG_BLCKSZ;
616 if (targetRecOff == 0)
621 RecPtr += pageHeaderSize;
622 targetRecOff = pageHeaderSize;
624 else if (targetRecOff < pageHeaderSize)
628 pageHeaderSize, targetRecOff);
633 targetRecOff == pageHeaderSize)
676 "invalid record length at %X/%X: expected at least %u, got %u",
705 "out of memory while trying to decode a record of length %u", total_len);
709 len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
723 if (total_len >
state->readRecordBufSize &&
733 memcpy(
state->readRecordBuf,
734 state->readBuf + RecPtr % XLOG_BLCKSZ,
len);
735 buffer =
state->readRecordBuf +
len;
741 targetPagePtr += XLOG_BLCKSZ;
767 state->overwrittenRecPtr = RecPtr;
768 RecPtr = targetPagePtr;
776 "there is no contrecord flag at %X/%X",
789 "invalid contrecord length %u (expected %lld) at %X/%X",
791 ((
long long) total_len) - gotlen,
805 contdata = (
char *)
state->readBuf + pageHeaderSize;
806 len = XLOG_BLCKSZ - pageHeaderSize;
812 pageHeaderSize +
len);
814 memcpy(buffer, (
char *) contdata,
len);
827 }
while (gotlen < total_len);
836 state->DecodeRecPtr = RecPtr;
837 state->NextRecPtr = targetPagePtr + pageHeaderSize
844 Min(targetRecOff + total_len, XLOG_BLCKSZ));
856 state->DecodeRecPtr = RecPtr;
862 if (record->
xl_rmid == RM_XLOG_ID &&
866 state->NextRecPtr +=
state->segcxt.ws_segsize - 1;
883 if ((
char *) decoded ==
state->decode_buffer)
886 state->decode_buffer_tail += decoded->
size;
891 if (
state->decode_queue_tail)
892 state->decode_queue_tail->
next = decoded;
893 state->decode_queue_tail = decoded;
894 if (!
state->decode_queue_head)
895 state->decode_queue_head = decoded;
913 state->abortedRecPtr = RecPtr;
914 state->missingContrecPtr = targetPagePtr;
922 if (!
state->errormsg_buf[0])
924 "missing contrecord at %X/%X",
959 if (
state->errormsg_deferred)
966 return state->decode_queue_tail;
996 Assert((pageptr % XLOG_BLCKSZ) == 0);
1002 if (targetSegNo ==
state->seg.ws_segno &&
1003 targetPageOff ==
state->segoff && reqLen <= state->
readLen)
1004 return state->readLen;
1025 if (targetSegNo !=
state->seg.ws_segno && targetPageOff != 0)
1027 XLogRecPtr targetSegmentPtr = pageptr - targetPageOff;
1086 state->seg.ws_segno = targetSegNo;
1087 state->segoff = targetPageOff;
1104 state->seg.ws_segno = 0;
1123 "invalid record length at %X/%X: expected at least %u, got %u",
1131 "invalid resource manager ID %u at %X/%X",
1141 if (!(record->
xl_prev < RecPtr))
1144 "record with incorrect prev-link %X/%X at %X/%X",
1157 if (record->
xl_prev != PrevRecPtr)
1160 "record with incorrect prev-link %X/%X at %X/%X",
1196 "incorrect resource manager data checksum in record at %X/%X",
1218 Assert((recptr % XLOG_BLCKSZ) == 0);
1230 "invalid magic number %04X in WAL segment %s, LSN %X/%X, offset %u",
1245 "invalid info bits %04X in WAL segment %s, LSN %X/%X, offset %u",
1257 if (
state->system_identifier &&
1261 "WAL file is from different database system: WAL file database system identifier is %llu, pg_control database system identifier is %llu",
1262 (
unsigned long long) longhdr->
xlp_sysid,
1263 (
unsigned long long)
state->system_identifier);
1269 "WAL file is from different database system: incorrect segment size in page header");
1275 "WAL file is from different database system: incorrect XLOG_BLCKSZ in page header");
1279 else if (offset == 0)
1287 "invalid info bits %04X in WAL segment %s, LSN %X/%X, offset %u",
1307 "unexpected pageaddr %X/%X in WAL segment %s, LSN %X/%X, offset %u",
1324 if (recptr >
state->latestPagePtr)
1333 "out-of-sequence timeline ID %u (after %u) in WAL segment %s, LSN %X/%X, offset %u",
1335 state->latestPageTLI,
1342 state->latestPagePtr = recptr;
1354 state->errormsg_buf[0] =
'\0';
1355 state->errormsg_deferred =
false;
1380 state->nonblocking =
false;
1403 targetRecOff = tmpRecPtr % XLOG_BLCKSZ;
1406 targetPagePtr = tmpRecPtr - targetRecOff;
1434 if (
MAXALIGN(
header->xlp_rem_len) >= (XLOG_BLCKSZ - pageHeaderSize))
1435 tmpRecPtr = targetPagePtr + XLOG_BLCKSZ;
1442 tmpRecPtr = targetPagePtr + pageHeaderSize
1449 tmpRecPtr = targetPagePtr + pageHeaderSize;
1463 if (RecPtr <= state->ReadRecPtr)
1466 found =
state->ReadRecPtr;
1518 if (
state->seg.ws_file < 0 ||
1520 tli !=
state->seg.ws_tli)
1524 if (
state->seg.ws_file >= 0)
1528 state->routine.segment_open(
state, nextSegNo, &tli);
1534 state->seg.ws_tli = tli;
1535 state->seg.ws_segno = nextSegNo;
1539 if (nbytes > (
state->segcxt.ws_segsize - startoff))
1540 segbytes =
state->segcxt.ws_segsize - startoff;
1550 readbytes =
pg_pread(
state->seg.ws_file, p, segbytes, (off_t) startoff);
1567 recptr += readbytes;
1568 nbytes -= readbytes;
1590 while ((r =
state->decode_queue_head) != NULL)
1596 state->decode_queue_tail = NULL;
1597 state->decode_queue_head = NULL;
1598 state->record = NULL;
1601 state->decode_buffer_tail =
state->decode_buffer;
1602 state->decode_buffer_head =
state->decode_buffer;
1605 state->errormsg_buf[0] =
'\0';
1606 state->errormsg_deferred =
false;
1630 size += (MAXIMUM_ALIGNOF - 1);
1634 size += (MAXIMUM_ALIGNOF - 1);
1661 #define COPY_HEADER_FIELD(_dst, _size) \
1663 if (remaining < _size) \
1664 goto shortdata_err; \
1665 memcpy(_dst, ptr, _size); \
1667 remaining -= _size; \
1677 decoded->
header = *record;
1679 decoded->
next = NULL;
1685 ptr = (
char *) record;
1698 uint8 main_data_len;
1703 datatotal += main_data_len;
1714 datatotal += main_data_len;
1736 if (block_id <= decoded->max_block_id)
1739 "out-of-order block_id %u at %X/%X",
1746 blk = &decoded->
blocks[block_id];
1752 blk->
flags = fork_flags;
1763 "BKPBLOCK_HAS_DATA set, but no data included at %X/%X",
1770 "BKPBLOCK_HAS_DATA not set, but data length is %u at %X/%X",
1806 "BKPIMAGE_HAS_HOLE set, but hole offset %u length %u block image length %u at %X/%X",
1822 "BKPIMAGE_HAS_HOLE not set, but hole offset %u length %u at %X/%X",
1836 "BKPIMAGE_COMPRESSED set, but block image length %u at %X/%X",
1851 "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %u at %X/%X",
1864 if (rlocator == NULL)
1867 "BKPBLOCK_SAME_REL set but no previous rel at %X/%X",
1879 "invalid block_id %u at %X/%X",
1895 out = ((
char *) decoded) +
1900 for (block_id = 0; block_id <= decoded->
max_block_id; block_id++)
1946 "record with invalid length at %X/%X",
1949 *errormsg =
state->errormsg_buf;
1969 elog(
ERROR,
"could not locate backup block with ID %d in WAL record",
1972 pg_fatal(
"could not locate backup block with ID %d in WAL record",
2003 *blknum = bkpb->
blkno;
2004 if (prefetch_buffer)
2056 "could not restore image at %X/%X with invalid block %d specified",
2075 bool decomp_success =
true;
2081 decomp_success =
false;
2086 if (LZ4_decompress_safe(ptr, tmp.
data,
2088 decomp_success =
false;
2090 report_invalid_record(record,
"could not restore image at %X/%X compressed with %s not supported by build, block %d",
2100 size_t decomp_result = ZSTD_decompress(tmp.
data,
2104 if (ZSTD_isError(decomp_result))
2105 decomp_success =
false;
2107 report_invalid_record(record,
"could not restore image at %X/%X compressed with %s not supported by build, block %d",
2116 report_invalid_record(record,
"could not restore image at %X/%X compressed with unknown method, block %d",
2122 if (!decomp_success)
2136 memcpy(page, ptr, BLCKSZ);
#define pg_attribute_printf(f, a)
#define MemSet(start, val, len)
elog(ERROR, "%s: %s", p2, msg)
void err(int eval, const char *fmt,...)
#define MCXT_ALLOC_NO_OOM
if(TABLE==NULL||TABLE_index==NULL)
static void const char * fmt
Assert(fmt[strlen(fmt) - 1] !='\n')
void pfree(void *pointer)
void * palloc_extended(Size size, int flags)
#define AllocSizeIsValid(size)
#define AmStartupProcess()
#define InvalidRepOriginId
#define COMP_CRC32C(crc, data, len)
#define EQ_CRC32C(c1, c2)
int32 pglz_decompress(const char *source, int32 slen, char *dest, int32 rawsize, bool check_complete)
static void header(const char *fmt,...) pg_attribute_printf(1
#define RmgrIdIsValid(rmid)
struct DecodedXLogRecord * next
TransactionId toplevel_xid
RepOriginId record_origin
DecodedBkpBlock blocks[FLEXIBLE_ARRAY_MEMBER]
FullTransactionId nextXid
DecodedXLogRecord * record
#define InvalidTransactionId
#define EpochFromFullTransactionId(x)
#define XidFromFullTransactionId(x)
static FullTransactionId FullTransactionIdFromEpochAndXid(uint32 epoch, TransactionId xid)
VariableCache ShmemVariableCache
static void pgstat_report_wait_start(uint32 wait_event_info)
static void pgstat_report_wait_end(void)
static const unsigned __int64 epoch
#define XLP_FIRST_IS_CONTRECORD
XLogLongPageHeaderData * XLogLongPageHeader
#define XLP_FIRST_IS_OVERWRITE_CONTRECORD
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
XLogPageHeaderData * XLogPageHeader
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define XRecOffIsValid(xlrp)
#define SizeOfXLogShortPHD
static void XLogFileName(char *fname, TimeLineID tli, XLogSegNo logSegNo, int wal_segsz_bytes)
#define XLogPageHeaderSize(hdr)
#define XLByteInSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr
static void static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength)
bool XLogRecGetBlockTagExtended(XLogReaderState *record, uint8 block_id, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum, Buffer *prefetch_buffer)
static XLogPageReadResult XLogDecodeNextRecord(XLogReaderState *state, bool nonblocking)
void XLogReaderSetDecodeBuffer(XLogReaderState *state, void *buffer, size_t size)
static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt, int segsize, const char *waldir)
static int ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, int reqLen)
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
static void report_invalid_record(XLogReaderState *state, const char *fmt,...) pg_attribute_printf(2
bool WALRead(XLogReaderState *state, char *buf, XLogRecPtr startptr, Size count, TimeLineID tli, WALReadError *errinfo)
DecodedXLogRecord * XLogNextRecord(XLogReaderState *state, char **errormsg)
void XLogReaderResetError(XLogReaderState *state)
static void XLogReaderInvalReadState(XLogReaderState *state)
#define COPY_HEADER_FIELD(_dst, _size)
bool XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr, char *phdr)
FullTransactionId XLogRecGetFullXid(XLogReaderState *record)
void XLogReaderFree(XLogReaderState *state)
void XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileLocator *rlocator, ForkNumber *forknum, BlockNumber *blknum)
DecodedXLogRecord * XLogReadAhead(XLogReaderState *state, bool nonblocking)
XLogReaderState * XLogReaderAllocate(int wal_segment_size, const char *waldir, XLogReaderRoutine *routine, void *private_data)
static void ResetDecoder(XLogReaderState *state)
bool DecodeXLogRecord(XLogReaderState *state, DecodedXLogRecord *decoded, XLogRecord *record, XLogRecPtr lsn, char **errormsg)
static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record, XLogRecPtr recptr)
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
#define DEFAULT_DECODE_BUFFER_SIZE
size_t DecodeXLogRecordRequiredSpace(size_t xl_tot_len)
XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
static DecodedXLogRecord * XLogReadRecordAlloc(XLogReaderState *state, size_t xl_tot_len, bool allow_oversized)
XLogRecPtr XLogReleasePreviousRecord(XLogReaderState *state)
static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr, XLogRecPtr PrevRecPtr, XLogRecord *record, bool randAccess)
static bool XLogReaderHasQueuedRecordOrError(XLogReaderState *state)
#define XLogRecGetXid(decoder)
#define XLogRecHasBlockRef(decoder, block_id)
#define BKPIMAGE_COMPRESS_ZSTD
#define BKPBLOCK_FORK_MASK
#define BKPBLOCK_HAS_DATA
#define BKPIMAGE_HAS_HOLE
#define XLR_BLOCK_ID_DATA_LONG
#define BKPIMAGE_COMPRESS_LZ4
#define BKPIMAGE_COMPRESSED(info)
#define XLR_BLOCK_ID_TOPLEVEL_XID
#define XLR_BLOCK_ID_DATA_SHORT
#define BKPBLOCK_SAME_REL
#define BKPIMAGE_COMPRESS_PGLZ
#define XLR_BLOCK_ID_ORIGIN
#define BKPBLOCK_HAS_IMAGE