public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 14/18] doc review: logical decode in prepare
9+ messages / 6 participants
[nested] [flat]
* [PATCH 14/18] doc review: logical decode in prepare
@ 2021-01-24 02:33 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Justin Pryzby @ 2021-01-24 02:33 UTC (permalink / raw)
a271a1b50e9bec07e2ef3a05e38e7285113e4ce6
---
doc/src/sgml/logicaldecoding.sgml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index cf705ed9cd..b854f2ccfc 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -1214,7 +1214,7 @@ stream_commit_cb(...); <-- commit of the streamed transaction
</para>
<para>
- When a prepared transaction is rollbacked using the
+ When a prepared transaction is rolled back using the
<command>ROLLBACK PREPARED</command>, then the
<function>rollback_prepared_cb</function> callback is invoked and when the
prepared transaction is committed using <command>COMMIT PREPARED</command>,
--
2.17.0
--lc9FT7cWel8HagAv
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0015-doc-review-for-multiranges.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* pgsql: Add ALTER SUBSCRIPTION ... SKIP.
@ 2022-03-22 01:56 Amit Kapila <[email protected]>
2022-03-22 23:59 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Andres Freund <[email protected]>
2022-09-04 08:18 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Alvaro Herrera <[email protected]>
0 siblings, 2 replies; 9+ messages in thread
From: Amit Kapila @ 2022-03-22 01:56 UTC (permalink / raw)
To: [email protected]
Add ALTER SUBSCRIPTION ... SKIP.
This feature allows skipping the transaction on subscriber nodes.
If incoming change violates any constraint, logical replication stops
until it's resolved. Currently, users need to either manually resolve the
conflict by updating a subscriber-side database or by using function
pg_replication_origin_advance() to skip the conflicting transaction. This
commit introduces a simpler way to skip the conflicting transactions.
The user can specify LSN by ALTER SUBSCRIPTION ... SKIP (lsn = XXX),
which allows the apply worker to skip the transaction finished at
specified LSN. The apply worker skips all data modification changes within
the transaction.
Author: Masahiko Sawada
Reviewed-by: Takamichi Osumi, Hou Zhijie, Peter Eisentraut, Amit Kapila, Shi Yu, Vignesh C, Greg Nancarrow, Haiying Tang, Euler Taveira
Discussion: https://postgr.es/m/CAD21AoDeScrsHhLyEPYqN3sydg6PxAPVBboK=30xJfUVihNZDA@mail.gmail.com
Branch
------
master
Details
-------
https://git.postgresql.org/pg/commitdiff/208c5d65bbd60e33e272964578cb74182ac726a8
Modified Files
--------------
doc/src/sgml/catalogs.sgml | 10 +
doc/src/sgml/logical-replication.sgml | 27 +--
doc/src/sgml/ref/alter_subscription.sgml | 42 +++++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 2 +-
src/backend/commands/subscriptioncmds.c | 73 ++++++++
src/backend/parser/gram.y | 9 +
src/backend/replication/logical/worker.c | 233 +++++++++++++++++++++++-
src/bin/pg_dump/pg_dump.c | 4 +
src/bin/psql/describe.c | 8 +-
src/bin/psql/tab-complete.c | 5 +-
src/include/catalog/catversion.h | 2 +-
src/include/catalog/pg_subscription.h | 5 +
src/include/nodes/parsenodes.h | 3 +-
src/test/regress/expected/subscription.out | 126 +++++++------
src/test/regress/sql/subscription.sql | 11 ++
src/test/subscription/t/029_disable_on_error.pl | 94 ----------
src/test/subscription/t/029_on_error.pl | 183 +++++++++++++++++++
18 files changed, 665 insertions(+), 173 deletions(-)
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
@ 2022-03-22 23:59 ` Andres Freund <[email protected]>
2022-03-23 01:56 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
1 sibling, 1 reply; 9+ messages in thread
From: Andres Freund @ 2022-03-22 23:59 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: [email protected]
Hi,
On 2022-03-22 01:56:03 +0000, Amit Kapila wrote:
> Add ALTER SUBSCRIPTION ... SKIP.
>
> This feature allows skipping the transaction on subscriber nodes.
>
> If incoming change violates any constraint, logical replication stops
> until it's resolved. Currently, users need to either manually resolve the
> conflict by updating a subscriber-side database or by using function
> pg_replication_origin_advance() to skip the conflicting transaction. This
> commit introduces a simpler way to skip the conflicting transactions.
>
> The user can specify LSN by ALTER SUBSCRIPTION ... SKIP (lsn = XXX),
> which allows the apply worker to skip the transaction finished at
> specified LSN. The apply worker skips all data modification changes within
> the transaction.
This was missing an include of xlogdefs.h in pg_subscription.h, thus failing
in headerscheck. See e.g.
https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=crake&dt=2022-03-22%2022%3A22%3A05
I've pushed the trivial fix for that. I'll propose adding headerscheck to CI /
cfbot.
Greetings,
Andres Freund
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-03-22 23:59 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Andres Freund <[email protected]>
@ 2022-03-23 01:56 ` Amit Kapila <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Amit Kapila @ 2022-03-23 01:56 UTC (permalink / raw)
To: Andres Freund <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-committers <[email protected]>
On Wed, Mar 23, 2022 at 5:29 AM Andres Freund <[email protected]> wrote:
>
> Hi,
>
> On 2022-03-22 01:56:03 +0000, Amit Kapila wrote:
> > Add ALTER SUBSCRIPTION ... SKIP.
> >
> > This feature allows skipping the transaction on subscriber nodes.
> >
> > If incoming change violates any constraint, logical replication stops
> > until it's resolved. Currently, users need to either manually resolve the
> > conflict by updating a subscriber-side database or by using function
> > pg_replication_origin_advance() to skip the conflicting transaction. This
> > commit introduces a simpler way to skip the conflicting transactions.
> >
> > The user can specify LSN by ALTER SUBSCRIPTION ... SKIP (lsn = XXX),
> > which allows the apply worker to skip the transaction finished at
> > specified LSN. The apply worker skips all data modification changes within
> > the transaction.
>
> This was missing an include of xlogdefs.h in pg_subscription.h, thus failing
> in headerscheck. See e.g.
> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=crake&dt=2022-03-22%2022%3A22%3A05
>
> I've pushed the trivial fix for that. I'll propose adding headerscheck to CI /
> cfbot.
>
Thanks.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
@ 2022-09-04 08:18 ` Alvaro Herrera <[email protected]>
2022-09-04 11:38 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
1 sibling, 1 reply; 9+ messages in thread
From: Alvaro Herrera @ 2022-09-04 08:18 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; +Cc: [email protected]
On 2022-Mar-22, Amit Kapila wrote:
> Add ALTER SUBSCRIPTION ... SKIP.
There are two messages here that seem oddly worded.
msgid "start skipping logical replication transaction finished at %X/%X"
msgid "done skipping logical replication transaction finished at %X/%X"
Two complaints here. First, the phrases "start / finished" and "done /
finished" look very strange. It took me a while to realize that
"finished" refers to the LSN, not to the skipping operation. Do we ever
talk about a transaction "finished at XYZ" as opposed to a transaction
whose LSN is XYZ? (This became particularly strange when I realized
that the LSN might come from a PREPARE.)
Second, "logical replication transaction". Is it not a regular
transaction that we happen to be processing via logical replication?
I think they should say something like
"logical replication starts skipping transaction with LSN %X/%X"
"logical replication completed skipping transaction with LSN %X/%X"
Other ideas?
--
Álvaro Herrera 48°01'N 7°57'E — https://www.EnterpriseDB.com/
Officer Krupke, what are we to do?
Gee, officer Krupke, Krup you! (West Side Story, "Gee, Officer Krupke")
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-04 08:18 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Alvaro Herrera <[email protected]>
@ 2022-09-04 11:38 ` Amit Kapila <[email protected]>
2022-09-06 14:10 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Masahiko Sawada <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Amit Kapila @ 2022-09-04 11:38 UTC (permalink / raw)
To: Alvaro Herrera <[email protected]>; +Cc: Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; [email protected]
On Sun, Sep 4, 2022 at 1:48 PM Alvaro Herrera <[email protected]> wrote:
>
> On 2022-Mar-22, Amit Kapila wrote:
>
> > Add ALTER SUBSCRIPTION ... SKIP.
>
> There are two messages here that seem oddly worded.
>
> msgid "start skipping logical replication transaction finished at %X/%X"
> msgid "done skipping logical replication transaction finished at %X/%X"
>
> Two complaints here. First, the phrases "start / finished" and "done /
> finished" look very strange. It took me a while to realize that
> "finished" refers to the LSN, not to the skipping operation. Do we ever
> talk about a transaction "finished at XYZ" as opposed to a transaction
> whose LSN is XYZ? (This became particularly strange when I realized
> that the LSN might come from a PREPARE.)
>
The reason to add "finished at ..." was to be explicit about whether
it is a starting LSN or an end LSN of a transaction. We do have such
differentiation in ReorderBufferTXN (first_lsn ... end_lsn).
> Second, "logical replication transaction". Is it not a regular
> transaction that we happen to be processing via logical replication?
>
> I think they should say something like
>
> "logical replication starts skipping transaction with LSN %X/%X"
> "logical replication completed skipping transaction with LSN %X/%X"
>
This looks better to me. If you find the above argument to
differentiate between the start and end LSN convincing then we can
think of replacing "with" in the above messages with "finished at". I
see your point related to using "finished at" for PREPARE may not be a
good idea but I don't have better ideas for the same.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-04 08:18 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Alvaro Herrera <[email protected]>
2022-09-04 11:38 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
@ 2022-09-06 14:10 ` Masahiko Sawada <[email protected]>
2022-09-07 03:46 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Masahiko Sawada @ 2022-09-06 14:10 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Alvaro Herrera <[email protected]>; Amit Kapila <[email protected]>; [email protected]
On Sun, Sep 4, 2022 at 8:39 PM Amit Kapila <[email protected]> wrote:
>
> On Sun, Sep 4, 2022 at 1:48 PM Alvaro Herrera <[email protected]> wrote:
> >
> > On 2022-Mar-22, Amit Kapila wrote:
> >
> > > Add ALTER SUBSCRIPTION ... SKIP.
> >
> > There are two messages here that seem oddly worded.
> >
> > msgid "start skipping logical replication transaction finished at %X/%X"
> > msgid "done skipping logical replication transaction finished at %X/%X"
> >
> > Two complaints here. First, the phrases "start / finished" and "done /
> > finished" look very strange. It took me a while to realize that
> > "finished" refers to the LSN, not to the skipping operation. Do we ever
> > talk about a transaction "finished at XYZ" as opposed to a transaction
> > whose LSN is XYZ? (This became particularly strange when I realized
> > that the LSN might come from a PREPARE.)
> >
>
> The reason to add "finished at ..." was to be explicit about whether
> it is a starting LSN or an end LSN of a transaction. We do have such
> differentiation in ReorderBufferTXN (first_lsn ... end_lsn).
>
> > Second, "logical replication transaction". Is it not a regular
> > transaction that we happen to be processing via logical replication?
> >
> > I think they should say something like
> >
> > "logical replication starts skipping transaction with LSN %X/%X"
> > "logical replication completed skipping transaction with LSN %X/%X"
> >
>
> This looks better to me.
+1
> If you find the above argument to
> differentiate between the start and end LSN convincing then we can
> think of replacing "with" in the above messages with "finished at". I
> see your point related to using "finished at" for PREPARE may not be a
> good idea but I don't have better ideas for the same.
Given that the user normally doesn't need to be aware of the
difference between start LSN and end LSN in the context of using this
feature, I think we can use "with LSN %X/%X".
Regards,
--
Masahiko Sawada
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP.
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-04 08:18 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Alvaro Herrera <[email protected]>
2022-09-04 11:38 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-06 14:10 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Masahiko Sawada <[email protected]>
@ 2022-09-07 03:46 ` Amit Kapila <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Amit Kapila @ 2022-09-07 03:46 UTC (permalink / raw)
To: Masahiko Sawada <[email protected]>; +Cc: Alvaro Herrera <[email protected]>; Amit Kapila <[email protected]>; [email protected]
On Tue, Sep 6, 2022 at 7:40 PM Masahiko Sawada <[email protected]> wrote:
>
> > > I think they should say something like
> > >
> > > "logical replication starts skipping transaction with LSN %X/%X"
> > > "logical replication completed skipping transaction with LSN %X/%X"
> > >
> >
> > This looks better to me.
>
> +1
>
> > If you find the above argument to
> > differentiate between the start and end LSN convincing then we can
> > think of replacing "with" in the above messages with "finished at". I
> > see your point related to using "finished at" for PREPARE may not be a
> > good idea but I don't have better ideas for the same.
>
> Given that the user normally doesn't need to be aware of the
> difference between start LSN and end LSN in the context of using this
> feature, I think we can use "with LSN %X/%X".
>
Fair enough.
Alvaro, would you like to push your proposed change? Otherwise, I am
happy to take care of this.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH va67 3/9] Support invalidating replication slots due to horizon and wal_level
@ 2023-04-07 16:32 Andres Freund <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Andres Freund @ 2023-04-07 16:32 UTC (permalink / raw)
Needed for supporting logical decoding on a standby. The new invalidation
methods will be used in a subsequent commit.
Author: "Drouvot, Bertrand" <[email protected]>
Author: Andres Freund <[email protected]>
Author: Amit Khandekar <[email protected]> (in an older version)
Reviewed-by: "Drouvot, Bertrand" <[email protected]>
Reviewed-by: Andres Freund <[email protected]>
Reviewed-by: Robert Haas <[email protected]>
Reviewed-by: Fabr�zio de Royes Mello <[email protected]>
Reviewed-by: Bharath Rupireddy <[email protected]>
Reviewed-by: Amit Kapila <[email protected]>
Discussion: https://postgr.es/m/[email protected]
---
src/include/replication/slot.h | 9 +-
src/backend/access/transam/xlog.c | 6 +-
src/backend/replication/logical/logical.c | 7 +
src/backend/replication/slot.c | 151 ++++++++++++++++++----
4 files changed, 144 insertions(+), 29 deletions(-)
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index ebcb637baed..bfc84193a7a 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -46,6 +46,10 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_NONE,
/* required WAL has been removed */
RS_INVAL_WAL,
+ /* required rows have been removed */
+ RS_INVAL_HORIZON,
+ /* wal_level insufficient for slot */
+ RS_INVAL_WAL_LEVEL,
} ReplicationSlotInvalidationCause;
/*
@@ -226,7 +230,10 @@ extern void ReplicationSlotsComputeRequiredLSN(void);
extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
extern void ReplicationSlotsDropDBSlots(Oid dboid);
-extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
+extern bool InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause cause,
+ XLogSegNo oldestSegno,
+ Oid dboid,
+ TransactionId snapshotConflictHorizon);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern int ReplicationSlotIndex(ReplicationSlot *slot);
extern bool ReplicationSlotName(int index, Name name);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 46821ad6056..1485e8f9ca9 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6806,7 +6806,8 @@ CreateCheckPoint(int flags)
*/
XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size);
KeepLogSeg(recptr, &_logSegNo);
- if (InvalidateObsoleteReplicationSlots(_logSegNo))
+ if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL, _logSegNo, InvalidOid,
+ InvalidTransactionId))
{
/*
* Some slots have been invalidated; recalculate the old-segment
@@ -7250,7 +7251,8 @@ CreateRestartPoint(int flags)
replayPtr = GetXLogReplayRecPtr(&replayTLI);
endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr;
KeepLogSeg(endptr, &_logSegNo);
- if (InvalidateObsoleteReplicationSlots(_logSegNo))
+ if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL, _logSegNo, InvalidOid,
+ InvalidTransactionId))
{
/*
* Some slots have been invalidated; recalculate the old-segment
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 85fc49f655d..27addd58f66 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -531,6 +531,13 @@ CreateDecodingContext(XLogRecPtr start_lsn,
NameStr(MyReplicationSlot->data.name)),
errdetail("This slot has been invalidated because it exceeded the maximum reserved size.")));
+ if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot read from logical replication slot \"%s\"",
+ NameStr(MyReplicationSlot->data.name)),
+ errdetail("This slot has been invalidated because it was conflicting with recovery.")));
+
Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
Assert(MyReplicationSlot->data.restart_lsn != InvalidXLogRecPtr);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index df23b7ed31e..c2a9accebf6 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1241,8 +1241,58 @@ ReplicationSlotReserveWal(void)
}
/*
- * Helper for InvalidateObsoleteReplicationSlots -- acquires the given slot
- * and mark it invalid, if necessary and possible.
+ * Report that replication slot needs to be invalidated
+ */
+static void
+ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
+ bool terminating,
+ int pid,
+ NameData slotname,
+ XLogRecPtr restart_lsn,
+ XLogRecPtr oldestLSN,
+ TransactionId snapshotConflictHorizon)
+{
+ StringInfoData err_detail;
+ bool hint = false;
+
+ initStringInfo(&err_detail);
+
+ switch (cause)
+ {
+ case RS_INVAL_WAL:
+ hint = true;
+ appendStringInfo(&err_detail, _("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes."),
+ LSN_FORMAT_ARGS(restart_lsn),
+ (unsigned long long) (oldestLSN - restart_lsn));
+ break;
+ case RS_INVAL_HORIZON:
+ appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
+ snapshotConflictHorizon);
+ break;
+
+ case RS_INVAL_WAL_LEVEL:
+ appendStringInfo(&err_detail, _("Logical decoding on standby requires wal_level to be at least logical on the primary server"));
+ break;
+ case RS_INVAL_NONE:
+ pg_unreachable();
+ }
+
+ ereport(LOG,
+ terminating ?
+ errmsg("terminating process %d to release replication slot \"%s\"",
+ pid, NameStr(slotname)) :
+ errmsg("invalidating obsolete replication slot \"%s\"",
+ NameStr(slotname)),
+ errdetail_internal("%s", err_detail.data),
+ hint ? errhint("You might need to increase max_slot_wal_keep_size.") : 0);
+
+ pfree(err_detail.data);
+}
+
+/*
+ * Helper for InvalidateObsoleteReplicationSlots
+ *
+ * Acquires the given slot and mark it invalid, if necessary and possible.
*
* Returns whether ReplicationSlotControlLock was released in the interim (and
* in that case we're not holding the lock at return, otherwise we are).
@@ -1253,7 +1303,10 @@ ReplicationSlotReserveWal(void)
* for syscalls, so caller must restart if we return true.
*/
static bool
-InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
+InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
+ ReplicationSlot *s,
+ XLogRecPtr oldestLSN,
+ Oid dboid, TransactionId snapshotConflictHorizon,
bool *invalidated)
{
int last_signaled_pid = 0;
@@ -1264,6 +1317,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
XLogRecPtr restart_lsn;
NameData slotname;
int active_pid = 0;
+ ReplicationSlotInvalidationCause conflict = RS_INVAL_NONE;
Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
@@ -1286,10 +1340,45 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
restart_lsn = s->data.restart_lsn;
/*
- * If the slot is already invalid or is fresh enough, we don't need to
- * do anything.
+ * If the slot is already invalid or is a non conflicting slot, we
+ * don't need to do anything.
*/
- if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
+ if (s->data.invalidated == RS_INVAL_NONE)
+ {
+ switch (cause)
+ {
+ case RS_INVAL_WAL:
+ if (s->data.restart_lsn != InvalidXLogRecPtr &&
+ s->data.restart_lsn < oldestLSN)
+ conflict = cause;
+ break;
+ case RS_INVAL_HORIZON:
+ if (!SlotIsLogical(s))
+ break;
+ /* invalid DB oid signals a shared relation */
+ if (dboid != InvalidOid && dboid != s->data.database)
+ break;
+ if (TransactionIdIsValid(s->effective_xmin) &&
+ TransactionIdPrecedesOrEquals(s->effective_xmin,
+ snapshotConflictHorizon))
+ conflict = cause;
+ else if (TransactionIdIsValid(s->effective_catalog_xmin) &&
+ TransactionIdPrecedesOrEquals(s->effective_catalog_xmin,
+ snapshotConflictHorizon))
+ conflict = cause;
+ break;
+ case RS_INVAL_WAL_LEVEL:
+ if (SlotIsLogical(s))
+ conflict = cause;
+ break;
+ default:
+ pg_unreachable();
+ break;
+ }
+ }
+
+ /* if there's no conflict, we're done */
+ if (conflict == RS_INVAL_NONE)
{
SpinLockRelease(&s->mutex);
if (released_lock)
@@ -1309,13 +1398,14 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
{
MyReplicationSlot = s;
s->active_pid = MyProcPid;
- s->data.invalidated = RS_INVAL_WAL;
+ s->data.invalidated = conflict;
/*
* XXX: We should consider not overwriting restart_lsn and instead
* just rely on .invalidated.
*/
- s->data.restart_lsn = InvalidXLogRecPtr;
+ if (conflict == RS_INVAL_WAL)
+ s->data.restart_lsn = InvalidXLogRecPtr;
/* Let caller know */
*invalidated = true;
@@ -1349,13 +1439,9 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
*/
if (last_signaled_pid != active_pid)
{
- ereport(LOG,
- errmsg("terminating process %d to release replication slot \"%s\"",
- active_pid, NameStr(slotname)),
- errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
- LSN_FORMAT_ARGS(restart_lsn),
- (unsigned long long) (oldestLSN - restart_lsn)),
- errhint("You might need to increase max_slot_wal_keep_size."));
+ ReportSlotInvalidation(conflict, true, active_pid,
+ slotname, restart_lsn,
+ oldestLSN, snapshotConflictHorizon);
(void) kill(active_pid, SIGTERM);
last_signaled_pid = active_pid;
@@ -1390,14 +1476,11 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
ReplicationSlotMarkDirty();
ReplicationSlotSave();
ReplicationSlotRelease();
+ pgstat_drop_replslot(s);
- ereport(LOG,
- errmsg("invalidating obsolete replication slot \"%s\"",
- NameStr(slotname)),
- errdetail("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
- LSN_FORMAT_ARGS(restart_lsn),
- (unsigned long long) (oldestLSN - restart_lsn)),
- errhint("You might need to increase max_slot_wal_keep_size."));
+ ReportSlotInvalidation(conflict, false, active_pid,
+ slotname, restart_lsn,
+ oldestLSN, snapshotConflictHorizon);
/* done with this slot for now */
break;
@@ -1410,19 +1493,33 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
}
/*
- * Mark any slot that points to an LSN older than the given segment
- * as invalid; it requires WAL that's about to be removed.
+ * Invalidate slots that require resources about to be removed.
*
* Returns true when any slot have got invalidated.
*
+ * Whether a slot needs to be invalidated depends on the cause. A slot is
+ * removed if it:
+ * - RS_INVAL_WAL: requires a LSN older than the given segment
+ * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon, in the given db
+ dboid may be InvalidOid for shared relations
+ * - RS_INVAL_WAL_LEVEL: is logical
+ *
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
bool
-InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno)
+InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause cause,
+ XLogSegNo oldestSegno, Oid dboid,
+ TransactionId snapshotConflictHorizon)
{
XLogRecPtr oldestLSN;
bool invalidated = false;
+ Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
+ Assert(cause != RS_INVAL_WAL || oldestSegno > 0);
+
+ if (max_replication_slots == 0)
+ return invalidated;
+
XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
restart:
@@ -1434,7 +1531,9 @@ restart:
if (!s->in_use)
continue;
- if (InvalidatePossiblyObsoleteSlot(s, oldestLSN, &invalidated))
+ if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
+ snapshotConflictHorizon,
+ &invalidated))
{
/* if the lock was released, start from scratch */
goto restart;
--
2.38.0
--syfz2cpck3lfbybq
Content-Type: text/x-diff; charset=iso-8859-1
Content-Disposition: attachment;
filename="va67-0004-Handle-logical-slot-conflicts-on-standby.patch"
Content-Transfer-Encoding: 8bit
^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2023-04-07 16:32 UTC | newest]
Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-01-24 02:33 [PATCH 14/18] doc review: logical decode in prepare Justin Pryzby <[email protected]>
2022-03-22 01:56 pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-03-22 23:59 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Andres Freund <[email protected]>
2022-03-23 01:56 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-04 08:18 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Alvaro Herrera <[email protected]>
2022-09-04 11:38 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2022-09-06 14:10 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Masahiko Sawada <[email protected]>
2022-09-07 03:46 ` Re: pgsql: Add ALTER SUBSCRIPTION ... SKIP. Amit Kapila <[email protected]>
2023-04-07 16:32 [PATCH va67 3/9] Support invalidating replication slots due to horizon and wal_level Andres Freund <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox