From 66b5a37a26613d830ead892411466329f5d53b6f Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Tue, 28 Oct 2025 06:08:37 +0000 Subject: [PATCH v1 09/20] make use of XLogRecPtrIsInvalid in logical.c --- src/backend/replication/logical/logical.c | 30 +++++++++++------------ 1 file changed, 15 insertions(+), 15 deletions(-) 100.0% src/backend/replication/logical/ diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 93ed2eb368e..295e83ad9d6 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -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(!XLogRecPtrIsInvalid(slot->data.restart_lsn)); - if (start_lsn == InvalidXLogRecPtr) + if (XLogRecPtrIsInvalid(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 (!XLogRecPtrIsInvalid(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 (XLogRecPtrIsInvalid(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(!XLogRecPtrIsInvalid(restart_lsn)); + Assert(!XLogRecPtrIsInvalid(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 (XLogRecPtrIsInvalid(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(!XLogRecPtrIsInvalid(lsn)); /* Do an unlocked check for candidate_lsn first. */ - if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr || - MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr) + if (!XLogRecPtrIsInvalid(MyReplicationSlot->candidate_xmin_lsn) || + !XLogRecPtrIsInvalid(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 (!XLogRecPtrIsInvalid(MyReplicationSlot->candidate_xmin_lsn) && MyReplicationSlot->candidate_xmin_lsn <= lsn) { /* @@ -1871,10 +1871,10 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn) } } - if (MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr && + if (!XLogRecPtrIsInvalid(MyReplicationSlot->candidate_restart_valid) && MyReplicationSlot->candidate_restart_valid <= lsn) { - Assert(MyReplicationSlot->candidate_restart_lsn != InvalidXLogRecPtr); + Assert(!XLogRecPtrIsInvalid(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(!XLogRecPtrIsInvalid(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 (!XLogRecPtrIsInvalid(ctx->reader->EndRecPtr)) { LogicalConfirmReceivedLocation(moveto); -- 2.34.1