public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
To: Ashutosh Bapat <[email protected]>
Cc: Fabrice Chapuis <[email protected]>
Cc: Euler Taveira <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Simon Riggs <[email protected]>
Cc: Petr Jelinek <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Ajin Cherian <[email protected]>
Subject: RE: Logical replication timeout problem
Date: Wed, 18 Jan 2023 08:19:08 +0000
Message-ID: <OS3PR01MB6275E5C6DC1E9E7CD2D11E039EC79@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1Jvfc+C3zj++crKVQ6h4xeME6EJO77Ccmv6dZ=9HPsY5w@mail.gmail.com>
References: <CAA4eK1+fQjndoBOFUn9Wy0hhm3MLyUWEpcT9O7iuCELktfdBiQ@mail.gmail.com>
<CAA4eK1LGnaPuWs2M4sDfpd6JQZjoh4DGAsgUvNW=Or8i9z6K8w@mail.gmail.com>
<OS3PR01MB62756599B78FA7D4C908109A9E179@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1JKZrOfCLMVZYfUnm_wfxucDjOzeGwQy3hJ_U9Y3AV05Q@mail.gmail.com>
<OS3PR01MB627542044A4C16511E63D5DA9E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAD21AoCEhZ6D1VtqvxK1vO0=NnJd6phWbijt-w=M5hwka=-c1A@mail.gmail.com>
<OS3PR01MB62750C20AD7AF4F13541FC109E1A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<TYAPR01MB586636517DED745651D6E75AF51D9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<OS3PR01MB6275C64F264662E84D2FB7AE9E1D9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275F0E4E27F5522847589ED9E1E9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275E0C2B4D9E488AD7CBA209E1F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1JUgO9F5fZb5P8quMYfDbD7SBNApsTsGLf6vYNNetpWTA@mail.gmail.com>
<CAD21AoCQi33mjd=bM0Jhkk3QaV-fHVB_soA7iQmTor2A-1Cx3Q@mail.gmail.com>
<[email protected]>
<CAA4eK1+wSvfDVUSe-4o6APFdO8=9G-HnZJ+01_LQOxqLmhqe5Q@mail.gmail.com>
<[email protected]>
<CAA4eK1+HNmgtWRm2U7EG2T0yWbAvhqSDDFBmKX5+vSgBsRmY=w@mail.gmail.com>
<CAA5-nLC9HdJPgCSOcVCZ7YMmhv3nLgGbReMB5jyEGbXA6VVsyg@mail.gmail.com>
<OS3PR01MB6275478E5D29E4A563302D3D9E2B9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA5-nLAqDq3+7Qo5xKf4uuQz6VZH--NFCcquAYa8_nhf3j3gig@mail.gmail.com>
<OS3PR01MB62751A8063A9A75A096000D89E3F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAExHW5uALmiA4ZHSAdVwjsqiUCRzy+BjKr4RS_MpDomHuvJcLA@mail.gmail.com>
<CAA4eK1KYbj+-=2ZdhqcW9m+xS9dHHKc1e5f2LG=R+Qg2zWbNcw@mail.gmail.com>
<OS3PR01MB6275D9F418DB8EAFB894373F9EFC9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAExHW5vPNnnA-zYDJGFH3OcA2J1BwTBKvE_5D+Z9d9n95suSXw@mail.gmail.com>
<CAA4eK1JWCzx2CTxExbxk6X0fHm1vrWbXQcJv2rbfwC42bmgQJg@mail.gmail.com>
<CAExHW5sOvWbXwMCpeWrE+Jfu=Xbi4y5AkrNedSPzELYq43VDYw@mail.gmail.com>
<CAA4eK1Jvfc+C3zj++crKVQ6h4xeME6EJO77Ccmv6dZ=9HPsY5w@mail.gmail.com>
On Wed, Jan 18, 2023 at 13:29 PM Amit Kapila <[email protected]> wrote:
> On Tue, Jan 17, 2023 at 6:41 PM Ashutosh Bapat
> <[email protected]> wrote:
> >
> > On Tue, Jan 17, 2023 at 3:34 PM Amit Kapila <[email protected]> wrote:
> >
> > > >
> > > > I am a bit worried about the indirections that the wrappers and hooks
> > > > create. Output plugins call OutputPluginUpdateProgress() in callbacks
> > > > but I don't see why ReorderBufferProcessTXN() needs a callback to
> > > > call OutputPluginUpdateProgress.
> > > >
> > >
> > > Yeah, I think we can do it as we are doing the previous approach but
> > > we need an additional wrapper (update_progress_cb_wrapper()) as the
> > > current patch has so that we can add error context information. This
> > > is similar to why we have a wrapper for all other callbacks like
> > > change_cb_wrapper.
> > >
> >
> > Ultimately OutputPluginUpdateProgress() will be called - which in turn
> > will call ctx->update_progress.
> >
>
> No, update_progress_cb_wrapper() should directly call
> ctx->update_progress(). The key reason to have a
> update_progress_cb_wrapper() is that it allows us to add error context
> information (see the usage of output_plugin_error_callback).
I think it makes sense. This also avoids the need for every output plugin to
implement the callback. So I tried to improve the patch based on this approach.
And I tried to add some comments for this new callback to distinguish it from
ctx->update_progress.
Attach the new patch.
Regards,
Wang Wei
Attachments:
[application/octet-stream] v3-0001-Fix-the-logical-replication-timeout-during-proces.patch (11.5K, ../OS3PR01MB6275E5C6DC1E9E7CD2D11E039EC79@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v3-0001-Fix-the-logical-replication-timeout-during-proces.patch)
download | inline diff:
From fe7015a8a000ecee60c2498e871a5a70c9eacddf Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Wed, 18 Jan 2023 13:45:32 +0800
Subject: [PATCH v3] Fix the logical replication timeout during processing of
DDL.
The problem is when there is a DDL in a transaction that generates lots of
temporary data due to rewrite rules, these temporary data will not be processed
by the pgoutput - plugin. Therefore, the previous fix (f95d53e) for DML had no
impact on this case.
To fix this, we introduced a new ReorderBuffer callback -
'ReorderBufferUpdateProgressCB'. This callback is called to try to update the
process after each change has been processed during sending data of a
transaction (and its subtransactions) to the output plugin.
---
src/backend/replication/logical/logical.c | 83 +++++++++++++++++++
.../replication/logical/reorderbuffer.c | 2 +
src/backend/replication/pgoutput/pgoutput.c | 54 ++----------
src/include/replication/reorderbuffer.h | 12 +++
src/tools/pgindent/typedefs.list | 1 +
5 files changed, 104 insertions(+), 48 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 1a58dd7649..a16f40e8ae 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -93,6 +93,11 @@ static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *tx
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
int nrelations, Relation relations[], ReorderBufferChange *change);
+/* update progress callback */
+static void update_progress_cb_wrapper(ReorderBuffer *cache,
+ ReorderBufferTXN *txn,
+ ReorderBufferChange *change);
+
static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin);
/*
@@ -278,6 +283,12 @@ StartupDecodingContext(List *output_plugin_options,
ctx->reorder->commit_prepared = commit_prepared_cb_wrapper;
ctx->reorder->rollback_prepared = rollback_prepared_cb_wrapper;
+ /*
+ * Callback to support updating progress during sending data of a
+ * transaction (and its subtransactions) to the output plugin.
+ */
+ ctx->reorder->update_progress = update_progress_cb_wrapper;
+
ctx->out = makeStringInfo();
ctx->prepare_write = prepare_write;
ctx->write = do_write;
@@ -1584,6 +1595,78 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}
+/*
+ * Update progress callback
+ *
+ * Try to update progress and send a keepalive message if too many changes were
+ * processed when processing txn.
+ *
+ * For a large transaction, if we don't send any change to the downstream for a
+ * long time (exceeds the wal_receiver_timeout of standby) then it can timeout.
+ * This can happen when all or most of the changes are either not published or
+ * got filtered out.
+ */
+static void
+update_progress_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ ReorderBufferChange *change)
+{
+ LogicalDecodingContext *ctx = cache->private_data;
+ LogicalErrorCallbackState state;
+ ErrorContextCallback errcallback;
+ static int changes_count = 0; /* Static variable used to accumulate
+ * the number of changes while
+ * processing txn. */
+
+ Assert(!ctx->fast_forward);
+
+ if (!ctx->update_progress)
+ return;
+
+ /* Push callback + info on the error context stack */
+ state.ctx = ctx;
+ state.callback_name = "update_progress";
+ state.report_location = change->lsn;
+ errcallback.callback = output_plugin_error_callback;
+ errcallback.arg = (void *) &state;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /* set output state */
+ ctx->accept_writes = false;
+ ctx->write_xid = txn->xid;
+
+ /*
+ * Report this change's lsn so replies from clients can give an up-to-date
+ * answer. This won't ever be enough (and shouldn't be!) to confirm
+ * receipt of this transaction, but it might allow another transaction's
+ * commit to be confirmed with one message.
+ */
+ ctx->write_location = change->lsn;
+
+ ctx->end_xact = false;
+
+ /*
+ * We don't want to try sending a keepalive message after processing each
+ * change as that can have overhead. Tests revealed that there is no
+ * noticeable overhead in doing it after continuously processing 100 or so
+ * changes.
+ */
+#define CHANGES_THRESHOLD 100
+
+ /*
+ * After continuously processing CHANGES_THRESHOLD changes, we
+ * try to send a keepalive message if required.
+ */
+ if (++changes_count >= CHANGES_THRESHOLD)
+ {
+ ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, false);
+ changes_count = 0;
+ }
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+}
+
/*
* Set the required catalog xmin horizon for historic snapshots in the current
* replication slot.
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 54ee824e6c..2e134bd011 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2446,6 +2446,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
elog(ERROR, "tuplecid value in changequeue");
break;
}
+
+ rb->update_progress(rb, txn, change);
}
/* speculative insertion record must be freed by now */
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 1a80d67bb9..0ba6b6b39c 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -92,8 +92,6 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid,
static void send_repl_origin(LogicalDecodingContext *ctx,
RepOriginId origin_id, XLogRecPtr origin_lsn,
bool send_origin);
-static void update_replication_progress(LogicalDecodingContext *ctx,
- bool skipped_xact);
/*
* Only 3 publication actions are used for row filtering ("insert", "update",
@@ -586,7 +584,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
* from this transaction has been sent to the downstream.
*/
sent_begin_txn = txndata->sent_begin_txn;
- update_replication_progress(ctx, !sent_begin_txn);
+ OutputPluginUpdateProgress(ctx, !sent_begin_txn);
pfree(txndata);
txn->output_plugin_private = NULL;
@@ -625,7 +623,7 @@ static void
pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
{
- update_replication_progress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
@@ -639,7 +637,7 @@ static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr commit_lsn)
{
- update_replication_progress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn);
@@ -655,7 +653,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
XLogRecPtr prepare_end_lsn,
TimestampTz prepare_time)
{
- update_replication_progress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn,
@@ -1401,8 +1399,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
- update_replication_progress(ctx, false);
-
if (!is_publishable_relation(relation))
return;
@@ -1637,8 +1633,6 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Oid *relids;
TransactionId xid = InvalidTransactionId;
- update_replication_progress(ctx, false);
-
/* Remember the xid for the change in streaming mode. See pgoutput_change. */
if (in_streaming)
xid = change->txn->xid;
@@ -1702,8 +1696,6 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
TransactionId xid = InvalidTransactionId;
- update_replication_progress(ctx, false);
-
if (!data->messages)
return;
@@ -1903,7 +1895,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
Assert(!in_streaming);
Assert(rbtxn_is_streamed(txn));
- update_replication_progress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_commit(ctx->out, txn, commit_lsn);
@@ -1924,7 +1916,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
{
Assert(rbtxn_is_streamed(txn));
- update_replication_progress(ctx, false);
+ OutputPluginUpdateProgress(ctx, false);
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
@@ -2424,37 +2416,3 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id,
}
}
}
-
-/*
- * Try to update progress and send a keepalive message if too many changes were
- * processed.
- *
- * For a large transaction, if we don't send any change to the downstream for a
- * long time (exceeds the wal_receiver_timeout of standby) then it can timeout.
- * This can happen when all or most of the changes are either not published or
- * got filtered out.
- */
-static void
-update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact)
-{
- static int changes_count = 0;
-
- /*
- * We don't want to try sending a keepalive message after processing each
- * change as that can have overhead. Tests revealed that there is no
- * noticeable overhead in doing it after continuously processing 100 or so
- * changes.
- */
-#define CHANGES_THRESHOLD 100
-
- /*
- * If we are at the end of transaction LSN, update progress tracking.
- * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we
- * try to send a keepalive message if required.
- */
- if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD)
- {
- OutputPluginUpdateProgress(ctx, skipped_xact);
- changes_count = 0;
- }
-}
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index f6c4dd75db..6dfb2c971e 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -525,6 +525,12 @@ typedef void (*ReorderBufferStreamTruncateCB) (
Relation relations[],
ReorderBufferChange *change);
+/* update progress callback signature */
+typedef void (*ReorderBufferUpdateProgressCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ ReorderBufferChange *change);
+
struct ReorderBuffer
{
/*
@@ -588,6 +594,12 @@ struct ReorderBuffer
ReorderBufferStreamMessageCB stream_message;
ReorderBufferStreamTruncateCB stream_truncate;
+ /*
+ * Callback to be called when updating progress during sending data of a
+ * transaction (and its subtransactions) to the output plugin.
+ */
+ ReorderBufferUpdateProgressCB update_progress;
+
/*
* Pointer that will be passed untouched to the callbacks.
*/
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 23bafec5f7..a1721d13bb 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2310,6 +2310,7 @@ ReorderBufferToastEnt
ReorderBufferTupleBuf
ReorderBufferTupleCidEnt
ReorderBufferTupleCidKey
+ReorderBufferUpdateProgressCB
ReorderTuple
RepOriginId
ReparameterizeForeignPathByChild_function
--
2.23.0.windows.1
view thread (10+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Logical replication timeout problem
In-Reply-To: <OS3PR01MB6275E5C6DC1E9E7CD2D11E039EC79@OS3PR01MB6275.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox