Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 38a9573

Browse filesBrowse files
hlinnakahlinnaka
committed
Refactor XLogReadRecord(), adding XLogBeginRead() function.
The signature of XLogReadRecord() required the caller to pass the starting WAL position as argument, or InvalidXLogRecPtr to continue reading at the end of previous record. That's slightly awkward to the callers, as most of them don't want to randomly jump around in the WAL stream, but start reading at one position and then read everything from that point onwards. Remove the 'RecPtr' argument and add a new function XLogBeginRead() to specify the starting position instead. That's more convenient for the callers. Also, xlogreader holds state that is reset when you change the starting position, so having a separate function for doing that feels like a more natural fit. This changes XLogFindNextRecord() function so that it doesn't reset the xlogreader's state to what it was before the call anymore. Instead, it positions the xlogreader to the found record, like XLogBeginRead(). Reviewed-by: Kyotaro Horiguchi, Alvaro Herrera Discussion: https://www.postgresql.org/message-id/5382a7a3-debe-be31-c860-cb810c08f366%40iki.fi
1 parent 1001368 commit 38a9573
Copy full SHA for 38a9573

File tree

Expand file treeCollapse file tree

10 files changed

+102
-97
lines changed
Filter options
Expand file treeCollapse file tree

10 files changed

+102
-97
lines changed

‎src/backend/access/transam/twophase.c

Copy file name to clipboardExpand all lines: src/backend/access/transam/twophase.c
+2-1Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1338,7 +1338,8 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
13381338
errmsg("out of memory"),
13391339
errdetail("Failed while allocating a WAL reading processor.")));
13401340

1341-
record = XLogReadRecord(xlogreader, lsn, &errormsg);
1341+
XLogBeginRead(xlogreader, lsn);
1342+
record = XLogReadRecord(xlogreader, &errormsg);
13421343
if (record == NULL)
13431344
ereport(ERROR,
13441345
(errcode_for_file_access(),

‎src/backend/access/transam/xlog.c

Copy file name to clipboardExpand all lines: src/backend/access/transam/xlog.c
+19-17Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -897,7 +897,7 @@ static void UpdateLastRemovedPtr(char *filename);
897897
static void ValidateXLOGDirectoryStructure(void);
898898
static void CleanupBackupHistory(void);
899899
static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
900-
static XLogRecord *ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
900+
static XLogRecord *ReadRecord(XLogReaderState *xlogreader,
901901
int emode, bool fetching_ckpt);
902902
static void CheckRecoveryConsistency(void);
903903
static XLogRecord *ReadCheckpointRecord(XLogReaderState *xlogreader,
@@ -4246,17 +4246,17 @@ CleanupBackupHistory(void)
42464246
}
42474247

42484248
/*
4249-
* Attempt to read an XLOG record.
4249+
* Attempt to read the next XLOG record.
42504250
*
4251-
* If RecPtr is valid, try to read a record at that position. Otherwise
4252-
* try to read a record just after the last one previously read.
4251+
* Before first call, the reader needs to be positioned to the first record
4252+
* by calling XLogBeginRead().
42534253
*
42544254
* If no valid record is available, returns NULL, or fails if emode is PANIC.
42554255
* (emode must be either PANIC, LOG). In standby mode, retries until a valid
42564256
* record is available.
42574257
*/
42584258
static XLogRecord *
4259-
ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
4259+
ReadRecord(XLogReaderState *xlogreader, int emode,
42604260
bool fetching_ckpt)
42614261
{
42624262
XLogRecord *record;
@@ -4265,7 +4265,7 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
42654265
/* Pass through parameters to XLogPageRead */
42664266
private->fetching_ckpt = fetching_ckpt;
42674267
private->emode = emode;
4268-
private->randAccess = (RecPtr != InvalidXLogRecPtr);
4268+
private->randAccess = (xlogreader->ReadRecPtr != InvalidXLogRecPtr);
42694269

42704270
/* This is the first attempt to read this page. */
42714271
lastSourceFailed = false;
@@ -4274,7 +4274,7 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
42744274
{
42754275
char *errormsg;
42764276

4277-
record = XLogReadRecord(xlogreader, RecPtr, &errormsg);
4277+
record = XLogReadRecord(xlogreader, &errormsg);
42784278
ReadRecPtr = xlogreader->ReadRecPtr;
42794279
EndRecPtr = xlogreader->EndRecPtr;
42804280
if (record == NULL)
@@ -4292,8 +4292,7 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
42924292
* shouldn't loop anymore in that case.
42934293
*/
42944294
if (errormsg)
4295-
ereport(emode_for_corrupt_record(emode,
4296-
RecPtr ? RecPtr : EndRecPtr),
4295+
ereport(emode_for_corrupt_record(emode, EndRecPtr),
42974296
(errmsg_internal("%s", errormsg) /* already translated */ ));
42984297
}
42994298

@@ -4311,8 +4310,7 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode,
43114310
wal_segment_size);
43124311
XLogFileName(fname, xlogreader->seg.ws_tli, segno,
43134312
wal_segment_size);
4314-
ereport(emode_for_corrupt_record(emode,
4315-
RecPtr ? RecPtr : EndRecPtr),
4313+
ereport(emode_for_corrupt_record(emode, EndRecPtr),
43164314
(errmsg("unexpected timeline ID %u in log segment %s, offset %u",
43174315
xlogreader->latestPageTLI,
43184316
fname,
@@ -6427,7 +6425,8 @@ StartupXLOG(void)
64276425
*/
64286426
if (checkPoint.redo < checkPointLoc)
64296427
{
6430-
if (!ReadRecord(xlogreader, checkPoint.redo, LOG, false))
6428+
XLogBeginRead(xlogreader, checkPoint.redo);
6429+
if (!ReadRecord(xlogreader, LOG, false))
64316430
ereport(FATAL,
64326431
(errmsg("could not find redo location referenced by checkpoint record"),
64336432
errhint("If you are restoring from a backup, touch \"%s/recovery.signal\" and add required recovery options.\n"
@@ -7034,12 +7033,13 @@ StartupXLOG(void)
70347033
if (checkPoint.redo < RecPtr)
70357034
{
70367035
/* back up to find the record */
7037-
record = ReadRecord(xlogreader, checkPoint.redo, PANIC, false);
7036+
XLogBeginRead(xlogreader, checkPoint.redo);
7037+
record = ReadRecord(xlogreader, PANIC, false);
70387038
}
70397039
else
70407040
{
70417041
/* just have to read next record after CheckPoint */
7042-
record = ReadRecord(xlogreader, InvalidXLogRecPtr, LOG, false);
7042+
record = ReadRecord(xlogreader, LOG, false);
70437043
}
70447044

70457045
if (record != NULL)
@@ -7263,7 +7263,7 @@ StartupXLOG(void)
72637263
}
72647264

72657265
/* Else, try to fetch the next WAL record */
7266-
record = ReadRecord(xlogreader, InvalidXLogRecPtr, LOG, false);
7266+
record = ReadRecord(xlogreader, LOG, false);
72677267
} while (record != NULL);
72687268

72697269
/*
@@ -7365,7 +7365,8 @@ StartupXLOG(void)
73657365
* Re-fetch the last valid or last applied record, so we can identify the
73667366
* exact endpoint of what we consider the valid portion of WAL.
73677367
*/
7368-
record = ReadRecord(xlogreader, LastRec, PANIC, false);
7368+
XLogBeginRead(xlogreader, LastRec);
7369+
record = ReadRecord(xlogreader, PANIC, false);
73697370
EndOfLog = EndRecPtr;
73707371

73717372
/*
@@ -8094,7 +8095,8 @@ ReadCheckpointRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
80948095
return NULL;
80958096
}
80968097

8097-
record = ReadRecord(xlogreader, RecPtr, LOG, true);
8098+
XLogBeginRead(xlogreader, RecPtr);
8099+
record = ReadRecord(xlogreader, LOG, true);
80988100

80998101
if (record == NULL)
81008102
{

‎src/backend/access/transam/xlogreader.c

Copy file name to clipboardExpand all lines: src/backend/access/transam/xlogreader.c
+50-29Lines changed: 50 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -218,11 +218,34 @@ WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
218218
snprintf(segcxt->ws_dir, MAXPGPATH, "%s", waldir);
219219
}
220220

221+
/*
222+
* Begin reading WAL at 'RecPtr'.
223+
*
224+
* 'RecPtr' should point to the beginnning of a valid WAL record. Pointing at
225+
* the beginning of a page is also OK, if there is a new record right after
226+
* the page header, i.e. not a continuation.
227+
*
228+
* This does not make any attempt to read the WAL yet, and hence cannot fail.
229+
* If the starting address is not correct, the first call to XLogReadRecord()
230+
* will error out.
231+
*/
232+
void
233+
XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
234+
{
235+
Assert(!XLogRecPtrIsInvalid(RecPtr));
236+
237+
ResetDecoder(state);
238+
239+
/* Begin at the passed-in record pointer. */
240+
state->EndRecPtr = RecPtr;
241+
state->ReadRecPtr = InvalidXLogRecPtr;
242+
}
243+
221244
/*
222245
* Attempt to read an XLOG record.
223246
*
224-
* If RecPtr is valid, try to read a record at that position. Otherwise
225-
* try to read a record just after the last one previously read.
247+
* XLogBeginRead() or XLogFindNextRecord() must be called before the first call
248+
* to XLogReadRecord().
226249
*
227250
* If the read_page callback fails to read the requested data, NULL is
228251
* returned. The callback is expected to have reported the error; errormsg
@@ -235,8 +258,9 @@ WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
235258
* valid until the next call to XLogReadRecord.
236259
*/
237260
XLogRecord *
238-
XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
261+
XLogReadRecord(XLogReaderState *state, char **errormsg)
239262
{
263+
XLogRecPtr RecPtr;
240264
XLogRecord *record;
241265
XLogRecPtr targetPagePtr;
242266
bool randAccess;
@@ -260,28 +284,26 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
260284

261285
ResetDecoder(state);
262286

263-
if (RecPtr == InvalidXLogRecPtr)
264-
{
265-
/* No explicit start point; read the record after the one we just read */
266-
RecPtr = state->EndRecPtr;
287+
RecPtr = state->EndRecPtr;
267288

268-
if (state->ReadRecPtr == InvalidXLogRecPtr)
269-
randAccess = true;
289+
if (state->ReadRecPtr != InvalidXLogRecPtr)
290+
{
291+
/* read the record after the one we just read */
270292

271293
/*
272-
* RecPtr is pointing to end+1 of the previous WAL record. If we're
273-
* at a page boundary, no more records can fit on the current page. We
274-
* must skip over the page header, but we can't do that until we've
275-
* read in the page, since the header size is variable.
294+
* EndRecPtr is pointing to end+1 of the previous WAL record. If
295+
* we're at a page boundary, no more records can fit on the current
296+
* page. We must skip over the page header, but we can't do that until
297+
* we've read in the page, since the header size is variable.
276298
*/
277299
}
278300
else
279301
{
280302
/*
281303
* Caller supplied a position to start at.
282304
*
283-
* In this case, the passed-in record pointer should already be
284-
* pointing to a valid record starting position.
305+
* In this case, EndRecPtr should already be pointing to a valid
306+
* record starting position.
285307
*/
286308
Assert(XRecOffIsValid(RecPtr));
287309
randAccess = true;
@@ -899,14 +921,17 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
899921
/*
900922
* Find the first record with an lsn >= RecPtr.
901923
*
902-
* Useful for checking whether RecPtr is a valid xlog address for reading, and
903-
* to find the first valid address after some address when dumping records for
904-
* debugging purposes.
924+
* This is different from XLogBeginRead() in that RecPtr doesn't need to point
925+
* to a valid record boundary. Useful for checking whether RecPtr is a valid
926+
* xlog address for reading, and to find the first valid address after some
927+
* address when dumping records for debugging purposes.
928+
*
929+
* This positions the reader, like XLogBeginRead(), so that the next call to
930+
* XLogReadRecord() will read the next valid record.
905931
*/
906932
XLogRecPtr
907933
XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
908934
{
909-
XLogReaderState saved_state = *state;
910935
XLogRecPtr tmpRecPtr;
911936
XLogRecPtr found = InvalidXLogRecPtr;
912937
XLogPageHeader header;
@@ -991,27 +1016,23 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
9911016
* because either we're at the first record after the beginning of a page
9921017
* or we just jumped over the remaining data of a continuation.
9931018
*/
994-
while (XLogReadRecord(state, tmpRecPtr, &errormsg) != NULL)
1019+
XLogBeginRead(state, tmpRecPtr);
1020+
while (XLogReadRecord(state, &errormsg) != NULL)
9951021
{
996-
/* continue after the record */
997-
tmpRecPtr = InvalidXLogRecPtr;
998-
9991022
/* past the record we've found, break out */
10001023
if (RecPtr <= state->ReadRecPtr)
10011024
{
1025+
/* Rewind the reader to the beginning of the last record. */
10021026
found = state->ReadRecPtr;
1003-
goto out;
1027+
XLogBeginRead(state, found);
1028+
return found;
10041029
}
10051030
}
10061031

10071032
err:
1008-
out:
1009-
/* Reset state to what we had before finding the record */
1010-
state->ReadRecPtr = saved_state.ReadRecPtr;
1011-
state->EndRecPtr = saved_state.EndRecPtr;
10121033
XLogReaderInvalReadState(state);
10131034

1014-
return found;
1035+
return InvalidXLogRecPtr;
10151036
}
10161037

10171038
#endif /* FRONTEND */

‎src/backend/replication/logical/logical.c

Copy file name to clipboardExpand all lines: src/backend/replication/logical/logical.c
+2-5Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -461,11 +461,10 @@ DecodingContextReady(LogicalDecodingContext *ctx)
461461
void
462462
DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
463463
{
464-
XLogRecPtr startptr;
465464
ReplicationSlot *slot = ctx->slot;
466465

467466
/* Initialize from where to start reading WAL. */
468-
startptr = slot->data.restart_lsn;
467+
XLogBeginRead(ctx->reader, slot->data.restart_lsn);
469468

470469
elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X",
471470
(uint32) (slot->data.restart_lsn >> 32),
@@ -478,14 +477,12 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
478477
char *err = NULL;
479478

480479
/* the read_page callback waits for new WAL */
481-
record = XLogReadRecord(ctx->reader, startptr, &err);
480+
record = XLogReadRecord(ctx->reader, &err);
482481
if (err)
483482
elog(ERROR, "%s", err);
484483
if (!record)
485484
elog(ERROR, "no record found"); /* shouldn't happen */
486485

487-
startptr = InvalidXLogRecPtr;
488-
489486
LogicalDecodingProcessRecord(ctx, ctx->reader);
490487

491488
/* only continue till we found a consistent spot */

‎src/backend/replication/logical/logicalfuncs.c

Copy file name to clipboardExpand all lines: src/backend/replication/logical/logicalfuncs.c
+3-11Lines changed: 3 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,6 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
127127
MemoryContext per_query_ctx;
128128
MemoryContext oldcontext;
129129
XLogRecPtr end_of_wal;
130-
XLogRecPtr startptr;
131130
LogicalDecodingContext *ctx;
132131
ResourceOwner old_resowner = CurrentResourceOwner;
133132
ArrayType *arr;
@@ -269,28 +268,21 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
269268
* xacts that committed after the slot's confirmed_flush can be
270269
* accumulated into reorder buffers.
271270
*/
272-
startptr = MyReplicationSlot->data.restart_lsn;
271+
XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
273272

274273
/* invalidate non-timetravel entries */
275274
InvalidateSystemCaches();
276275

277276
/* Decode until we run out of records */
278-
while ((startptr != InvalidXLogRecPtr && startptr < end_of_wal) ||
279-
(ctx->reader->EndRecPtr != InvalidXLogRecPtr && ctx->reader->EndRecPtr < end_of_wal))
277+
while (ctx->reader->EndRecPtr < end_of_wal)
280278
{
281279
XLogRecord *record;
282280
char *errm = NULL;
283281

284-
record = XLogReadRecord(ctx->reader, startptr, &errm);
282+
record = XLogReadRecord(ctx->reader, &errm);
285283
if (errm)
286284
elog(ERROR, "%s", errm);
287285

288-
/*
289-
* Now that we've set up the xlog reader state, subsequent calls
290-
* pass InvalidXLogRecPtr to say "continue from last record"
291-
*/
292-
startptr = InvalidXLogRecPtr;
293-
294286
/*
295287
* The {begin_txn,change,commit_txn}_wrapper callbacks above will
296288
* store the description into our tuplestore.

‎src/backend/replication/slotfuncs.c

Copy file name to clipboardExpand all lines: src/backend/replication/slotfuncs.c
+3-10Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -391,7 +391,6 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
391391
{
392392
LogicalDecodingContext *ctx;
393393
ResourceOwner old_resowner = CurrentResourceOwner;
394-
XLogRecPtr startlsn;
395394
XLogRecPtr retlsn;
396395

397396
PG_TRY();
@@ -411,7 +410,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
411410
* Start reading at the slot's restart_lsn, which we know to point to
412411
* a valid record.
413412
*/
414-
startlsn = MyReplicationSlot->data.restart_lsn;
413+
XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
415414

416415
/* Initialize our return value in case we don't do anything */
417416
retlsn = MyReplicationSlot->data.confirmed_flush;
@@ -420,10 +419,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
420419
InvalidateSystemCaches();
421420

422421
/* Decode at least one record, until we run out of records */
423-
while ((!XLogRecPtrIsInvalid(startlsn) &&
424-
startlsn < moveto) ||
425-
(!XLogRecPtrIsInvalid(ctx->reader->EndRecPtr) &&
426-
ctx->reader->EndRecPtr < moveto))
422+
while (ctx->reader->EndRecPtr < moveto)
427423
{
428424
char *errm = NULL;
429425
XLogRecord *record;
@@ -432,13 +428,10 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
432428
* Read records. No changes are generated in fast_forward mode,
433429
* but snapbuilder/slot statuses are updated properly.
434430
*/
435-
record = XLogReadRecord(ctx->reader, startlsn, &errm);
431+
record = XLogReadRecord(ctx->reader, &errm);
436432
if (errm)
437433
elog(ERROR, "%s", errm);
438434

439-
/* Read sequentially from now on */
440-
startlsn = InvalidXLogRecPtr;
441-
442435
/*
443436
* Process the record. Storage-level changes are ignored in
444437
* fast_forward mode, but other modules (such as snapbuilder)

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.