open-dsi
/
postgres
mirror of https://github.com/postgres/postgres
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use XLogRecPtrIsValid() in various places

Browse Source 
Now that commit 06edbed478 has introduced XLogRecPtrIsValid(), we can
use that instead of:

- XLogRecPtrIsInvalid()
- direct comparisons with InvalidXLogRecPtr
- direct comparisons with literal 0

This makes the code more consistent.

Author: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Discussion: https://postgr.es/m/aQB7EvGqrbZXrMlg@ip-10-97-1-34.eu-west-3.compute.internal
pull/246/head
Álvaro Herrera 1 month ago
parent 06edbed478
commit a2b02293bc
No known key found for this signature in database
GPG Key ID: 1C20ACB9D5C564AE
46 changed files with 239 additions and 238 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 4
      contrib/pg_walinspect/pg_walinspect.c
  2. 4
      src/backend/access/gist/gist.c
  3. 4
      src/backend/access/gist/gistget.c
  4. 2
      src/backend/access/gist/gistutil.c
  5. 4
      src/backend/access/heap/rewriteheap.c
  6. 2
      src/backend/access/heap/vacuumlazy.c
  7. 4
      src/backend/access/heap/visibilitymap.c
  8. 5
      src/backend/access/transam/clog.c
  9. 2
      src/backend/access/transam/slru.c
  10. 4
      src/backend/access/transam/timeline.c
  11. 6
      src/backend/access/transam/twophase.c
  12. 48
      src/backend/access/transam/xlog.c
  13. 4
      src/backend/access/transam/xlogbackup.c
  14. 4
      src/backend/access/transam/xlogfuncs.c
  15. 4
      src/backend/access/transam/xloginsert.c
  16. 8
      src/backend/access/transam/xlogreader.c
  17. 20
      src/backend/access/transam/xlogrecovery.c
  18. 8
      src/backend/access/transam/xlogutils.c
  19. 2
      src/backend/access/transam/xlogwait.c
  20. 4
      src/backend/backup/backup_manifest.c
  21. 2
      src/backend/backup/basebackup_incremental.c
  22. 8
      src/backend/backup/walsummary.c
  23. 4
      src/backend/catalog/pg_subscription.c
  24. 6
      src/backend/commands/subscriptioncmds.c
  25. 18
      src/backend/postmaster/walsummarizer.c
  26. 4
      src/backend/replication/logical/applyparallelworker.c
  27. 4
      src/backend/replication/logical/launcher.c
  28. 32
      src/backend/replication/logical/logical.c
  29. 6
      src/backend/replication/logical/logicalfuncs.c
  30. 18
      src/backend/replication/logical/origin.c
  31. 18
      src/backend/replication/logical/proto.c
  32. 38
      src/backend/replication/logical/reorderbuffer.c
  33. 6
      src/backend/replication/logical/slotsync.c
  34. 14
      src/backend/replication/logical/snapbuild.c
  35. 16
      src/backend/replication/logical/worker.c
  36. 28
      src/backend/replication/slot.c
  37. 22
      src/backend/replication/slotfuncs.c
  38. 10
      src/backend/replication/syncrep.c
  39. 8
      src/backend/replication/walreceiver.c
  40. 2
      src/backend/replication/walreceiverfuncs.c
  41. 30
      src/backend/replication/walsender.c
  42. 2
      src/backend/storage/buffer/bufmgr.c
  43. 8
      src/bin/pg_basebackup/pg_receivewal.c
  44. 12
      src/bin/pg_basebackup/pg_recvlogical.c
  45. 4
      src/bin/pg_rewind/pg_rewind.c
  46. 14
      src/bin/pg_waldump/pg_waldump.c

4
contrib/pg_walinspect/pg_walinspect.c
Unescape View File

@ -83,7 +83,7 @@ GetCurrentLSN(void)
else
curr_lsn = GetXLogReplayRecPtr(NULL);
Assert(!XLogRecPtrIsInvalid(curr_lsn));
Assert(XLogRecPtrIsValid(curr_lsn));
return curr_lsn;
}
@ -127,7 +127,7 @@ InitXLogReaderState(XLogRecPtr lsn)
/* first find a valid recptr to start from */
first_valid_record = XLogFindNextRecord(xlogreader, lsn);
if (XLogRecPtrIsInvalid(first_valid_record))
if (!XLogRecPtrIsValid(first_valid_record))
ereport(ERROR,
errmsg("could not find a valid record after %X/%08X",
LSN_FORMAT_ARGS(lsn)));

4
src/backend/access/gist/gist.c
Unescape View File

@ -682,7 +682,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace,
state.stack = stack = stack->parent;
}
if (XLogRecPtrIsInvalid(stack->lsn))
if (!XLogRecPtrIsValid(stack->lsn))
stack->buffer = ReadBuffer(state.r, stack->blkno);
/*
@ -698,7 +698,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace,
stack->page = BufferGetPage(stack->buffer);
stack->lsn = xlocked ?
PageGetLSN(stack->page) : BufferGetLSNAtomic(stack->buffer);
Assert(!RelationNeedsWAL(state.r) || !XLogRecPtrIsInvalid(stack->lsn));
Assert(!RelationNeedsWAL(state.r) || XLogRecPtrIsValid(stack->lsn));
/*
* If this page was split but the downlink was never inserted to the

4
src/backend/access/gist/gistget.c
Unescape View File

@ -46,7 +46,7 @@ gistkillitems(IndexScanDesc scan)
bool killedsomething = false;
Assert(so->curBlkno != InvalidBlockNumber);
Assert(!XLogRecPtrIsInvalid(so->curPageLSN));
Assert(XLogRecPtrIsValid(so->curPageLSN));
Assert(so->killedItems != NULL);
buffer = ReadBuffer(scan->indexRelation, so->curBlkno);
@ -353,7 +353,7 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem,
* parentlsn < nsn), or if the system crashed after a page split but
* before the downlink was inserted into the parent.
*/
if (!XLogRecPtrIsInvalid(pageItem->data.parentlsn) &&
if (XLogRecPtrIsValid(pageItem->data.parentlsn) &&
(GistFollowRight(page) ||
pageItem->data.parentlsn < GistPageGetNSN(page)) &&
opaque->rightlink != InvalidBlockNumber /* sanity check */ )

2
src/backend/access/gist/gistutil.c
Unescape View File

@ -1040,7 +1040,7 @@ gistGetFakeLSN(Relation rel)
Assert(!RelationNeedsWAL(rel));
/* No need for an actual record if we already have a distinct LSN */
if (!XLogRecPtrIsInvalid(lastlsn) && lastlsn == currlsn)
if (XLogRecPtrIsValid(lastlsn) && lastlsn == currlsn)
currlsn = gistXLogAssignLSN();
lastlsn = currlsn;

4
src/backend/access/heap/rewriteheap.c
Unescape View File

@ -1169,7 +1169,7 @@ CheckPointLogicalRewriteHeap(void)
cutoff = ReplicationSlotsComputeLogicalRestartLSN();
/* don't start earlier than the restart lsn */
if (cutoff != InvalidXLogRecPtr && redo < cutoff)
if (XLogRecPtrIsValid(cutoff) && redo < cutoff)
cutoff = redo;
mappings_dir = AllocateDir(PG_LOGICAL_MAPPINGS_DIR);
@ -1204,7 +1204,7 @@ CheckPointLogicalRewriteHeap(void)
lsn = ((uint64) hi) << 32 | lo;
if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
{
elog(DEBUG1, "removing logical rewrite file \"%s\"", path);
if (unlink(path) < 0)

2
src/backend/access/heap/vacuumlazy.c
Unescape View File

@ -1901,7 +1901,7 @@ lazy_scan_new_or_empty(LVRelState *vacrel, Buffer buf, BlockNumber blkno,
* WAL-logged, and if not, do that now.
*/
if (RelationNeedsWAL(vacrel->rel) &&
PageGetLSN(page) == InvalidXLogRecPtr)
!XLogRecPtrIsValid(PageGetLSN(page)))
log_newpage_buffer(buf, true);
PageSetAllVisible(page);

4
src/backend/access/heap/visibilitymap.c
Unescape View File

@ -260,7 +260,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
flags, RelationGetRelationName(rel), heapBlk);
#endif
Assert(InRecovery || XLogRecPtrIsInvalid(recptr));
Assert(InRecovery || !XLogRecPtrIsValid(recptr));
Assert(InRecovery || PageIsAllVisible(BufferGetPage(heapBuf)));
Assert((flags & VISIBILITYMAP_VALID_BITS) == flags);
@ -292,7 +292,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
if (RelationNeedsWAL(rel))
{
if (XLogRecPtrIsInvalid(recptr))
if (!XLogRecPtrIsValid(recptr))
{
Assert(!InRecovery);
recptr = log_heap_visible(rel, heapBuf, vmBuf, cutoff_xid, flags);

5
src/backend/access/transam/clog.c
Unescape View File

@ -381,7 +381,8 @@ TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
* write-busy, since we don't care if the update reaches disk sooner than
* we think.
*/
slotno = SimpleLruReadPage(XactCtl, pageno, XLogRecPtrIsInvalid(lsn), xid);
slotno = SimpleLruReadPage(XactCtl, pageno, !XLogRecPtrIsValid(lsn),
xid);
/*
* Set the main transaction id, if any.
@ -705,7 +706,7 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
* recovery. After recovery completes the next clog change will set the
* LSN correctly.
*/
if (!XLogRecPtrIsInvalid(lsn))
if (XLogRecPtrIsValid(lsn))
{
int lsnindex = GetLSNIndex(slotno, xid);

2
src/backend/access/transam/slru.c
Unescape View File

@ -937,7 +937,7 @@ SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
max_lsn = this_lsn;
}
if (!XLogRecPtrIsInvalid(max_lsn))
if (XLogRecPtrIsValid(max_lsn))
{
/*
* As noted above, elog(ERROR) is not acceptable here, so if

4
src/backend/access/transam/timeline.c
Unescape View File

@ -549,8 +549,8 @@ tliOfPointInHistory(XLogRecPtr ptr, List *history)
{
TimeLineHistoryEntry *tle = (TimeLineHistoryEntry *) lfirst(cell);
if ((XLogRecPtrIsInvalid(tle->begin) || tle->begin <= ptr) &&
(XLogRecPtrIsInvalid(tle->end) || ptr < tle->end))
if ((!XLogRecPtrIsValid(tle->begin) || tle->begin <= ptr) &&
(!XLogRecPtrIsValid(tle->end) || ptr < tle->end))
{
/* found it */
return tle->tli;

6
src/backend/access/transam/twophase.c
Unescape View File

@ -2198,7 +2198,7 @@ ProcessTwoPhaseBuffer(FullTransactionId fxid,
Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
if (!fromdisk)
Assert(prepare_start_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(prepare_start_lsn));
/* Already processed? */
if (TransactionIdDidCommit(XidFromFullTransactionId(fxid)) ||
@ -2547,7 +2547,7 @@ PrepareRedoAdd(FullTransactionId fxid, char *buf,
* the record is added to TwoPhaseState and it should have no
* corresponding file in pg_twophase.
*/
if (!XLogRecPtrIsInvalid(start_lsn))
if (XLogRecPtrIsValid(start_lsn))
{
char path[MAXPGPATH];
@ -2587,7 +2587,7 @@ PrepareRedoAdd(FullTransactionId fxid, char *buf,
gxact->owner = hdr->owner;
gxact->locking_backend = INVALID_PROC_NUMBER;
gxact->valid = false;
gxact->ondisk = XLogRecPtrIsInvalid(start_lsn);
gxact->ondisk = !XLogRecPtrIsValid(start_lsn);
gxact->inredo = true; /* yes, added in redo */
strcpy(gxact->gid, gid);

48
src/backend/access/transam/xlog.c
Unescape View File

@ -849,7 +849,7 @@ XLogInsertRecord(XLogRecData *rdata,
if (doPageWrites &&
(!prevDoPageWrites ||
(fpw_lsn != InvalidXLogRecPtr && fpw_lsn <= RedoRecPtr)))
(XLogRecPtrIsValid(fpw_lsn) && fpw_lsn <= RedoRecPtr)))
{
/*
* Oops, some buffer now needs to be backed up that the caller
@ -883,7 +883,7 @@ XLogInsertRecord(XLogRecData *rdata,
* Those checks are only needed for records that can contain buffer
* references, and an XLOG_SWITCH record never does.
*/
Assert(fpw_lsn == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(fpw_lsn));
WALInsertLockAcquireExclusive();
inserted = ReserveXLogSwitch(&StartPos, &EndPos, &rechdr->xl_prev);
}
@ -898,7 +898,7 @@ XLogInsertRecord(XLogRecData *rdata,
* not check RedoRecPtr before inserting the record; we just need to
* update it afterwards.
*/
Assert(fpw_lsn == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(fpw_lsn));
WALInsertLockAcquireExclusive();
ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
&rechdr->xl_prev);
@ -1603,7 +1603,7 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto)
*/
} while (insertingat < upto);
if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto)
if (XLogRecPtrIsValid(insertingat) && insertingat < finishedUpto)
finishedUpto = insertingat;
}
@ -1762,7 +1762,7 @@ WALReadFromBuffers(char *dstbuf, XLogRecPtr startptr, Size count,
if (RecoveryInProgress() || tli != GetWALInsertionTimeLine())
return 0;
Assert(!XLogRecPtrIsInvalid(startptr));
Assert(XLogRecPtrIsValid(startptr));
/*
* Caller should ensure that the requested data has been inserted into WAL
@ -2717,7 +2717,7 @@ UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force)
* available is replayed in this case. This also saves from extra locks
* taken on the control file from the startup process.
*/
if (XLogRecPtrIsInvalid(LocalMinRecoveryPoint) && InRecovery)
if (!XLogRecPtrIsValid(LocalMinRecoveryPoint) && InRecovery)
{
updateMinRecoveryPoint = false;
return;
@ -2729,7 +2729,7 @@ UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force)
LocalMinRecoveryPoint = ControlFile->minRecoveryPoint;
LocalMinRecoveryPointTLI = ControlFile->minRecoveryPointTLI;
if (XLogRecPtrIsInvalid(LocalMinRecoveryPoint))
if (!XLogRecPtrIsValid(LocalMinRecoveryPoint))
updateMinRecoveryPoint = false;
else if (force || LocalMinRecoveryPoint < lsn)
{
@ -3149,7 +3149,7 @@ XLogNeedsFlush(XLogRecPtr record)
* which cannot update its local copy of minRecoveryPoint as long as
* it has not replayed all WAL available when doing crash recovery.
*/
if (XLogRecPtrIsInvalid(LocalMinRecoveryPoint) && InRecovery)
if (!XLogRecPtrIsValid(LocalMinRecoveryPoint) && InRecovery)
{
updateMinRecoveryPoint = false;
return false;
@ -3170,7 +3170,7 @@ XLogNeedsFlush(XLogRecPtr record)
* process doing crash recovery, which should not update the control
* file value if crash recovery is still running.
*/
if (XLogRecPtrIsInvalid(LocalMinRecoveryPoint))
if (!XLogRecPtrIsValid(LocalMinRecoveryPoint))
updateMinRecoveryPoint = false;
/* check again */
@ -5935,7 +5935,7 @@ StartupXLOG(void)
*/
if (InRecovery &&
(EndOfLog < LocalMinRecoveryPoint ||
!XLogRecPtrIsInvalid(ControlFile->backupStartPoint)))
XLogRecPtrIsValid(ControlFile->backupStartPoint)))
{
/*
* Ran off end of WAL before reaching end-of-backup WAL record, or
@ -5945,7 +5945,7 @@ StartupXLOG(void)
*/
if (ArchiveRecoveryRequested || ControlFile->backupEndRequired)
{
if (!XLogRecPtrIsInvalid(ControlFile->backupStartPoint) || ControlFile->backupEndRequired)
if (XLogRecPtrIsValid(ControlFile->backupStartPoint) || ControlFile->backupEndRequired)
ereport(FATAL,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("WAL ends before end of online backup"),
@ -6047,7 +6047,7 @@ StartupXLOG(void)
* (It's critical to first write an OVERWRITE_CONTRECORD message, which
* we'll do as soon as we're open for writing new WAL.)
*/
if (!XLogRecPtrIsInvalid(missingContrecPtr))
if (XLogRecPtrIsValid(missingContrecPtr))
{
/*
* We should only have a missingContrecPtr if we're not switching to a
@ -6057,7 +6057,7 @@ StartupXLOG(void)
* disregard.
*/
Assert(newTLI == endOfRecoveryInfo->lastRecTLI);
Assert(!XLogRecPtrIsInvalid(abortedRecPtr));
Assert(XLogRecPtrIsValid(abortedRecPtr));
EndOfLog = missingContrecPtr;
}
@ -6161,9 +6161,9 @@ StartupXLOG(void)
LocalSetXLogInsertAllowed();
/* If necessary, write overwrite-contrecord before doing anything else */
if (!XLogRecPtrIsInvalid(abortedRecPtr))
if (XLogRecPtrIsValid(abortedRecPtr))
{
Assert(!XLogRecPtrIsInvalid(missingContrecPtr));
Assert(XLogRecPtrIsValid(missingContrecPtr));
CreateOverwriteContrecordRecord(abortedRecPtr, missingContrecPtr, newTLI);
}
@ -7363,7 +7363,7 @@ CreateCheckPoint(int flags)
* Update the average distance between checkpoints if the prior checkpoint
* exists.
*/
if (PriorRedoPtr != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(PriorRedoPtr))
UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr);
INJECTION_POINT("checkpoint-before-old-wal-removal", NULL);
@ -7693,7 +7693,7 @@ CreateRestartPoint(int flags)
* restartpoint. It's assumed that flushing the buffers will do that as a
* side-effect.
*/
if (XLogRecPtrIsInvalid(lastCheckPointRecPtr) ||
if (!XLogRecPtrIsValid(lastCheckPointRecPtr) ||
lastCheckPoint.redo <= ControlFile->checkPointCopy.redo)
{
ereport(DEBUG2,
@ -7811,7 +7811,7 @@ CreateRestartPoint(int flags)
* Update the average distance between checkpoints/restartpoints if the
* prior checkpoint exists.
*/
if (PriorRedoPtr != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(PriorRedoPtr))
UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr);
/*
@ -7936,7 +7936,7 @@ GetWALAvailability(XLogRecPtr targetLSN)
/*
* slot does not reserve WAL. Either deactivated, or has never been active
*/
if (XLogRecPtrIsInvalid(targetLSN))
if (!XLogRecPtrIsValid(targetLSN))
return WALAVAIL_INVALID_LSN;
/*
@ -8018,7 +8018,7 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
/* Calculate how many segments are kept by slots. */
keep = XLogGetReplicationSlotMinimumLSN();
if (keep != InvalidXLogRecPtr && keep < recptr)
if (XLogRecPtrIsValid(keep) && keep < recptr)
{
XLByteToSeg(keep, segno, wal_segment_size);
@ -8045,7 +8045,7 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
* summarized.
*/
keep = GetOldestUnsummarizedLSN(NULL, NULL);
if (keep != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(keep))
{
XLogSegNo unsummarized_segno;
@ -8352,8 +8352,8 @@ xlog_redo(XLogReaderState *record)
* never arrive.
*/
if (ArchiveRecoveryRequested &&
!XLogRecPtrIsInvalid(ControlFile->backupStartPoint) &&
XLogRecPtrIsInvalid(ControlFile->backupEndPoint))
XLogRecPtrIsValid(ControlFile->backupStartPoint) &&
!XLogRecPtrIsValid(ControlFile->backupEndPoint))
ereport(PANIC,
(errmsg("online backup was canceled, recovery cannot continue")));
@ -8603,7 +8603,7 @@ xlog_redo(XLogReaderState *record)
LocalMinRecoveryPoint = ControlFile->minRecoveryPoint;
LocalMinRecoveryPointTLI = ControlFile->minRecoveryPointTLI;
}
if (LocalMinRecoveryPoint != InvalidXLogRecPtr && LocalMinRecoveryPoint < lsn)
if (XLogRecPtrIsValid(LocalMinRecoveryPoint) && LocalMinRecoveryPoint < lsn)
{
TimeLineID replayTLI;

4
src/backend/access/transam/xlogbackup.c
Unescape View File

@ -79,8 +79,8 @@ build_backup_content(BackupState *state, bool ishistoryfile)
}
/* either both istartpoint and istarttli should be set, or neither */
Assert(XLogRecPtrIsInvalid(state->istartpoint) == (state->istarttli == 0));
if (!XLogRecPtrIsInvalid(state->istartpoint))
Assert(XLogRecPtrIsValid(state->istartpoint) == (state->istarttli != 0));
if (XLogRecPtrIsValid(state->istartpoint))
{
appendStringInfo(&result, "INCREMENTAL FROM LSN: %X/%08X\n",
LSN_FORMAT_ARGS(state->istartpoint));

4
src/backend/access/transam/xlogfuncs.c
Unescape View File

@ -341,7 +341,7 @@ pg_last_wal_receive_lsn(PG_FUNCTION_ARGS)
recptr = GetWalRcvFlushRecPtr(NULL, NULL);
if (recptr == 0)
if (!XLogRecPtrIsValid(recptr))
PG_RETURN_NULL();
PG_RETURN_LSN(recptr);
@ -360,7 +360,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
recptr = GetXLogReplayRecPtr(NULL);
if (recptr == 0)
if (!XLogRecPtrIsValid(recptr))
PG_RETURN_NULL();
PG_RETURN_LSN(recptr);

4
src/backend/access/transam/xloginsert.c
Unescape View File

@ -528,7 +528,7 @@ XLogInsert(RmgrId rmid, uint8 info)
EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi,
fpi_bytes, topxid_included);
} while (EndPos == InvalidXLogRecPtr);
} while (!XLogRecPtrIsValid(EndPos));
XLogResetInsertion();
@ -639,7 +639,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
needs_backup = (page_lsn <= RedoRecPtr);
if (!needs_backup)
{
if (*fpw_lsn == InvalidXLogRecPtr || page_lsn < *fpw_lsn)
if (!XLogRecPtrIsValid(*fpw_lsn) || page_lsn < *fpw_lsn)
*fpw_lsn = page_lsn;
}
}

8
src/backend/access/transam/xlogreader.c
Unescape View File

@ -231,7 +231,7 @@ WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
void
XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
{
Assert(!XLogRecPtrIsInvalid(RecPtr));
Assert(XLogRecPtrIsValid(RecPtr));
ResetDecoder(state);
@ -343,7 +343,7 @@ XLogNextRecord(XLogReaderState *state, char **errormsg)
* XLogBeginRead() or XLogNextRecord(), and is the location of the
* error.
*/
Assert(!XLogRecPtrIsInvalid(state->EndRecPtr));
Assert(XLogRecPtrIsValid(state->EndRecPtr));
return NULL;
}
@ -558,7 +558,7 @@ XLogDecodeNextRecord(XLogReaderState *state, bool nonblocking)
RecPtr = state->NextRecPtr;
if (state->DecodeRecPtr != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(state->DecodeRecPtr))
{
/* read the record after the one we just read */
@ -1398,7 +1398,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
XLogPageHeader header;
char *errormsg;
Assert(!XLogRecPtrIsInvalid(RecPtr));
Assert(XLogRecPtrIsValid(RecPtr));
/* Make sure ReadPageInternal() can't return XLREAD_WOULDBLOCK. */
state->nonblocking = false;

20
src/backend/access/transam/xlogrecovery.c
Unescape View File

@ -758,9 +758,9 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
* end-of-backup record), and we can enter archive recovery directly.
*/
if (ArchiveRecoveryRequested &&
(ControlFile->minRecoveryPoint != InvalidXLogRecPtr ||
(XLogRecPtrIsValid(ControlFile->minRecoveryPoint) ||
ControlFile->backupEndRequired ||
ControlFile->backupEndPoint != InvalidXLogRecPtr ||
XLogRecPtrIsValid(ControlFile->backupEndPoint) ||
ControlFile->state == DB_SHUTDOWNED))
{
InArchiveRecovery = true;
@ -773,7 +773,7 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
* emit a log message when we continue initializing from a base
* backup.
*/
if (!XLogRecPtrIsInvalid(ControlFile->backupStartPoint))
if (XLogRecPtrIsValid(ControlFile->backupStartPoint))
ereport(LOG,
errmsg("restarting backup recovery with redo LSN %X/%08X",
LSN_FORMAT_ARGS(ControlFile->backupStartPoint)));
@ -868,7 +868,7 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
* The min recovery point should be part of the requested timeline's
* history, too.
*/
if (!XLogRecPtrIsInvalid(ControlFile->minRecoveryPoint) &&
if (XLogRecPtrIsValid(ControlFile->minRecoveryPoint) &&
tliOfPointInHistory(ControlFile->minRecoveryPoint - 1, expectedTLEs) !=
ControlFile->minRecoveryPointTLI)
ereport(FATAL,
@ -2204,7 +2204,7 @@ CheckRecoveryConsistency(void)
* During crash recovery, we don't reach a consistent state until we've
* replayed all the WAL.
*/
if (XLogRecPtrIsInvalid(minRecoveryPoint))
if (!XLogRecPtrIsValid(minRecoveryPoint))
return;
Assert(InArchiveRecovery);
@ -2219,7 +2219,7 @@ CheckRecoveryConsistency(void)
/*
* Have we reached the point where our base backup was completed?
*/
if (!XLogRecPtrIsInvalid(backupEndPoint) &&
if (XLogRecPtrIsValid(backupEndPoint) &&
backupEndPoint <= lastReplayedEndRecPtr)
{
XLogRecPtr saveBackupStartPoint = backupStartPoint;
@ -2425,7 +2425,7 @@ checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI, TimeLineID prevTLI,
* branched before the timeline the min recovery point is on, and you
* attempt to do PITR to the new timeline.
*/
if (!XLogRecPtrIsInvalid(minRecoveryPoint) &&
if (XLogRecPtrIsValid(minRecoveryPoint) &&
lsn < minRecoveryPoint &&
newTLI > minRecoveryPointTLI)
ereport(PANIC,
@ -3164,7 +3164,7 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
/* Pass through parameters to XLogPageRead */
private->fetching_ckpt = fetching_ckpt;
private->emode = emode;
private->randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr);
private->randAccess = !XLogRecPtrIsValid(xlogreader->ReadRecPtr);
private->replayTLI = replayTLI;
/* This is the first attempt to read this page. */
@ -3190,7 +3190,7 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
* overwrite contrecord in the wrong place, breaking everything.
*/
if (!ArchiveRecoveryRequested &&
!XLogRecPtrIsInvalid(xlogreader->abortedRecPtr))
XLogRecPtrIsValid(xlogreader->abortedRecPtr))
{
abortedRecPtr = xlogreader->abortedRecPtr;
missingContrecPtr = xlogreader->missingContrecPtr;
@ -4370,7 +4370,7 @@ XLogFileReadAnyTLI(XLogSegNo segno, XLogSource source)
* Skip scanning the timeline ID that the logfile segment to read
* doesn't belong to
*/
if (hent->begin != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(hent->begin))
{
XLogSegNo beginseg = 0;

8
src/backend/access/transam/xlogutils.c
Unescape View File

@ -710,7 +710,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage,
const XLogRecPtr lastReadPage = (state->seg.ws_segno *
state->segcxt.ws_segsize + state->segoff);
Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0);
Assert(XLogRecPtrIsValid(wantPage) && wantPage % XLOG_BLCKSZ == 0);
Assert(wantLength <= XLOG_BLCKSZ);
Assert(state->readLen == 0 || state->readLen <= XLOG_BLCKSZ);
Assert(currTLI != 0);
@ -741,7 +741,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage,
*/
if (state->currTLI == currTLI && wantPage >= lastReadPage)
{
Assert(state->currTLIValidUntil == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(state->currTLIValidUntil));
return;
}
@ -750,7 +750,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage,
* timeline and the timeline we're reading from is valid until the end of
* the current segment we can just keep reading.
*/
if (state->currTLIValidUntil != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(state->currTLIValidUntil) &&
state->currTLI != currTLI &&
state->currTLI != 0 &&
((wantPage + wantLength) / state->segcxt.ws_segsize) <
@ -790,7 +790,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage,
state->currTLIValidUntil = tliSwitchPoint(state->currTLI, timelineHistory,
&state->nextTLI);
Assert(state->currTLIValidUntil == InvalidXLogRecPtr ||
Assert(!XLogRecPtrIsValid(state->currTLIValidUntil) ||
wantPage + wantLength < state->currTLIValidUntil);
list_free_deep(timelineHistory);

2
src/backend/access/transam/xlogwait.c
Unescape View File

@ -230,7 +230,7 @@ wakeupWaiters(WaitLSNType lsnType, XLogRecPtr currentLSN)
/* Get procInfo using appropriate heap node */
procInfo = pairingheap_container(WaitLSNProcInfo, heapNode[i], node);
if (!XLogRecPtrIsInvalid(currentLSN) && procInfo->waitLSN > currentLSN)
if (XLogRecPtrIsValid(currentLSN) && procInfo->waitLSN > currentLSN)
break;
Assert(numWakeUpProcs < WAKEUP_PROC_STATIC_ARRAY_SIZE);

4
src/backend/backup/backup_manifest.c
Unescape View File

@ -242,7 +242,7 @@ AddWALInfoToBackupManifest(backup_manifest_info *manifest, XLogRecPtr startptr,
* entry->end is InvalidXLogRecPtr, it means that the timeline has not
* yet ended.)
*/
if (!XLogRecPtrIsInvalid(entry->end) && entry->end < startptr)
if (XLogRecPtrIsValid(entry->end) && entry->end < startptr)
continue;
/*
@ -274,7 +274,7 @@ AddWALInfoToBackupManifest(backup_manifest_info *manifest, XLogRecPtr startptr,
* better have arrived at the expected starting TLI. If not,
* something's gone horribly wrong.
*/
if (XLogRecPtrIsInvalid(entry->begin))
if (!XLogRecPtrIsValid(entry->begin))
ereport(ERROR,
errmsg("expected start timeline %u but found timeline %u",
starttli, entry->tli));

2
src/backend/backup/basebackup_incremental.c
Unescape View File

@ -519,7 +519,7 @@ PrepareForIncrementalBackup(IncrementalBackupInfo *ib,
if (!WalSummariesAreComplete(tli_wslist, tli_start_lsn, tli_end_lsn,
&tli_missing_lsn))
{
if (XLogRecPtrIsInvalid(tli_missing_lsn))
if (!XLogRecPtrIsValid(tli_missing_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("WAL summaries are required on timeline %u from %X/%08X to %X/%08X, but no summaries for that timeline and LSN range exist",

8
src/backend/backup/walsummary.c
Unescape View File

@ -67,9 +67,9 @@ GetWalSummaries(TimeLineID tli, XLogRecPtr start_lsn, XLogRecPtr end_lsn)
/* Skip if it doesn't match the filter criteria. */
if (tli != 0 && tli != file_tli)
continue;
if (!XLogRecPtrIsInvalid(start_lsn) && start_lsn >= file_end_lsn)
if (XLogRecPtrIsValid(start_lsn) && start_lsn >= file_end_lsn)
continue;
if (!XLogRecPtrIsInvalid(end_lsn) && end_lsn <= file_start_lsn)
if (XLogRecPtrIsValid(end_lsn) && end_lsn <= file_start_lsn)
continue;
/* Add it to the list. */
@ -111,9 +111,9 @@ FilterWalSummaries(List *wslist, TimeLineID tli,
/* Skip if it doesn't match the filter criteria. */
if (tli != 0 && tli != ws->tli)
continue;
if (!XLogRecPtrIsInvalid(start_lsn) && start_lsn > ws->end_lsn)
if (XLogRecPtrIsValid(start_lsn) && start_lsn > ws->end_lsn)
continue;
if (!XLogRecPtrIsInvalid(end_lsn) && end_lsn < ws->start_lsn)
if (XLogRecPtrIsValid(end_lsn) && end_lsn < ws->start_lsn)
continue;
/* Add it to the result list. */

4
src/backend/catalog/pg_subscription.c
Unescape View File

@ -293,7 +293,7 @@ AddSubscriptionRelState(Oid subid, Oid relid, char state,
values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
if (sublsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(sublsn))
values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
else
nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
@ -366,7 +366,7 @@ UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
if (sublsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(sublsn))
values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
else
nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

6
src/backend/commands/subscriptioncmds.c
Unescape View File

@ -393,7 +393,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
CStringGetDatum(lsn_str)));
if (XLogRecPtrIsInvalid(lsn))
if (!XLogRecPtrIsValid(lsn))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("invalid WAL location (LSN): %s", lsn_str)));
@ -1895,7 +1895,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* If the user sets subskiplsn, we do a sanity check to make
* sure that the specified LSN is a probable value.
*/
if (!XLogRecPtrIsInvalid(opts.lsn))
if (XLogRecPtrIsValid(opts.lsn))
{
RepOriginId originid;
char originname[NAMEDATALEN];
@ -1907,7 +1907,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
remote_lsn = replorigin_get_progress(originid, false);
/* Check the given LSN is at least a future LSN */
if (!XLogRecPtrIsInvalid(remote_lsn) && opts.lsn < remote_lsn)
if (XLogRecPtrIsValid(remote_lsn) && opts.lsn < remote_lsn)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("skip WAL location (LSN %X/%08X) must be greater than origin LSN %X/%08X",

18
src/backend/postmaster/walsummarizer.c
Unescape View File

@ -342,7 +342,7 @@ WalSummarizerMain(const void *startup_data, size_t startup_data_len)
* If we discover that WAL summarization is not enabled, just exit.
*/
current_lsn = GetOldestUnsummarizedLSN(&current_tli, &exact);
if (XLogRecPtrIsInvalid(current_lsn))
if (!XLogRecPtrIsValid(current_lsn))
proc_exit(0);
/*
@ -379,7 +379,7 @@ WalSummarizerMain(const void *startup_data, size_t startup_data_len)
* only have to do this once per timeline switch, we probably wouldn't
* save any significant amount of work in practice.
*/
if (current_tli != latest_tli && XLogRecPtrIsInvalid(switch_lsn))
if (current_tli != latest_tli && !XLogRecPtrIsValid(switch_lsn))
{
List *tles = readTimeLineHistory(latest_tli);
@ -394,7 +394,7 @@ WalSummarizerMain(const void *startup_data, size_t startup_data_len)
* on this timeline. Switch to the next timeline and go around again,
* backing up to the exact switch point if we passed it.
*/
if (!XLogRecPtrIsInvalid(switch_lsn) && current_lsn >= switch_lsn)
if (XLogRecPtrIsValid(switch_lsn) && current_lsn >= switch_lsn)
{
/* Restart summarization from switch point. */
current_tli = switch_tli;
@ -419,7 +419,7 @@ WalSummarizerMain(const void *startup_data, size_t startup_data_len)
end_of_summary_lsn = SummarizeWAL(current_tli,
current_lsn, exact,
switch_lsn, latest_lsn);
Assert(!XLogRecPtrIsInvalid(end_of_summary_lsn));
Assert(XLogRecPtrIsValid(end_of_summary_lsn));
Assert(end_of_summary_lsn >= current_lsn);
/*
@ -923,7 +923,7 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact,
private_data = (SummarizerReadLocalXLogPrivate *)
palloc0(sizeof(SummarizerReadLocalXLogPrivate));
private_data->tli = tli;
private_data->historic = !XLogRecPtrIsInvalid(switch_lsn);
private_data->historic = XLogRecPtrIsValid(switch_lsn);
private_data->read_upto = maximum_lsn;
/* Create xlogreader. */
@ -971,7 +971,7 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact,
else
{
summary_start_lsn = XLogFindNextRecord(xlogreader, start_lsn);
if (XLogRecPtrIsInvalid(summary_start_lsn))
if (!XLogRecPtrIsValid(summary_start_lsn))
{
/*
* If we hit end-of-WAL while trying to find the next valid
@ -1058,7 +1058,7 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact,
/* We shouldn't go backward. */
Assert(summary_start_lsn <= xlogreader->EndRecPtr);
if (!XLogRecPtrIsInvalid(switch_lsn) &&
if (XLogRecPtrIsValid(switch_lsn) &&
xlogreader->ReadRecPtr >= switch_lsn)
{
/*
@ -1180,7 +1180,7 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact,
* If we have a switch LSN and have reached it, stop before reading
* the next record.
*/
if (!XLogRecPtrIsInvalid(switch_lsn) &&
if (XLogRecPtrIsValid(switch_lsn) &&
xlogreader->EndRecPtr >= switch_lsn)
break;
}
@ -1723,7 +1723,7 @@ MaybeRemoveOldWalSummaries(void)
* If the WAL doesn't exist any more, we can remove it if the file
* modification time is old enough.
*/
if (XLogRecPtrIsInvalid(oldest_lsn) || ws->end_lsn <= oldest_lsn)
if (!XLogRecPtrIsValid(oldest_lsn) || ws->end_lsn <= oldest_lsn)
RemoveWalSummaryIfOlderThan(ws, cutoff_time);
/*

4
src/backend/replication/logical/applyparallelworker.c
Unescape View File

@ -299,7 +299,7 @@ pa_can_start(void)
* STREAM START message, and it doesn't seem worth sending the extra eight
* bytes with the STREAM START to enable parallelism for this case.
*/
if (!XLogRecPtrIsInvalid(MySubscription->skiplsn))
if (XLogRecPtrIsValid(MySubscription->skiplsn))
return false;
/*
@ -1640,7 +1640,7 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
*/
pa_wait_for_xact_finish(winfo);
if (!XLogRecPtrIsInvalid(remote_lsn))
if (XLogRecPtrIsValid(remote_lsn))
store_flush_position(remote_lsn, winfo->shared->last_commit_end);
pa_free_worker(winfo);

4
src/backend/replication/logical/launcher.c
Unescape View File

@ -1661,7 +1661,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
else
nulls[3] = true;
if (XLogRecPtrIsInvalid(worker.last_lsn))
if (!XLogRecPtrIsValid(worker.last_lsn))
nulls[4] = true;
else
values[4] = LSNGetDatum(worker.last_lsn);
@ -1673,7 +1673,7 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
nulls[6] = true;
else
values[6] = TimestampTzGetDatum(worker.last_recv_time);
if (XLogRecPtrIsInvalid(worker.reply_lsn))
if (!XLogRecPtrIsValid(worker.reply_lsn))
nulls[7] = true;
else
values[7] = LSNGetDatum(worker.reply_lsn);

32
src/backend/replication/logical/logical.c
Unescape View File

@ -388,7 +388,7 @@ CreateInitDecodingContext(const char *plugin,
slot->data.plugin = plugin_name;
SpinLockRelease(&slot->mutex);
if (XLogRecPtrIsInvalid(restart_lsn))
if (!XLogRecPtrIsValid(restart_lsn))
ReplicationSlotReserveWal();
else
{
@ -546,9 +546,9 @@ CreateDecodingContext(XLogRecPtr start_lsn,
/* slot must be valid to allow decoding */
Assert(slot->data.invalidated == RS_INVAL_NONE);
Assert(slot->data.restart_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(slot->data.restart_lsn));
if (start_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(start_lsn))
{
/* continue from last position */
start_lsn = slot->data.confirmed_flush;
@ -757,7 +757,7 @@ output_plugin_error_callback(void *arg)
LogicalErrorCallbackState *state = (LogicalErrorCallbackState *) arg;
/* not all callbacks have an associated LSN */
if (state->report_location != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(state->report_location))
errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%08X",
NameStr(state->ctx->slot->data.name),
NameStr(state->ctx->slot->data.plugin),
@ -1711,7 +1711,7 @@ LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
* Only increase if the previous values have been applied, otherwise we
* might never end up updating if the receiver acks too slowly.
*/
else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
else if (!XLogRecPtrIsValid(slot->candidate_xmin_lsn))
{
slot->candidate_catalog_xmin = xmin;
slot->candidate_xmin_lsn = current_lsn;
@ -1749,8 +1749,8 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
slot = MyReplicationSlot;
Assert(slot != NULL);
Assert(restart_lsn != InvalidXLogRecPtr);
Assert(current_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(restart_lsn));
Assert(XLogRecPtrIsValid(current_lsn));
SpinLockAcquire(&slot->mutex);
@ -1779,7 +1779,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
* might never end up updating if the receiver acks too slowly. A missed
* value here will just cause some extra effort after reconnecting.
*/
else if (slot->candidate_restart_valid == InvalidXLogRecPtr)
else if (!XLogRecPtrIsValid(slot->candidate_restart_valid))
{
slot->candidate_restart_valid = current_lsn;
slot->candidate_restart_lsn = restart_lsn;
@ -1819,11 +1819,11 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
void
LogicalConfirmReceivedLocation(XLogRecPtr lsn)
{
Assert(lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(lsn));
/* Do an unlocked check for candidate_lsn first. */
if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr ||
MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(MyReplicationSlot->candidate_xmin_lsn) ||
XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_valid))
{
bool updated_xmin = false;
bool updated_restart = false;
@ -1849,7 +1849,7 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
MyReplicationSlot->data.confirmed_flush = lsn;
/* if we're past the location required for bumping xmin, do so */
if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(MyReplicationSlot->candidate_xmin_lsn) &&
MyReplicationSlot->candidate_xmin_lsn <= lsn)
{
/*
@ -1871,10 +1871,10 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
}
}
if (MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_valid) &&
MyReplicationSlot->candidate_restart_valid <= lsn)
{
Assert(MyReplicationSlot->candidate_restart_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_lsn));
MyReplicationSlot->data.restart_lsn = MyReplicationSlot->candidate_restart_lsn;
MyReplicationSlot->candidate_restart_lsn = InvalidXLogRecPtr;
@ -2089,7 +2089,7 @@ LogicalSlotAdvanceAndCheckSnapState(XLogRecPtr moveto,
ResourceOwner old_resowner PG_USED_FOR_ASSERTS_ONLY = CurrentResourceOwner;
XLogRecPtr retlsn;
Assert(moveto != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(moveto));
if (found_consistent_snapshot)
*found_consistent_snapshot = false;
@ -2163,7 +2163,7 @@ LogicalSlotAdvanceAndCheckSnapState(XLogRecPtr moveto,
if (found_consistent_snapshot && DecodingContextReady(ctx))
*found_consistent_snapshot = true;
if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(ctx->reader->EndRecPtr))
{
LogicalConfirmReceivedLocation(moveto);

6
src/backend/replication/logical/logicalfuncs.c
Unescape View File

@ -229,7 +229,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
* Wait for specified streaming replication standby servers (if any)
* to confirm receipt of WAL up to wait_for_wal_lsn.
*/
if (XLogRecPtrIsInvalid(upto_lsn))
if (!XLogRecPtrIsValid(upto_lsn))
wait_for_wal_lsn = end_of_wal;
else
wait_for_wal_lsn = Min(upto_lsn, end_of_wal);
@ -276,7 +276,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
}
/* check limits */
if (upto_lsn != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(upto_lsn) &&
upto_lsn <= ctx->reader->EndRecPtr)
break;
if (upto_nchanges != 0 &&
@ -289,7 +289,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
* Next time, start where we left off. (Hunting things, the family
* business..)
*/
if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
if (XLogRecPtrIsValid(ctx->reader->EndRecPtr) && confirm)
{
LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);

18
src/backend/replication/logical/origin.c
Unescape View File

@ -984,8 +984,8 @@ replorigin_advance(RepOriginId node,
/* initialize new slot */
LWLockAcquire(&free_state->lock, LW_EXCLUSIVE);
replication_state = free_state;
Assert(replication_state->remote_lsn == InvalidXLogRecPtr);
Assert(replication_state->local_lsn == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(replication_state->remote_lsn));
Assert(!XLogRecPtrIsValid(replication_state->local_lsn));
replication_state->roident = node;
}
@ -1020,7 +1020,7 @@ replorigin_advance(RepOriginId node,
*/
if (go_backward || replication_state->remote_lsn < remote_commit)
replication_state->remote_lsn = remote_commit;
if (local_commit != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(local_commit) &&
(go_backward || replication_state->local_lsn < local_commit))
replication_state->local_lsn = local_commit;
LWLockRelease(&replication_state->lock);
@ -1064,7 +1064,7 @@ replorigin_get_progress(RepOriginId node, bool flush)
LWLockRelease(ReplicationOriginLock);
if (flush && local_lsn != InvalidXLogRecPtr)
if (flush && XLogRecPtrIsValid(local_lsn))
XLogFlush(local_lsn);
return remote_lsn;
@ -1197,8 +1197,8 @@ replorigin_session_setup(RepOriginId node, int acquired_by)
/* initialize new slot */
session_replication_state = &replication_states[free_slot];
Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(session_replication_state->remote_lsn));
Assert(!XLogRecPtrIsValid(session_replication_state->local_lsn));
session_replication_state->roident = node;
}
@ -1282,7 +1282,7 @@ replorigin_session_get_progress(bool flush)
local_lsn = session_replication_state->local_lsn;
LWLockRelease(&session_replication_state->lock);
if (flush && local_lsn != InvalidXLogRecPtr)
if (flush && XLogRecPtrIsValid(local_lsn))
XLogFlush(local_lsn);
return remote_lsn;
@ -1454,7 +1454,7 @@ pg_replication_origin_session_progress(PG_FUNCTION_ARGS)
remote_lsn = replorigin_session_get_progress(flush);
if (remote_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(remote_lsn))
PG_RETURN_NULL();
PG_RETURN_LSN(remote_lsn);
@ -1543,7 +1543,7 @@ pg_replication_origin_progress(PG_FUNCTION_ARGS)
remote_lsn = replorigin_get_progress(roident, flush);
if (remote_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(remote_lsn))
PG_RETURN_NULL();
PG_RETURN_LSN(remote_lsn);

18
src/backend/replication/logical/proto.c
Unescape View File

@ -64,7 +64,7 @@ logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data)
{
/* read fields */
begin_data->final_lsn = pq_getmsgint64(in);
if (begin_data->final_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(begin_data->final_lsn))
elog(ERROR, "final_lsn not set in begin message");
begin_data->committime = pq_getmsgint64(in);
begin_data->xid = pq_getmsgint(in, 4);
@ -135,10 +135,10 @@ logicalrep_read_begin_prepare(StringInfo in, LogicalRepPreparedTxnData *begin_da
{
/* read fields */
begin_data->prepare_lsn = pq_getmsgint64(in);
if (begin_data->prepare_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(begin_data->prepare_lsn))
elog(ERROR, "prepare_lsn not set in begin prepare message");
begin_data->end_lsn = pq_getmsgint64(in);
if (begin_data->end_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(begin_data->end_lsn))
elog(ERROR, "end_lsn not set in begin prepare message");
begin_data->prepare_time = pq_getmsgint64(in);
begin_data->xid = pq_getmsgint(in, 4);
@ -207,10 +207,10 @@ logicalrep_read_prepare_common(StringInfo in, char *msgtype,
/* read fields */
prepare_data->prepare_lsn = pq_getmsgint64(in);
if (prepare_data->prepare_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(prepare_data->prepare_lsn))
elog(ERROR, "prepare_lsn is not set in %s message", msgtype);
prepare_data->end_lsn = pq_getmsgint64(in);
if (prepare_data->end_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(prepare_data->end_lsn))
elog(ERROR, "end_lsn is not set in %s message", msgtype);
prepare_data->prepare_time = pq_getmsgint64(in);
prepare_data->xid = pq_getmsgint(in, 4);
@ -274,10 +274,10 @@ logicalrep_read_commit_prepared(StringInfo in, LogicalRepCommitPreparedTxnData *
/* read fields */
prepare_data->commit_lsn = pq_getmsgint64(in);
if (prepare_data->commit_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(prepare_data->commit_lsn))
elog(ERROR, "commit_lsn is not set in commit prepared message");
prepare_data->end_lsn = pq_getmsgint64(in);
if (prepare_data->end_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(prepare_data->end_lsn))
elog(ERROR, "end_lsn is not set in commit prepared message");
prepare_data->commit_time = pq_getmsgint64(in);
prepare_data->xid = pq_getmsgint(in, 4);
@ -333,10 +333,10 @@ logicalrep_read_rollback_prepared(StringInfo in,
/* read fields */
rollback_data->prepare_end_lsn = pq_getmsgint64(in);
if (rollback_data->prepare_end_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(rollback_data->prepare_end_lsn))
elog(ERROR, "prepare_end_lsn is not set in rollback prepared message");
rollback_data->rollback_end_lsn = pq_getmsgint64(in);
if (rollback_data->rollback_end_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(rollback_data->rollback_end_lsn))
elog(ERROR, "rollback_end_lsn is not set in rollback prepared message");
rollback_data->prepare_time = pq_getmsgint64(in);
rollback_data->rollback_time = pq_getmsgint64(in);

38
src/backend/replication/logical/reorderbuffer.c
Unescape View File

@ -701,7 +701,7 @@ ReorderBufferTXNByXid(ReorderBuffer *rb, TransactionId xid, bool create,
{
/* initialize the new entry, if creation was requested */
Assert(ent != NULL);
Assert(lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(lsn));
ent->txn = ReorderBufferAllocTXN(rb);
ent->txn->xid = xid;
@ -849,7 +849,7 @@ ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
change->lsn = lsn;
change->txn = txn;
Assert(InvalidXLogRecPtr != lsn);
Assert(XLogRecPtrIsValid(lsn));
dlist_push_tail(&txn->changes, &change->node);
txn->nentries++;
txn->nentries_mem++;
@ -966,14 +966,14 @@ AssertTXNLsnOrder(ReorderBuffer *rb)
iter.cur);
/* start LSN must be set */
Assert(cur_txn->first_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(cur_txn->first_lsn));
/* If there is an end LSN, it must be higher than start LSN */
if (cur_txn->end_lsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(cur_txn->end_lsn))
Assert(cur_txn->first_lsn <= cur_txn->end_lsn);
/* Current initial LSN must be strictly higher than previous */
if (prev_first_lsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(prev_first_lsn))
Assert(prev_first_lsn < cur_txn->first_lsn);
/* known-as-subtxn txns must not be listed */
@ -990,10 +990,10 @@ AssertTXNLsnOrder(ReorderBuffer *rb)
/* base snapshot (and its LSN) must be set */
Assert(cur_txn->base_snapshot != NULL);
Assert(cur_txn->base_snapshot_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(cur_txn->base_snapshot_lsn));
/* current LSN must be strictly higher than previous */
if (prev_base_snap_lsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(prev_base_snap_lsn))
Assert(prev_base_snap_lsn < cur_txn->base_snapshot_lsn);
/* known-as-subtxn txns must not be listed */
@ -1022,11 +1022,11 @@ AssertChangeLsnOrder(ReorderBufferTXN *txn)
cur_change = dlist_container(ReorderBufferChange, node, iter.cur);
Assert(txn->first_lsn != InvalidXLogRecPtr);
Assert(cur_change->lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->first_lsn));
Assert(XLogRecPtrIsValid(cur_change->lsn));
Assert(txn->first_lsn <= cur_change->lsn);
if (txn->end_lsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(txn->end_lsn))
Assert(cur_change->lsn <= txn->end_lsn);
Assert(prev_lsn <= cur_change->lsn);
@ -1053,7 +1053,7 @@ ReorderBufferGetOldestTXN(ReorderBuffer *rb)
txn = dlist_head_element(ReorderBufferTXN, node, &rb->toplevel_by_lsn);
Assert(!rbtxn_is_known_subxact(txn));
Assert(txn->first_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->first_lsn));
return txn;
}
@ -2276,7 +2276,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
* We can't call start stream callback before processing first
* change.
*/
if (prev_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(prev_lsn))
{
if (streaming)
{
@ -2291,7 +2291,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
* subtransactions. The changes may have the same LSN due to
* MULTI_INSERT xlog records.
*/
Assert(prev_lsn == InvalidXLogRecPtr || prev_lsn <= change->lsn);
Assert(!XLogRecPtrIsValid(prev_lsn) || prev_lsn <= change->lsn);
prev_lsn = change->lsn;
@ -2975,7 +2975,7 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
* have been updated in it by now.
*/
Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == RBTXN_IS_PREPARED);
Assert(txn->final_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->final_lsn));
txn->gid = pstrdup(gid);
@ -3041,7 +3041,7 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
*/
Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) ==
(RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE));
Assert(txn->final_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->final_lsn));
/*
* By this time the txn has the prepare record information and it is
@ -4552,8 +4552,8 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
dlist_mutable_iter cleanup_iter;
File *fd = &file->vfd;
Assert(txn->first_lsn != InvalidXLogRecPtr);
Assert(txn->final_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->first_lsn));
Assert(XLogRecPtrIsValid(txn->final_lsn));
/* free current entries, so we have memory for more */
dlist_foreach_modify(cleanup_iter, &txn->changes)
@ -4860,8 +4860,8 @@ ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn)
XLogSegNo cur;
XLogSegNo last;
Assert(txn->first_lsn != InvalidXLogRecPtr);
Assert(txn->final_lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(txn->first_lsn));
Assert(XLogRecPtrIsValid(txn->final_lsn));
XLByteToSeg(txn->first_lsn, first, wal_segment_size);
XLByteToSeg(txn->final_lsn, last, wal_segment_size);

6
src/backend/replication/logical/slotsync.c
Unescape View File

@ -493,7 +493,7 @@ reserve_wal_for_local_slot(XLogRecPtr restart_lsn)
ReplicationSlot *slot = MyReplicationSlot;
Assert(slot != NULL);
Assert(XLogRecPtrIsInvalid(slot->data.restart_lsn));
Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
while (true)
{
@ -899,8 +899,8 @@ synchronize_slots(WalReceiverConn *wrconn)
* pg_replication_slots view, then we can avoid fetching RS_EPHEMERAL
* slots in the first place.
*/
if ((XLogRecPtrIsInvalid(remote_slot->restart_lsn) ||
XLogRecPtrIsInvalid(remote_slot->confirmed_lsn) ||
if ((!XLogRecPtrIsValid(remote_slot->restart_lsn) ||
!XLogRecPtrIsValid(remote_slot->confirmed_lsn) ||
!TransactionIdIsValid(remote_slot->catalog_xmin)) &&
remote_slot->invalidated == RS_INVAL_NONE)
pfree(remote_slot);

14
src/backend/replication/logical/snapbuild.c
Unescape View File

@ -1210,7 +1210,7 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
* oldest ongoing txn might have started when we didn't yet serialize
* anything because we hadn't reached a consistent state yet.
*/
if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
if (txn != NULL && XLogRecPtrIsValid(txn->restart_decoding_lsn))
LogicalIncreaseRestartDecodingForSlot(lsn, txn->restart_decoding_lsn);
/*
@ -1218,8 +1218,8 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
* we have one.
*/
else if (txn == NULL &&
builder->reorder->current_restart_decoding_lsn != InvalidXLogRecPtr &&
builder->last_serialized_snapshot != InvalidXLogRecPtr)
XLogRecPtrIsValid(builder->reorder->current_restart_decoding_lsn) &&
XLogRecPtrIsValid(builder->last_serialized_snapshot))
LogicalIncreaseRestartDecodingForSlot(lsn,
builder->last_serialized_snapshot);
}
@ -1293,7 +1293,7 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
*/
if (running->oldestRunningXid == running->nextXid)
{
if (builder->start_decoding_at == InvalidXLogRecPtr ||
if (!XLogRecPtrIsValid(builder->start_decoding_at) ||
builder->start_decoding_at <= lsn)
/* can decode everything after this */
builder->start_decoding_at = lsn + 1;
@ -1509,8 +1509,8 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
struct stat stat_buf;
Size sz;
Assert(lsn != InvalidXLogRecPtr);
Assert(builder->last_serialized_snapshot == InvalidXLogRecPtr ||
Assert(XLogRecPtrIsValid(lsn));
Assert(!XLogRecPtrIsValid(builder->last_serialized_snapshot) ||
builder->last_serialized_snapshot <= lsn);
/*
@ -2029,7 +2029,7 @@ CheckPointSnapBuild(void)
lsn = ((uint64) hi) << 32 | lo;
/* check whether we still need it */
if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
{
elog(DEBUG1, "removing snapbuild snapshot %s", path);

16
src/backend/replication/logical/worker.c
Unescape View File

@ -514,7 +514,7 @@ bool InitializingApplyWorker = false;
* by the user.
*/
static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
#define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
#define is_skipping_changes() (unlikely(XLogRecPtrIsValid(skip_xact_finish_lsn)))
/* BufFile handle of the current streaming file */
static BufFile *stream_fd = NULL;
@ -4126,7 +4126,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
* due to a bug, we don't want to proceed as it can
* incorrectly advance oldest_nonremovable_xid.
*/
if (XLogRecPtrIsInvalid(rdt_data.remote_lsn))
if (!XLogRecPtrIsValid(rdt_data.remote_lsn))
elog(ERROR, "cannot get the latest WAL position from the publisher");
maybe_advance_nonremovable_xid(&rdt_data, true);
@ -4613,7 +4613,7 @@ wait_for_publisher_status(RetainDeadTuplesData *rdt_data,
static void
wait_for_local_flush(RetainDeadTuplesData *rdt_data)
{
Assert(!XLogRecPtrIsInvalid(rdt_data->remote_lsn) &&
Assert(XLogRecPtrIsValid(rdt_data->remote_lsn) &&
TransactionIdIsValid(rdt_data->candidate_xid));
/*
@ -6031,7 +6031,7 @@ maybe_start_skipping_changes(XLogRecPtr finish_lsn)
* function is called for every remote transaction and we assume that
* skipping the transaction is not used often.
*/
if (likely(XLogRecPtrIsInvalid(MySubscription->skiplsn) ||
if (likely(!XLogRecPtrIsValid(MySubscription->skiplsn) ||
MySubscription->skiplsn != finish_lsn))
return;
@ -6077,7 +6077,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
XLogRecPtr myskiplsn = MySubscription->skiplsn;
bool started_tx = false;
if (likely(XLogRecPtrIsInvalid(myskiplsn)) || am_parallel_apply_worker())
if (likely(!XLogRecPtrIsValid(myskiplsn)) || am_parallel_apply_worker())
return;
if (!IsTransactionState())
@ -6173,7 +6173,7 @@ apply_error_callback(void *arg)
errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
errarg->origin_name,
logicalrep_message_type(errarg->command));
else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
else if (!XLogRecPtrIsValid(errarg->finish_lsn))
errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
errarg->origin_name,
logicalrep_message_type(errarg->command),
@ -6189,7 +6189,7 @@ apply_error_callback(void *arg)
{
if (errarg->remote_attnum < 0)
{
if (XLogRecPtrIsInvalid(errarg->finish_lsn))
if (!XLogRecPtrIsValid(errarg->finish_lsn))
errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
errarg->origin_name,
logicalrep_message_type(errarg->command),
@ -6207,7 +6207,7 @@ apply_error_callback(void *arg)
}
else
{
if (XLogRecPtrIsInvalid(errarg->finish_lsn))
if (!XLogRecPtrIsValid(errarg->finish_lsn))
errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
errarg->origin_name,
logicalrep_message_type(errarg->command),

28
src/backend/replication/slot.c
Unescape View File

@ -1270,15 +1270,15 @@ ReplicationSlotsComputeRequiredLSN(void)
*/
if (persistency == RS_PERSISTENT)
{
if (last_saved_restart_lsn != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
restart_lsn > last_saved_restart_lsn)
{
restart_lsn = last_saved_restart_lsn;
}
}
if (restart_lsn != InvalidXLogRecPtr &&
(min_required == InvalidXLogRecPtr ||
if (XLogRecPtrIsValid(restart_lsn) &&
(!XLogRecPtrIsValid(min_required) ||
restart_lsn < min_required))
min_required = restart_lsn;
}
@ -1350,17 +1350,17 @@ ReplicationSlotsComputeLogicalRestartLSN(void)
*/
if (persistency == RS_PERSISTENT)
{
if (last_saved_restart_lsn != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
restart_lsn > last_saved_restart_lsn)
{
restart_lsn = last_saved_restart_lsn;
}
}
if (restart_lsn == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(restart_lsn))
continue;
if (result == InvalidXLogRecPtr ||
if (!XLogRecPtrIsValid(result) ||
restart_lsn < result)
result = restart_lsn;
}
@ -1573,8 +1573,8 @@ ReplicationSlotReserveWal(void)
ReplicationSlot *slot = MyReplicationSlot;
Assert(slot != NULL);
Assert(slot->data.restart_lsn == InvalidXLogRecPtr);
Assert(slot->last_saved_restart_lsn == InvalidXLogRecPtr);
Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
Assert(!XLogRecPtrIsValid(slot->last_saved_restart_lsn));
/*
* The replication slot mechanism is used to prevent removal of required
@ -1730,7 +1730,7 @@ static inline bool
CanInvalidateIdleSlot(ReplicationSlot *s)
{
return (idle_replication_slot_timeout_secs != 0 &&
!XLogRecPtrIsInvalid(s->data.restart_lsn) &&
XLogRecPtrIsValid(s->data.restart_lsn) &&
s->inactive_since > 0 &&
!(RecoveryInProgress() && s->data.synced));
}
@ -1754,7 +1754,7 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s,
{
XLogRecPtr restart_lsn = s->data.restart_lsn;
if (restart_lsn != InvalidXLogRecPtr &&
if (XLogRecPtrIsValid(restart_lsn) &&
restart_lsn < oldestLSN)
return RS_INVAL_WAL_REMOVED;
}
@ -2922,7 +2922,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
* Don't need to wait for the standbys to catch up if they are already
* beyond the specified WAL location.
*/
if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
if (XLogRecPtrIsValid(ss_oldest_flush_lsn) &&
ss_oldest_flush_lsn >= wait_for_lsn)
return true;
@ -2993,7 +2993,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
break;
}
if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
if (!XLogRecPtrIsValid(restart_lsn) || restart_lsn < wait_for_lsn)
{
/* Log a message if no active_pid for this physical slot */
if (inactive)
@ -3012,7 +3012,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
Assert(restart_lsn >= wait_for_lsn);
if (XLogRecPtrIsInvalid(min_restart_lsn) ||
if (!XLogRecPtrIsValid(min_restart_lsn) ||
min_restart_lsn > restart_lsn)
min_restart_lsn = restart_lsn;
@ -3031,7 +3031,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
return false;
/* The ss_oldest_flush_lsn must not retreat. */
Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
Assert(!XLogRecPtrIsValid(ss_oldest_flush_lsn) ||
min_restart_lsn >= ss_oldest_flush_lsn);
ss_oldest_flush_lsn = min_restart_lsn;

22
src/backend/replication/slotfuncs.c
Unescape View File

@ -46,7 +46,7 @@ create_physical_replication_slot(char *name, bool immediately_reserve,
if (immediately_reserve)
{
/* Reserve WAL as the user asked for it */
if (XLogRecPtrIsInvalid(restart_lsn))
if (!XLogRecPtrIsValid(restart_lsn))
ReplicationSlotReserveWal();
else
MyReplicationSlot->data.restart_lsn = restart_lsn;
@ -308,12 +308,12 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
else
nulls[i++] = true;
if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(slot_contents.data.confirmed_flush))
values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
else
nulls[i++] = true;
@ -357,7 +357,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
*
* If we do change it, save the state for safe_wal_size below.
*/
if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
{
int pid;
@ -407,7 +407,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.two_phase);
if (slot_contents.data.two_phase &&
!XLogRecPtrIsInvalid(slot_contents.data.two_phase_at))
XLogRecPtrIsValid(slot_contents.data.two_phase_at))
values[i++] = LSNGetDatum(slot_contents.data.two_phase_at);
else
nulls[i++] = true;
@ -467,7 +467,7 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
XLogRecPtr startlsn = MyReplicationSlot->data.restart_lsn;
XLogRecPtr retlsn = startlsn;
Assert(moveto != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(moveto));
if (startlsn < moveto)
{
@ -523,7 +523,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
CheckSlotPermissions();
if (XLogRecPtrIsInvalid(moveto))
if (!XLogRecPtrIsValid(moveto))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("invalid target WAL LSN")));
@ -545,7 +545,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
ReplicationSlotAcquire(NameStr(*slotname), true, true);
/* A slot whose restart_lsn has never been reserved cannot be advanced */
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
if (!XLogRecPtrIsValid(MyReplicationSlot->data.restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("replication slot \"%s\" cannot be advanced",
@ -679,7 +679,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
NameStr(*src_name))));
/* Copying non-reserved slot doesn't make sense */
if (XLogRecPtrIsInvalid(src_restart_lsn))
if (!XLogRecPtrIsValid(src_restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("cannot copy a replication slot that doesn't reserve WAL")));
@ -785,7 +785,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
errdetail("The source replication slot was modified incompatibly during the copy operation.")));
/* The source slot must have a consistent snapshot */
if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
if (src_islogical && !XLogRecPtrIsValid(copy_confirmed_flush))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot copy unfinished logical replication slot \"%s\"",
@ -840,7 +840,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
/* All done. Set up the return values */
values[0] = NameGetDatum(dst_name);
nulls[0] = false;
if (!XLogRecPtrIsInvalid(MyReplicationSlot->data.confirmed_flush))
if (XLogRecPtrIsValid(MyReplicationSlot->data.confirmed_flush))
{
values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
nulls[1] = false;

10
src/backend/replication/syncrep.c
Unescape View File

@ -493,7 +493,7 @@ SyncRepReleaseWaiters(void)
if (MyWalSnd->sync_standby_priority == 0 ||
(MyWalSnd->state != WALSNDSTATE_STREAMING &&
MyWalSnd->state != WALSNDSTATE_STOPPING) ||
XLogRecPtrIsInvalid(MyWalSnd->flush))
!XLogRecPtrIsValid(MyWalSnd->flush))
{
announce_next_takeover = true;
return;
@ -676,11 +676,11 @@ SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr,
XLogRecPtr flush = sync_standbys[i].flush;
XLogRecPtr apply = sync_standbys[i].apply;
if (XLogRecPtrIsInvalid(*writePtr) || *writePtr > write)
if (!XLogRecPtrIsValid(*writePtr) || *writePtr > write)
*writePtr = write;
if (XLogRecPtrIsInvalid(*flushPtr) || *flushPtr > flush)
if (!XLogRecPtrIsValid(*flushPtr) || *flushPtr > flush)
*flushPtr = flush;
if (XLogRecPtrIsInvalid(*applyPtr) || *applyPtr > apply)
if (!XLogRecPtrIsValid(*applyPtr) || *applyPtr > apply)
*applyPtr = apply;
}
}
@ -799,7 +799,7 @@ SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys)
continue;
/* Must have a valid flush position */
if (XLogRecPtrIsInvalid(stby->flush))
if (!XLogRecPtrIsValid(stby->flush))
continue;
/* OK, it's a candidate */

8
src/backend/replication/walreceiver.c
Unescape View File

@ -1469,16 +1469,16 @@ pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
{
values[1] = CStringGetTextDatum(WalRcvGetStateString(state));
if (XLogRecPtrIsInvalid(receive_start_lsn))
if (!XLogRecPtrIsValid(receive_start_lsn))
nulls[2] = true;
else
values[2] = LSNGetDatum(receive_start_lsn);
values[3] = Int32GetDatum(receive_start_tli);
if (XLogRecPtrIsInvalid(written_lsn))
if (!XLogRecPtrIsValid(written_lsn))
nulls[4] = true;
else
values[4] = LSNGetDatum(written_lsn);
if (XLogRecPtrIsInvalid(flushed_lsn))
if (!XLogRecPtrIsValid(flushed_lsn))
nulls[5] = true;
else
values[5] = LSNGetDatum(flushed_lsn);
@ -1491,7 +1491,7 @@ pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
nulls[8] = true;
else
values[8] = TimestampTzGetDatum(last_receipt_time);
if (XLogRecPtrIsInvalid(latest_end_lsn))
if (!XLogRecPtrIsValid(latest_end_lsn))
nulls[9] = true;
else
values[9] = LSNGetDatum(latest_end_lsn);

2
src/backend/replication/walreceiverfuncs.c
Unescape View File

@ -315,7 +315,7 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
* If this is the first startup of walreceiver (on this timeline),
* initialize flushedUpto and latestChunkStart to the starting point.
*/
if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
if (!XLogRecPtrIsValid(walrcv->receiveStart) || walrcv->receivedTLI != tli)
{
walrcv->flushedUpto = recptr;
walrcv->receivedTLI = tli;

30
src/backend/replication/walsender.c
Unescape View File

@ -529,7 +529,7 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
i++;
/* start LSN */
if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
{
char xloc[64];
@ -541,7 +541,7 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
i++;
/* timeline this WAL was produced on */
if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
{
TimeLineID slots_position_timeline;
TimeLineID current_timeline;
@ -906,7 +906,7 @@ StartReplication(StartReplicationCmd *cmd)
* that's older than the switchpoint, if it's still in the same
* WAL segment.
*/
if (!XLogRecPtrIsInvalid(switchpoint) &&
if (XLogRecPtrIsValid(switchpoint) &&
switchpoint < cmd->startpoint)
{
ereport(ERROR,
@ -1827,7 +1827,7 @@ WalSndWaitForWal(XLogRecPtr loc)
* receipt of WAL up to RecentFlushPtr. This is particularly interesting
* if we're far behind.
*/
if (!XLogRecPtrIsInvalid(RecentFlushPtr) &&
if (XLogRecPtrIsValid(RecentFlushPtr) &&
!NeedToWaitForWal(loc, RecentFlushPtr, &wait_event))
return RecentFlushPtr;
@ -2397,7 +2397,7 @@ PhysicalConfirmReceivedLocation(XLogRecPtr lsn)
bool changed = false;
ReplicationSlot *slot = MyReplicationSlot;
Assert(lsn != InvalidXLogRecPtr);
Assert(XLogRecPtrIsValid(lsn));
SpinLockAcquire(&slot->mutex);
if (slot->data.restart_lsn != lsn)
{
@ -2519,7 +2519,7 @@ ProcessStandbyReplyMessage(void)
/*
* Advance our local xmin horizon when the client confirmed a flush.
*/
if (MyReplicationSlot && flushPtr != InvalidXLogRecPtr)
if (MyReplicationSlot && XLogRecPtrIsValid(flushPtr))
{
if (SlotIsLogical(MyReplicationSlot))
LogicalConfirmReceivedLocation(flushPtr);
@ -3536,7 +3536,7 @@ XLogSendLogical(void)
* If first time through in this session, initialize flushPtr. Otherwise,
* we only need to update flushPtr if EndRecPtr is past it.
*/
if (flushPtr == InvalidXLogRecPtr ||
if (!XLogRecPtrIsValid(flushPtr) ||
logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
{
/*
@ -3597,8 +3597,8 @@ WalSndDone(WalSndSendDataCallback send_data)
* flush location if valid, write otherwise. Tools like pg_receivewal will
* usually (unless in synchronous mode) return an invalid flush location.
*/
replicatedPtr = XLogRecPtrIsInvalid(MyWalSnd->flush) ?
MyWalSnd->write : MyWalSnd->flush;
replicatedPtr = XLogRecPtrIsValid(MyWalSnd->flush) ?
MyWalSnd->flush : MyWalSnd->write;
if (WalSndCaughtUp && sentPtr == replicatedPtr &&
!pq_is_send_pending())
@ -4073,19 +4073,19 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
{
values[1] = CStringGetTextDatum(WalSndGetStateString(state));
if (XLogRecPtrIsInvalid(sent_ptr))
if (!XLogRecPtrIsValid(sent_ptr))
nulls[2] = true;
values[2] = LSNGetDatum(sent_ptr);
if (XLogRecPtrIsInvalid(write))
if (!XLogRecPtrIsValid(write))
nulls[3] = true;
values[3] = LSNGetDatum(write);
if (XLogRecPtrIsInvalid(flush))
if (!XLogRecPtrIsValid(flush))
nulls[4] = true;
values[4] = LSNGetDatum(flush);
if (XLogRecPtrIsInvalid(apply))
if (!XLogRecPtrIsValid(apply))
nulls[5] = true;
values[5] = LSNGetDatum(apply);
@ -4094,7 +4094,7 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
* which always returns an invalid flush location, as an
* asynchronous standby.
*/
priority = XLogRecPtrIsInvalid(flush) ? 0 : priority;
priority = XLogRecPtrIsValid(flush) ? priority : 0;
if (writeLag < 0)
nulls[6] = true;
@ -4165,7 +4165,7 @@ WalSndKeepalive(bool requestReply, XLogRecPtr writePtr)
/* construct the message... */
resetStringInfo(&output_message);
pq_sendbyte(&output_message, PqReplMsg_Keepalive);
pq_sendint64(&output_message, XLogRecPtrIsInvalid(writePtr) ? sentPtr : writePtr);
pq_sendint64(&output_message, XLogRecPtrIsValid(writePtr) ? writePtr : sentPtr);
pq_sendint64(&output_message, GetCurrentTimestamp());
pq_sendbyte(&output_message, requestReply ? 1 : 0);

2
src/backend/storage/buffer/bufmgr.c
Unescape View File

@ -5545,7 +5545,7 @@ MarkBufferDirtyHint(Buffer buffer, bool buffer_std)
* checksum here. That will happen when the page is written
* sometime later in this checkpoint cycle.
*/
if (!XLogRecPtrIsInvalid(lsn))
if (XLogRecPtrIsValid(lsn))
PageSetLSN(page, lsn);
}

8
src/bin/pg_basebackup/pg_receivewal.c
Unescape View File

@ -192,7 +192,7 @@ stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
LSN_FORMAT_ARGS(xlogpos),
timeline);
if (!XLogRecPtrIsInvalid(endpos) && endpos < xlogpos)
if (XLogRecPtrIsValid(endpos) && endpos < xlogpos)
{
if (verbose)
pg_log_info("stopped log streaming at %X/%08X (timeline %u)",
@ -535,7 +535,7 @@ StreamLog(void)
* Figure out where to start streaming. First scan the local directory.
*/
stream.startpos = FindStreamingStart(&stream.timeline);
if (stream.startpos == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(stream.startpos))
{
/*
* Try to get the starting point from the slot if any. This is
@ -556,14 +556,14 @@ StreamLog(void)
* If it the starting point is still not known, use the current WAL
* flush value as last resort.
*/
if (stream.startpos == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(stream.startpos))
{
stream.startpos = serverpos;
stream.timeline = servertli;
}
}
Assert(stream.startpos != InvalidXLogRecPtr &&
Assert(XLogRecPtrIsValid(stream.startpos) &&
stream.timeline != 0);
/*

12
src/bin/pg_basebackup/pg_recvlogical.c
Unescape View File

@ -482,7 +482,7 @@ StreamLogicalLog(void)
}
replyRequested = copybuf[pos];
if (endpos != InvalidXLogRecPtr && walEnd >= endpos)
if (XLogRecPtrIsValid(endpos) && walEnd >= endpos)
{
/*
* If there's nothing to read on the socket until a keepalive
@ -535,7 +535,7 @@ StreamLogicalLog(void)
/* Extract WAL location for this block */
cur_record_lsn = fe_recvint64(&copybuf[1]);
if (endpos != InvalidXLogRecPtr && cur_record_lsn > endpos)
if (XLogRecPtrIsValid(endpos) && cur_record_lsn > endpos)
{
/*
* We've read past our endpoint, so prepare to go away being
@ -583,7 +583,7 @@ StreamLogicalLog(void)
goto error;
}
if (endpos != InvalidXLogRecPtr && cur_record_lsn == endpos)
if (XLogRecPtrIsValid(endpos) && cur_record_lsn == endpos)
{
/* endpos was exactly the record we just processed, we're done */
if (!flushAndSendFeedback(conn, &now))
@ -913,14 +913,14 @@ main(int argc, char **argv)
exit(1);
}
if (startpos != InvalidXLogRecPtr && (do_create_slot || do_drop_slot))
if (XLogRecPtrIsValid(startpos) && (do_create_slot || do_drop_slot))
{
pg_log_error("cannot use --create-slot or --drop-slot together with --startpos");
pg_log_error_hint("Try \"%s --help\" for more information.", progname);
exit(1);
}
if (endpos != InvalidXLogRecPtr && !do_start_slot)
if (XLogRecPtrIsValid(endpos) && !do_start_slot)
{
pg_log_error("--endpos may only be specified with --start");
pg_log_error_hint("Try \"%s --help\" for more information.", progname);
@ -1080,7 +1080,7 @@ prepareToTerminate(PGconn *conn, XLogRecPtr endpos, StreamStopReason reason,
LSN_FORMAT_ARGS(endpos));
break;
case STREAM_STOP_END_OF_WAL:
Assert(!XLogRecPtrIsInvalid(lsn));
Assert(XLogRecPtrIsValid(lsn));
pg_log_info("end position %X/%08X reached by WAL record at %X/%08X",
LSN_FORMAT_ARGS(endpos), LSN_FORMAT_ARGS(lsn));
break;

4
src/bin/pg_rewind/pg_rewind.c
Unescape View File

@ -850,9 +850,9 @@ progress_report(bool finished)
static XLogRecPtr
MinXLogRecPtr(XLogRecPtr a, XLogRecPtr b)
{
if (XLogRecPtrIsInvalid(a))
if (!XLogRecPtrIsValid(a))
return b;
else if (XLogRecPtrIsInvalid(b))
else if (!XLogRecPtrIsValid(b))
return a;
else
return Min(a, b);

14
src/bin/pg_waldump/pg_waldump.c
Unescape View File

@ -393,7 +393,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
int count = XLOG_BLCKSZ;
WALReadError errinfo;
if (private->endptr != InvalidXLogRecPtr)
if (XLogRecPtrIsValid(private->endptr))
{
if (targetPagePtr + XLOG_BLCKSZ <= private->endptr)
count = XLOG_BLCKSZ;
@ -637,7 +637,7 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats)
/*
* Leave if no stats have been computed yet, as tracked by the end LSN.
*/
if (XLogRecPtrIsInvalid(stats->endptr))
if (!XLogRecPtrIsValid(stats->endptr))
return;
/*
@ -1136,7 +1136,7 @@ main(int argc, char **argv)
/* parse position from file */
XLogFromFileName(fname, &private.timeline, &segno, WalSegSz);
if (XLogRecPtrIsInvalid(private.startptr))
if (!XLogRecPtrIsValid(private.startptr))
XLogSegNoOffsetToRecPtr(segno, 0, WalSegSz, private.startptr);
else if (!XLByteInSeg(private.startptr, segno, WalSegSz))
{
@ -1147,7 +1147,7 @@ main(int argc, char **argv)
}
/* no second file specified, set end position */
if (!(optind + 1 < argc) && XLogRecPtrIsInvalid(private.endptr))
if (!(optind + 1 < argc) && !XLogRecPtrIsValid(private.endptr))
XLogSegNoOffsetToRecPtr(segno + 1, 0, WalSegSz, private.endptr);
/* parse ENDSEG if passed */
@ -1170,7 +1170,7 @@ main(int argc, char **argv)
pg_fatal("ENDSEG %s is before STARTSEG %s",
argv[optind + 1], argv[optind]);
if (XLogRecPtrIsInvalid(private.endptr))
if (!XLogRecPtrIsValid(private.endptr))
XLogSegNoOffsetToRecPtr(endsegno + 1, 0, WalSegSz,
private.endptr);
@ -1192,7 +1192,7 @@ main(int argc, char **argv)
waldir = identify_target_directory(waldir, NULL);
/* we don't know what to print */
if (XLogRecPtrIsInvalid(private.startptr))
if (!XLogRecPtrIsValid(private.startptr))
{
pg_log_error("no start WAL location given");
goto bad_argument;
@ -1213,7 +1213,7 @@ main(int argc, char **argv)
/* first find a valid recptr to start from */
first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
if (first_record == InvalidXLogRecPtr)
if (!XLogRecPtrIsValid(first_record))
pg_fatal("could not find a valid record after %X/%08X",
LSN_FORMAT_ARGS(private.startptr));

Write Preview
Loading…
Cancel
Save