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;
293 record =
state->record;
295 state->record = NULL;
299 if (
state->decode_queue_tail == record)
300 state->decode_queue_tail = NULL;
311 Assert(
state->decode_buffer_head == (
char *) record);
318 record = record->next;
319 while (
unlikely(record && record->oversized))
320 record = record->next;
325 state->decode_buffer_head = (
char *) record;
335 state->decode_buffer_head =
state->decode_buffer;
336 state->decode_buffer_tail =
state->decode_buffer;
359 if (
state->decode_queue_head == NULL)
362 if (
state->errormsg_deferred)
364 if (
state->errormsg_buf[0] !=
'\0')
365 *errormsg =
state->errormsg_buf;
366 state->errormsg_deferred =
false;
398 return state->record;
475 if (
state->decode_buffer_size == 0)
478 state->decode_buffer_head =
state->decode_buffer;
479 state->decode_buffer_tail =
state->decode_buffer;
480 state->free_decode_buffer =
true;
484 if (
state->decode_buffer_tail >=
state->decode_buffer_head)
487 if (
state->decode_buffer_tail + required_space <=
488 state->decode_buffer +
state->decode_buffer_size)
495 else if (
state->decode_buffer + required_space <
496 state->decode_buffer_head)
507 if (
state->decode_buffer_tail + required_space <
508 state->decode_buffer_head)
555 state->errormsg_buf[0] =
'\0';
561 RecPtr =
state->NextRecPtr;
587 state->nonblocking = nonblocking;
588 state->currRecPtr = RecPtr;
591 targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
592 targetRecOff = RecPtr % XLOG_BLCKSZ;
611 if (targetRecOff == 0)
616 RecPtr += pageHeaderSize;
617 targetRecOff = pageHeaderSize;
619 else if (targetRecOff < pageHeaderSize)
627 targetRecOff == pageHeaderSize)
670 "invalid record length at %X/%X: wanted %u, got %u",
699 "out of memory while trying to decode a record of length %u", total_len);
703 len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
717 if (total_len >
state->readRecordBufSize &&
727 memcpy(
state->readRecordBuf,
728 state->readBuf + RecPtr % XLOG_BLCKSZ,
len);
729 buffer =
state->readRecordBuf +
len;
735 targetPagePtr += XLOG_BLCKSZ;
761 state->overwrittenRecPtr = RecPtr;
762 RecPtr = targetPagePtr;
770 "there is no contrecord flag at %X/%X",
783 "invalid contrecord length %u (expected %lld) at %X/%X",
785 ((
long long) total_len) - gotlen,
799 contdata = (
char *)
state->readBuf + pageHeaderSize;
800 len = XLOG_BLCKSZ - pageHeaderSize;
806 pageHeaderSize +
len);
808 memcpy(buffer, (
char *) contdata,
len);
821 }
while (gotlen < total_len);
830 state->DecodeRecPtr = RecPtr;
831 state->NextRecPtr = targetPagePtr + pageHeaderSize
838 Min(targetRecOff + total_len, XLOG_BLCKSZ));
850 state->DecodeRecPtr = RecPtr;
856 if (record->
xl_rmid == RM_XLOG_ID &&
860 state->NextRecPtr +=
state->segcxt.ws_segsize - 1;
877 if ((
char *) decoded ==
state->decode_buffer)
880 state->decode_buffer_tail += decoded->
size;
885 if (
state->decode_queue_tail)
886 state->decode_queue_tail->
next = decoded;
887 state->decode_queue_tail = decoded;
888 if (!
state->decode_queue_head)
889 state->decode_queue_head = decoded;
907 state->abortedRecPtr = RecPtr;
908 state->missingContrecPtr = targetPagePtr;
942 if (
state->errormsg_deferred)
949 return state->decode_queue_tail;
979 Assert((pageptr % XLOG_BLCKSZ) == 0);
985 if (targetSegNo ==
state->seg.ws_segno &&
986 targetPageOff ==
state->segoff && reqLen <= state->
readLen)
987 return state->readLen;
1001 if (targetSegNo !=
state->seg.ws_segno && targetPageOff != 0)
1003 XLogRecPtr targetSegmentPtr = pageptr - targetPageOff;
1062 state->seg.ws_segno = targetSegNo;
1063 state->segoff = targetPageOff;
1069 if (
state->errormsg_buf[0] !=
'\0')
1071 state->errormsg_deferred =
true;
1083 state->seg.ws_segno = 0;
1102 "invalid record length at %X/%X: wanted %u, got %u",
1110 "invalid resource manager ID %u at %X/%X",
1120 if (!(record->
xl_prev < RecPtr))
1123 "record with incorrect prev-link %X/%X at %X/%X",
1136 if (record->
xl_prev != PrevRecPtr)
1139 "record with incorrect prev-link %X/%X at %X/%X",
1175 "incorrect resource manager data checksum in record at %X/%X",
1198 Assert((recptr % XLOG_BLCKSZ) == 0);
1212 "invalid magic number %04X in log segment %s, offset %u",
1226 "invalid info bits %04X in log segment %s, offset %u",
1237 if (
state->system_identifier &&
1241 "WAL file is from different database system: WAL file database system identifier is %llu, pg_control database system identifier is %llu",
1242 (
unsigned long long) longhdr->
xlp_sysid,
1243 (
unsigned long long)
state->system_identifier);
1249 "WAL file is from different database system: incorrect segment size in page header");
1255 "WAL file is from different database system: incorrect XLOG_BLCKSZ in page header");
1259 else if (offset == 0)
1267 "invalid info bits %04X in log segment %s, offset %u",
1286 "unexpected pageaddr %X/%X in log segment %s, offset %u",
1302 if (recptr >
state->latestPagePtr)
1311 "out-of-sequence timeline ID %u (after %u) in log segment %s, offset %u",
1313 state->latestPageTLI,
1319 state->latestPagePtr = recptr;
1347 state->nonblocking =
false;
1370 targetRecOff = tmpRecPtr % XLOG_BLCKSZ;
1373 targetPagePtr = tmpRecPtr - targetRecOff;
1401 if (
MAXALIGN(
header->xlp_rem_len) >= (XLOG_BLCKSZ - pageHeaderSize))
1402 tmpRecPtr = targetPagePtr + XLOG_BLCKSZ;
1409 tmpRecPtr = targetPagePtr + pageHeaderSize
1416 tmpRecPtr = targetPagePtr + pageHeaderSize;
1430 if (RecPtr <= state->ReadRecPtr)
1433 found =
state->ReadRecPtr;
1485 if (
state->seg.ws_file < 0 ||
1487 tli !=
state->seg.ws_tli)
1491 if (
state->seg.ws_file >= 0)
1495 state->routine.segment_open(
state, nextSegNo, &tli);
1501 state->seg.ws_tli = tli;
1502 state->seg.ws_segno = nextSegNo;
1506 if (nbytes > (
state->segcxt.ws_segsize - startoff))
1507 segbytes =
state->segcxt.ws_segsize - startoff;
1517 readbytes =
pg_pread(
state->seg.ws_file, p, segbytes, (off_t) startoff);
1534 recptr += readbytes;
1535 nbytes -= readbytes;
1557 while ((r =
state->decode_queue_head) != NULL)
1563 state->decode_queue_tail = NULL;
1564 state->decode_queue_head = NULL;
1565 state->record = NULL;
1568 state->decode_buffer_tail =
state->decode_buffer;
1569 state->decode_buffer_head =
state->decode_buffer;
1572 state->errormsg_buf[0] =
'\0';
1573 state->errormsg_deferred =
false;
1597 size += (MAXIMUM_ALIGNOF - 1);
1601 size += (MAXIMUM_ALIGNOF - 1);
1628 #define COPY_HEADER_FIELD(_dst, _size) \
1630 if (remaining < _size) \
1631 goto shortdata_err; \
1632 memcpy(_dst, ptr, _size); \
1634 remaining -= _size; \
1644 decoded->
header = *record;
1646 decoded->
next = NULL;
1652 ptr = (
char *) record;
1665 uint8 main_data_len;
1670 datatotal += main_data_len;
1681 datatotal += main_data_len;
1703 if (block_id <= decoded->max_block_id)
1706 "out-of-order block_id %u at %X/%X",
1713 blk = &decoded->
blocks[block_id];
1719 blk->
flags = fork_flags;
1730 "BKPBLOCK_HAS_DATA set, but no data included at %X/%X",
1737 "BKPBLOCK_HAS_DATA not set, but data length is %u at %X/%X",
1773 "BKPIMAGE_HAS_HOLE set, but hole offset %u length %u block image length %u at %X/%X",
1789 "BKPIMAGE_HAS_HOLE not set, but hole offset %u length %u at %X/%X",
1803 "BKPIMAGE_COMPRESSED set, but block image length %u at %X/%X",
1818 "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %u at %X/%X",
1827 rnode = &blk->
rnode;
1834 "BKPBLOCK_SAME_REL set but no previous rel at %X/%X",
1839 blk->
rnode = *rnode;
1846 "invalid block_id %u at %X/%X",
1862 out = ((
char *) decoded) +
1867 for (block_id = 0; block_id <= decoded->
max_block_id; block_id++)
1913 "record with invalid length at %X/%X",
1916 *errormsg =
state->errormsg_buf;
1935 elog(
ERROR,
"failed to locate backup block with ID %d in WAL record",
1938 pg_fatal(
"failed to locate backup block with ID %d in WAL record",
1965 *rnode = bkpb->
rnode;
1969 *blknum = bkpb->
blkno;
1970 if (prefetch_buffer)
2029 bool decomp_success =
true;
2035 decomp_success =
false;
2040 if (LZ4_decompress_safe(ptr, tmp.
data,
2042 decomp_success =
false;
2054 size_t decomp_result = ZSTD_decompress(tmp.
data,
2058 if (ZSTD_isError(decomp_result))
2059 decomp_success =
false;
2076 if (!decomp_success)
2090 memcpy(page, ptr, BLCKSZ);
#define offsetof(type, field)
#define pg_attribute_printf(f, a)
#define MemSet(start, val, len)
#define MCXT_ALLOC_NO_OOM
static const unsigned __int64 epoch
Assert(fmt[strlen(fmt) - 1] !='\n')
static void const char * fmt
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
ssize_t pg_pread(int fd, void *buf, size_t nbyte, off_t offset)
#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)
#define XLP_FIRST_IS_CONTRECORD
XLogLongPageHeaderData * XLogLongPageHeader
#define XLP_FIRST_IS_OVERWRITE_CONTRECORD
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
#define XLogFileName(fname, tli, logSegNo, wal_segsz_bytes)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
XLogPageHeaderData * XLogPageHeader
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define XRecOffIsValid(xlrp)
#define SizeOfXLogShortPHD
#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)
void XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
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)
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)
static XLogPageReadResult XLogDecodeNextRecord(XLogReaderState *state, bool non_blocking)
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)
void XLogReleasePreviousRecord(XLogReaderState *state)
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)
bool XLogRecGetBlockTagExtended(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum, Buffer *prefetch_buffer)
static DecodedXLogRecord * XLogReadRecordAlloc(XLogReaderState *state, size_t xl_tot_len, bool allow_oversized)
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