Movatterモバイル変換

[0]ホーム
URL:

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

 Sign up
Appearance settings

Commit38a9573

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

File tree

10 files changed

+102
-97
lines changed

10 files changed

+102
-97
lines changed

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

Lines 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

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -897,7 +897,7 @@ static void UpdateLastRemovedPtr(char *filename);
897897
staticvoidValidateXLOGDirectoryStructure(void);
898898
staticvoidCleanupBackupHistory(void);
899899
staticvoidUpdateMinRecoveryPoint(XLogRecPtrlsn,boolforce);
900-
staticXLogRecord*ReadRecord(XLogReaderState*xlogreader,XLogRecPtrRecPtr,
900+
staticXLogRecord*ReadRecord(XLogReaderState*xlogreader,
901901
intemode,boolfetching_ckpt);
902902
staticvoidCheckRecoveryConsistency(void);
903903
staticXLogRecord*ReadCheckpointRecord(XLogReaderState*xlogreader,
@@ -4246,17 +4246,17 @@ CleanupBackupHistory(void)
42464246
}
42474247

42484248
/*
4249-
* Attempt to readan XLOG record.
4249+
* Attempt to readthe 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
staticXLogRecord*
4259-
ReadRecord(XLogReaderState*xlogreader,XLogRecPtrRecPtr,intemode,
4259+
ReadRecord(XLogReaderState*xlogreader,intemode,
42604260
boolfetching_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
returnNULL;
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

Lines 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,XLogRecPtrRecPtr)
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-
*trytoread a record just after the last one previously read.
247+
*XLogBeginRead() or XLogFindNextRecord() must be called before the first call
248+
* toXLogReadRecord().
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,XLogRecPtrRecPtr,char**errormsg)
261+
XLogReadRecord(XLogReaderState*state,char**errormsg)
239262
{
263+
XLogRecPtrRecPtr;
240264
XLogRecord*record;
241265
XLogRecPtrtargetPagePtr;
242266
boolrandAccess;
@@ -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'reat a page boundary, no more records can fit on the current
296+
*page. Wemust skip over the page header, but we can't do that until
297+
*we'veread 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 pointershould already be
284-
*pointing to a validrecord starting position.
305+
* In this case,EndRecPtrshould 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,XLogRecPtrRecPtr)
908934
{
909-
XLogReaderStatesaved_state=*state;
910935
XLogRecPtrtmpRecPtr;
911936
XLogRecPtrfound=InvalidXLogRecPtr;
912937
XLogPageHeaderheader;
@@ -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-
gotoout;
1027+
XLogBeginRead(state,found);
1028+
returnfound;
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-
returnfound;
1035+
returnInvalidXLogRecPtr;
10151036
}
10161037

10171038
#endif/* FRONTEND */

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

Lines 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-
XLogRecPtrstartptr;
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

Lines 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
MemoryContextper_query_ctx;
128128
MemoryContextoldcontext;
129129
XLogRecPtrend_of_wal;
130-
XLogRecPtrstartptr;
131130
LogicalDecodingContext*ctx;
132131
ResourceOwnerold_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

Lines 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
ResourceOwnerold_resowner=CurrentResourceOwner;
394-
XLogRecPtrstartlsn;
395394
XLogRecPtrretlsn;
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)
[8]ページ先頭
©2009-2025 Movatter.jp