public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Dilip Kumar <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Tue, 10 Jan 2023 04:55:55 +0000
Message-ID: <OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1KK7Xty7SaCReY_b5moPURowMGOsfvhOb6xQ1EAaJAh6A@mail.gmail.com>
References: <CAA4eK1LGmZsevrqJra0V4O8oBU_eKyzm2VMpSAYQaDgC6n4fkA@mail.gmail.com>
<CAFiTN-uhnJkrUsQByBHMK268T-GEx_D8DQ6b2T+aW6RiU75pbQ@mail.gmail.com>
<OS0PR01MB571621ED532C2D7C3E01625894FB9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAFiTN-t5+_zTf86EzgCwObs1ED-P_5Ab1KMw340d3oPY_+Cwpw@mail.gmail.com>
<OS3PR01MB5718AE486227CE3ACB844C5E94F89@OS3PR01MB5718.jpnprd01.prod.outlook.com>
<CAFiTN-sgQ-K9BnMWKLR7Hm36QNjS7ZOBF6hsXJmQz5ChC0Desw@mail.gmail.com>
<CAA4eK1+P9WOFBupaTZCzYtranSH=8TSUYV=kk5xQJ-VNKnvgmQ@mail.gmail.com>
<OS0PR01MB571685A645E509267AC5F30894F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57168A5A1C31BC316CB63D3C94F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KK7Xty7SaCReY_b5moPURowMGOsfvhOb6xQ1EAaJAh6A@mail.gmail.com>
On Monday, January 9, 2023 4:51 PM Amit Kapila <[email protected]> wrote:
>
> On Sun, Jan 8, 2023 at 11:32 AM [email protected]
> <[email protected]> wrote:
> >
> > On Sunday, January 8, 2023 11:59 AM [email protected]
> <[email protected]> wrote:
> > > Attach the updated patch set.
> >
> > Sorry, the commit message of 0001 was accidentally deleted, just
> > attach the same patch set again with commit message.
> >
>
> Pushed the first (0001) patch.
Thanks for pushing, here are the remaining patches.
I reordered the patch number to put patches that are easier to
commit in the front of others.
Best regards,
Hou zj
Attachments:
[application/octet-stream] v78-0004-Retry-to-apply-streaming-xact-only-in-apply-work.patch (21.1K, ../OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v78-0004-Retry-to-apply-streaming-xact-only-in-apply-work.patch)
download | inline diff:
From c1cfd9f8ed4fcc162367a6185a29f7639d151fa8 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Sun, 11 Dec 2022 19:16:34 +0800
Subject: [PATCH v78 4/4] Retry to apply streaming xact only in apply worker
When the subscription parameter is set streaming=parallel, the logic tries to
apply the streaming transaction using a parallel apply worker. If this
fails the parallel worker exits with an error.
In this case, retry applying the streaming transaction using the normal
streaming=on mode. This is done to avoid getting caught in a loop of the same
retry errors.
A new flag field "subretry" has been introduced to catalog "pg_subscription".
If there are any active parallel apply workers and the subscriber exits with an
error, this flag will be set true, and whenever the transaction is applied
successfully, this flag is reset false.
Now, when deciding how to apply a streaming transaction, the logic can know if
this transaction has previously failed or not (by checking the "subretry"
field).
Note: Since we add a new field 'subretry' to catalog 'pg_subscription' has been
expanded, we need bump catalog version.
---
doc/src/sgml/catalogs.sgml | 10 ++
doc/src/sgml/logical-replication.sgml | 11 +-
doc/src/sgml/ref/create_subscription.sgml | 5 +
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 2 +-
src/backend/commands/subscriptioncmds.c | 1 +
.../replication/logical/applyparallelworker.c | 30 ++++
src/backend/replication/logical/worker.c | 182 +++++++++++++++------
src/bin/pg_dump/pg_dump.c | 5 +-
src/include/catalog/pg_subscription.h | 7 +
src/include/replication/worker_internal.h | 2 +
src/test/subscription/t/015_stream.pl | 63 +++++++
12 files changed, 263 insertions(+), 56 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index c1e4048..d918327 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7950,6 +7950,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<row>
<entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subretry</structfield> <type>bool</type>
+ </para>
+ <para>
+ True if previous change failed to be applied while there were any
+ active parallel apply workers, necessitating a retry.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
</para>
<para>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 8029fab..6254388 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1504,12 +1504,11 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<para>
When the streaming mode is <literal>parallel</literal>, the finish LSN of
- failed transactions may not be logged. In that case, it may be necessary to
- change the streaming mode to <literal>on</literal> or <literal>off</literal> and
- cause the same conflicts again so the finish LSN of the failed transaction will
- be written to the server log. For the usage of finish LSN, please refer to <link
- linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
- SKIP</command></link>.
+ failed transactions may not be logged. In that case, the failed transaction
+ will be retried in <literal>streaming = on</literal> mode. If it fails
+ again, the finish LSN of the failed transaction will be written to the
+ server log. For the usage of finish LSN, please refer to
+ <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ... SKIP</command></link>.
</para>
</sect1>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index eba72c6..decd554 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -251,6 +251,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
transaction is committed. Note that if an error happens in a
parallel apply worker, the finish LSN of the remote transaction
might not be reported in the server log.
+ When applying streaming transactions, if a deadlock is detected, the
+ parallel apply worker will exit with an error. The
+ <literal>parallel</literal> mode is disregarded when retrying;
+ instead the transaction will be applied using <literal>on</literal>
+ mode.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index a56ae31..0eb5ffb 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->retry = subform->subretry;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 72e46e5..aef926a 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1302,7 +1302,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subslotname, subsynccommit, subpublications, suborigin)
+ subretry, subslotname, subsynccommit, subpublications, suborigin)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index baff00d..b2053f7 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -636,6 +636,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED);
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
+ values[Anum_pg_subscription_subretry - 1] = BoolGetDatum(false);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index a7ce57b..e14785a 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -315,6 +315,17 @@ pa_can_start(void)
if (!AllTablesyncsReady())
return false;
+ /*
+ * Don't use parallel apply workers for retries, because it is possible
+ * that a deadlock was detected the last time we tried to apply a
+ * transaction using a parallel apply worker.
+ */
+ if (MySubscription->retry)
+ {
+ elog(DEBUG1, "parallel apply workers are not used for retries");
+ return false;
+ }
+
return true;
}
@@ -1678,3 +1689,22 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
pa_free_worker(winfo);
}
+
+/* Check if any active parallel apply workers. */
+bool
+pa_have_active_worker(void)
+{
+ ListCell *lc;
+
+ foreach(lc, ParallelApplyWorkerPool)
+ {
+ ParallelApplyWorkerInfo *tmp_winfo;
+
+ tmp_winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
+
+ if (tmp_winfo->in_use)
+ return true;
+ }
+
+ return false;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5c8ce97..0c7bf67 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -394,7 +394,7 @@ static void stream_close_file(void);
static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
-static void DisableSubscriptionAndExit(void);
+static void DisableSubscriptionOnError(void);
static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
static void apply_handle_insert_internal(ApplyExecutionData *edata,
@@ -431,6 +431,8 @@ static inline void reset_apply_error_context_info(void);
static TransApplyAction get_transaction_apply_action(TransactionId xid,
ParallelApplyWorkerInfo **winfo);
+static void set_subscription_retry(bool retry);
+
/*
* Return the name of the logical replication worker.
*/
@@ -1046,6 +1048,9 @@ apply_handle_commit(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
@@ -1152,6 +1157,9 @@ apply_handle_prepare(StringInfo s)
in_remote_transaction = false;
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1208,6 +1216,9 @@ apply_handle_commit_prepared(StringInfo s)
store_flush_position(prepare_data.end_lsn, XactLastCommitEnd);
in_remote_transaction = false;
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1269,6 +1280,9 @@ apply_handle_rollback_prepared(StringInfo s)
store_flush_position(rollback_data.rollback_end_lsn, XactLastCommitEnd);
in_remote_transaction = false;
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(rollback_data.rollback_end_lsn);
@@ -1394,6 +1408,9 @@ apply_handle_stream_prepare(StringInfo s)
break;
}
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
pgstat_report_stat(false);
/* Process any tables that are being synchronized in parallel. */
@@ -1970,6 +1987,10 @@ apply_handle_stream_abort(StringInfo s)
break;
}
+ /* Reset the retry flag. */
+ if (toplevel_xact)
+ set_subscription_retry(false);
+
reset_apply_error_context_info();
}
@@ -2252,6 +2273,9 @@ apply_handle_stream_commit(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
+ /* Reset the retry flag. */
+ set_subscription_retry(false);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
@@ -4348,20 +4372,28 @@ start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
}
PG_CATCH();
{
+ /*
+ * Emit the error message, and recover from the error state to an idle
+ * state
+ */
+ HOLD_INTERRUPTS();
+
+ EmitErrorReport();
+ AbortOutOfAnyTransaction();
+ FlushErrorState();
+
+ RESUME_INTERRUPTS();
+
+ /* Report the worker failed during table synchronization */
+ pgstat_report_subscription_error(MySubscription->oid, false);
+
if (MySubscription->disableonerr)
- DisableSubscriptionAndExit();
- else
- {
- /*
- * Report the worker failed during table synchronization. Abort
- * the current transaction so that the stats message is sent in an
- * idle state.
- */
- AbortOutOfAnyTransaction();
- pgstat_report_subscription_error(MySubscription->oid, false);
+ DisableSubscriptionOnError();
- PG_RE_THROW();
- }
+ /* Set the retry flag. */
+ set_subscription_retry(true);
+
+ proc_exit(0);
}
PG_END_TRY();
@@ -4386,20 +4418,27 @@ start_apply(XLogRecPtr origin_startpos)
}
PG_CATCH();
{
+ /*
+ * Emit the error message, and recover from the error state to an idle
+ * state
+ */
+ HOLD_INTERRUPTS();
+
+ EmitErrorReport();
+ AbortOutOfAnyTransaction();
+ FlushErrorState();
+
+ RESUME_INTERRUPTS();
+
+ /* Report the worker failed while applying changes */
+ pgstat_report_subscription_error(MySubscription->oid,
+ !am_tablesync_worker());
+
if (MySubscription->disableonerr)
- DisableSubscriptionAndExit();
- else
- {
- /*
- * Report the worker failed while applying changes. Abort the
- * current transaction so that the stats message is sent in an
- * idle state.
- */
- AbortOutOfAnyTransaction();
- pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+ DisableSubscriptionOnError();
- PG_RE_THROW();
- }
+ /* Set the retry flag. */
+ set_subscription_retry(true);
}
PG_END_TRY();
}
@@ -4675,39 +4714,20 @@ ApplyWorkerMain(Datum main_arg)
}
/*
- * After error recovery, disable the subscription in a new transaction
- * and exit cleanly.
+ * Disable the subscription in a new transaction.
*/
static void
-DisableSubscriptionAndExit(void)
+DisableSubscriptionOnError(void)
{
- /*
- * Emit the error message, and recover from the error state to an idle
- * state
- */
- HOLD_INTERRUPTS();
-
- EmitErrorReport();
- AbortOutOfAnyTransaction();
- FlushErrorState();
-
- RESUME_INTERRUPTS();
-
- /* Report the worker failed during either table synchronization or apply */
- pgstat_report_subscription_error(MyLogicalRepWorker->subid,
- !am_tablesync_worker());
-
/* Disable the subscription */
StartTransactionCommand();
DisableSubscription(MySubscription->oid);
CommitTransactionCommand();
- /* Notify the subscription has been disabled and exit */
+ /* Notify the subscription has been disabled */
ereport(LOG,
errmsg("subscription \"%s\" has been disabled because of an error",
MySubscription->name));
-
- proc_exit(0);
}
/*
@@ -5050,3 +5070,71 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
return TRANS_LEADER_SEND_TO_PARALLEL;
}
}
+
+/*
+ * Set subretry of pg_subscription catalog.
+ *
+ * If retry is true, subscriber is about to exit with an error. Otherwise, it
+ * means that the transaction was applied successfully.
+ */
+static void
+set_subscription_retry(bool retry)
+{
+ Relation rel;
+ HeapTuple tup;
+ bool started_tx = false;
+ bool nulls[Natts_pg_subscription];
+ bool replaces[Natts_pg_subscription];
+ Datum values[Natts_pg_subscription];
+
+ /* Fast path - if no state change then nothing to do */
+ if (MySubscription->retry == retry)
+ return;
+
+ /* Fast path - skip for parallel apply workers */
+ if (am_parallel_apply_worker())
+ return;
+
+ /* Fast path - skip set retry if no active parallel apply workers */
+ if (retry && !pa_have_active_worker())
+ return;
+
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
+
+ /* Look up the subscription in the catalog */
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+ tup = SearchSysCacheCopy1(SUBSCRIPTIONOID,
+ ObjectIdGetDatum(MySubscription->oid));
+
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "subscription \"%s\" does not exist", MySubscription->name);
+
+ LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
+ AccessShareLock);
+
+ /* Form a new tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ /* Set subretry */
+ values[Anum_pg_subscription_subretry - 1] = BoolGetDatum(retry);
+ replaces[Anum_pg_subscription_subretry - 1] = true;
+
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+
+ /* Update the catalog. */
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+ /* Cleanup. */
+ heap_freetuple(tup);
+ table_close(rel, NoLock);
+
+ if (started_tx)
+ CommitTransactionCommand();
+}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 5e800dc..931233f 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4560,8 +4560,9 @@ getSubscriptions(Archive *fout)
ntups = PQntuples(res);
/*
- * Get subscription fields. We don't include subskiplsn in the dump as
- * after restoring the dump this value may no longer be relevant.
+ * Get subscription fields. We don't include subskiplsn and subretry in
+ * the dump as after restoring the dump this value may no longer be
+ * relevant.
*/
i_tableoid = PQfnumber(res, "tableoid");
i_oid = PQfnumber(res, "oid");
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index b0f2a17..8edbbca 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -88,6 +88,11 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
+ bool subretry BKI_DEFAULT(f); /* True if previous change failed
+ * to be applied while there were
+ * any active parallel apply
+ * workers */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -131,6 +136,8 @@ typedef struct Subscription
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
* occurs */
+ bool retry; /* Indicates if previous change failed to be
+ * applied using a parallel apply worker */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index f3b2f2d..42c33a4 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -308,6 +308,8 @@ extern void pa_decr_and_wait_stream_block(void);
extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
XLogRecPtr remote_lsn);
+extern bool pa_have_active_worker(void);
+
#define isParallelApplyWorker(worker) ((worker)->apply_leader_pid != InvalidPid)
static inline bool
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 83d6956..4054a5f 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -74,8 +74,16 @@ sub test_streaming
'check extra columns contain local defaults');
# Test the streaming in binary mode
+ my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ );
$node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tap_sub SET (binary = on)");
+ $node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
# Check the subscriber log from now on.
$offset = -s $node_subscriber->logfile;
@@ -450,6 +458,61 @@ $result =
is($result, qq(5000),
'data replicated to subscriber by serializing messages to disk');
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# ============================================================================
+# Test re-apply a failed streaming transaction using the leader apply
+# worker and apply subsequent streaming transaction using the parallel apply
+# worker after this retry succeeds.
+# ============================================================================
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE UNIQUE INDEX idx_tab on test_tab_2(a)");
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', qq{
+BEGIN;
+INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
+INSERT INTO test_tab_2 values(1);
+COMMIT;});
+
+# Check if the parallel apply worker is not started because the above
+# transaction failed to be applied.
+$node_subscriber->wait_for_log(
+ qr/DEBUG: ( [A-Z0-9]+:)? parallel apply workers are not used for retries/,
+ $offset);
+
+# Drop the unique index on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres', "DROP INDEX idx_tab");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(5001), 'data replicated to subscriber after dropping index');
+
+# After successfully retrying to apply a failed streaming transaction, apply
+# the following streaming transaction using the parallel apply worker.
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab_2 SELECT i FROM generate_series(5001, 10000) s(i)");
+
+check_parallel_log($node_subscriber, $offset, 1, 'COMMIT');
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(10001),
+ 'data replicated to subscriber using the parallel apply worker');
+
$node_subscriber->stop;
$node_publisher->stop;
--
2.7.2.windows.1
[application/octet-stream] v78-0001-Add-a-main_worker_pid-to-pg_stat_subscription.patch (9.4K, ../OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v78-0001-Add-a-main_worker_pid-to-pg_stat_subscription.patch)
download | inline diff:
From c2feee2f08c68608674da45c894f3a110dbad3bd Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <[email protected]>
Date: Thu, 6 Oct 2022 14:42:24 +0800
Subject: [PATCH v78 1/4] Add a main_worker_pid to pg_stat_subscription
main_worker_pid is Process ID of the leader apply worker, if this process is a
apply parallel worker. NULL if this process is a leader apply worker or a
synchronization worker.
The new column can make it easier to distinguish leader apply worker and apply
parallel worker which is also similar to the 'leader_pid' column in
pg_stat_activity.
---
doc/src/sgml/logical-replication.sgml | 3 ++-
doc/src/sgml/monitoring.sgml | 26 ++++++++++++++++++------
src/backend/catalog/system_views.sql | 1 +
src/backend/replication/logical/launcher.c | 32 ++++++++++++++++--------------
src/include/catalog/pg_proc.dat | 6 +++---
src/test/regress/expected/rules.out | 3 ++-
6 files changed, 45 insertions(+), 26 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 54f48be..8029fab 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1692,7 +1692,8 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
subscription. A disabled subscription or a crashed subscription will have
zero rows in this view. If the initial data synchronization of any
table is in progress, there will be additional workers for the tables
- being synchronized.
+ being synchronized. Moreover, if the streaming transaction is applied in
+ parallel, there will be additional workers.
</para>
</sect1>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index cf220c3..c2e7bdb 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -3198,11 +3198,22 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
<row>
<entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>apply_leader_pid</structfield> <type>integer</type>
+ </para>
+ <para>
+ Process ID of the leader apply worker, if this process is a apply
+ parallel worker. NULL if this process is a leader apply worker or a
+ synchronization worker.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
<structfield>relid</structfield> <type>oid</type>
</para>
<para>
OID of the relation that the worker is synchronizing; null for the
- main apply worker
+ main apply worker and the parallel apply worker
</para></entry>
</row>
@@ -3212,7 +3223,7 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
</para>
<para>
Last write-ahead log location received, the initial value of
- this field being 0
+ this field being 0; null for the parallel apply worker
</para></entry>
</row>
@@ -3221,7 +3232,8 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
<structfield>last_msg_send_time</structfield> <type>timestamp with time zone</type>
</para>
<para>
- Send time of last message received from origin WAL sender
+ Send time of last message received from origin WAL sender; null for the
+ parallel apply worker
</para></entry>
</row>
@@ -3230,7 +3242,8 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
<structfield>last_msg_receipt_time</structfield> <type>timestamp with time zone</type>
</para>
<para>
- Receipt time of last message received from origin WAL sender
+ Receipt time of last message received from origin WAL sender; null for
+ the parallel apply worker
</para></entry>
</row>
@@ -3239,7 +3252,8 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
<structfield>latest_end_lsn</structfield> <type>pg_lsn</type>
</para>
<para>
- Last write-ahead log location reported to origin WAL sender
+ Last write-ahead log location reported to origin WAL sender; null for
+ the parallel apply worker
</para></entry>
</row>
@@ -3249,7 +3263,7 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
</para>
<para>
Time of last write-ahead log location reported to origin WAL
- sender
+ sender; null for the parallel apply worker
</para></entry>
</row>
</tbody>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 447c9b9..72e46e5 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -949,6 +949,7 @@ CREATE VIEW pg_stat_subscription AS
su.oid AS subid,
su.subname,
st.pid,
+ st.apply_leader_pid,
st.relid,
st.received_lsn,
st.last_msg_send_time,
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index afb7acd..b39e0f1 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -1072,7 +1072,7 @@ IsLogicalLauncher(void)
Datum
pg_stat_get_subscription(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_SUBSCRIPTION_COLS 8
+#define PG_STAT_GET_SUBSCRIPTION_COLS 9
Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
int i;
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
@@ -1098,10 +1098,6 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
if (OidIsValid(subid) && worker.subid != subid)
continue;
- /* Skip if this is a parallel apply worker */
- if (isParallelApplyWorker(&worker))
- continue;
-
worker_pid = worker.proc->pid;
values[0] = ObjectIdGetDatum(worker.subid);
@@ -1110,26 +1106,32 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
else
nulls[1] = true;
values[2] = Int32GetDatum(worker_pid);
- if (XLogRecPtrIsInvalid(worker.last_lsn))
+
+ if (worker.apply_leader_pid == InvalidPid)
nulls[3] = true;
else
- values[3] = LSNGetDatum(worker.last_lsn);
- if (worker.last_send_time == 0)
+ values[3] = Int32GetDatum(worker.apply_leader_pid);
+
+ if (XLogRecPtrIsInvalid(worker.last_lsn))
nulls[4] = true;
else
- values[4] = TimestampTzGetDatum(worker.last_send_time);
- if (worker.last_recv_time == 0)
+ values[4] = LSNGetDatum(worker.last_lsn);
+ if (worker.last_send_time == 0)
nulls[5] = true;
else
- values[5] = TimestampTzGetDatum(worker.last_recv_time);
- if (XLogRecPtrIsInvalid(worker.reply_lsn))
+ values[5] = TimestampTzGetDatum(worker.last_send_time);
+ if (worker.last_recv_time == 0)
nulls[6] = true;
else
- values[6] = LSNGetDatum(worker.reply_lsn);
- if (worker.reply_time == 0)
+ values[6] = TimestampTzGetDatum(worker.last_recv_time);
+ if (XLogRecPtrIsInvalid(worker.reply_lsn))
nulls[7] = true;
else
- values[7] = TimestampTzGetDatum(worker.reply_time);
+ values[7] = LSNGetDatum(worker.reply_lsn);
+ if (worker.reply_time == 0)
+ nulls[8] = true;
+ else
+ values[8] = TimestampTzGetDatum(worker.reply_time);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
values, nulls);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 3810de7..dbca793 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5430,9 +5430,9 @@
proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', proparallel => 'r',
prorettype => 'record', proargtypes => 'oid',
- proallargtypes => '{oid,oid,oid,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz}',
- proargmodes => '{i,o,o,o,o,o,o,o,o}',
- proargnames => '{subid,subid,relid,pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time}',
+ proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz}',
+ proargmodes => '{i,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{subid,subid,relid,pid,apply_leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time}',
prosrc => 'pg_stat_get_subscription' },
{ oid => '2026', descr => 'statistics: current backend PID',
proname => 'pg_backend_pid', provolatile => 's', proparallel => 'r',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index fb9f936..7e6ea60 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2094,6 +2094,7 @@ pg_stat_ssl| SELECT s.pid,
pg_stat_subscription| SELECT su.oid AS subid,
su.subname,
st.pid,
+ st.apply_leader_pid,
st.relid,
st.received_lsn,
st.last_msg_send_time,
@@ -2101,7 +2102,7 @@ pg_stat_subscription| SELECT su.oid AS subid,
st.latest_end_lsn,
st.latest_end_time
FROM (pg_subscription su
- LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid)));
+ LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, apply_leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid)));
pg_stat_subscription_stats| SELECT ss.subid,
s.subname,
ss.apply_error_count,
--
2.7.2.windows.1
[application/octet-stream] v78-0002-Stop-extra-worker-if-GUC-was-changed.patch (4.2K, ../OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com/4-v78-0002-Stop-extra-worker-if-GUC-was-changed.patch)
download | inline diff:
From c9431440f992f76883c14f7519c187365f27a09d Mon Sep 17 00:00:00 2001
From: sherlockcpp <[email protected]>
Date: Sat, 17 Dec 2022 20:43:21 +0800
Subject: [PATCH v78 2/4] Stop extra worker if GUC was changed
If the max_parallel_apply_workers_per_subscription is changed to a
lower value, try to stop free workers in the pool to keep the number of
workers lower than half of the max_parallel_apply_workers_per_subscription
---
.../replication/logical/applyparallelworker.c | 60 ++++++++++++++++++----
src/backend/replication/logical/worker.c | 7 +++
src/include/replication/worker_internal.h | 1 +
3 files changed, 57 insertions(+), 11 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 2e5914d..ab758fe 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -543,6 +543,25 @@ pa_find_worker(TransactionId xid)
}
/*
+ * Stop the given parallel apply worker and free the corresponding info.
+ */
+static void
+pa_stop_worker(ParallelApplyWorkerInfo *winfo)
+{
+ int slot_no;
+ uint16 generation;
+
+ SpinLockAcquire(&winfo->shared->mutex);
+ generation = winfo->shared->logicalrep_worker_generation;
+ slot_no = winfo->shared->logicalrep_worker_slot_no;
+ SpinLockRelease(&winfo->shared->mutex);
+
+ logicalrep_pa_worker_stop(slot_no, generation);
+
+ pa_free_worker_info(winfo);
+}
+
+/*
* Makes the worker available for reuse.
*
* This removes the parallel apply worker entry from the hash table so that it
@@ -577,23 +596,42 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
list_length(ParallelApplyWorkerPool) >
(max_parallel_apply_workers_per_subscription / 2))
{
- int slot_no;
- uint16 generation;
-
- SpinLockAcquire(&winfo->shared->mutex);
- generation = winfo->shared->logicalrep_worker_generation;
- slot_no = winfo->shared->logicalrep_worker_slot_no;
- SpinLockRelease(&winfo->shared->mutex);
+ pa_stop_worker(winfo);
+ return;
+ }
- logicalrep_pa_worker_stop(slot_no, generation);
+ winfo->in_use = false;
+ winfo->serialize_changes = false;
+}
- pa_free_worker_info(winfo);
+/*
+ * Try to stop parallel apply workers that are not in use to keep the number of
+ * workers lower than half of the max_parallel_apply_workers_per_subscription.
+ */
+void
+pa_stop_idle_workers(void)
+{
+ List *active_workers;
+ ListCell *lc;
+ int max_applyworkers = max_parallel_apply_workers_per_subscription / 2;
+ if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
return;
+
+ active_workers = list_copy(ParallelApplyWorkerPool);
+
+ foreach(lc, active_workers)
+ {
+ ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
+
+ pa_stop_worker(winfo);
+
+ /* Recheck the number of workers. */
+ if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
+ break;
}
- winfo->in_use = false;
- winfo->serialize_changes = false;
+ list_free(active_workers);
}
/*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 79cda39..c7be76d 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3630,6 +3630,13 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
{
ConfigReloadPending = false;
ProcessConfigFile(PGC_SIGHUP);
+
+ /*
+ * Try to stop free workers in the pool in case the
+ * max_parallel_apply_workers_per_subscription is changed to a
+ * lower value.
+ */
+ pa_stop_idle_workers();
}
if (rc & WL_TIMEOUT)
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index db891ee..1d9d7d6 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -274,6 +274,7 @@ extern void set_apply_error_context_origin(char *originname);
/* Parallel apply worker setup and interactions */
extern void pa_allocate_worker(TransactionId xid);
extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);
+extern void pa_stop_idle_workers(void);
extern void pa_detach_all_error_mq(void);
extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
--
2.7.2.windows.1
[application/octet-stream] v78-0003-Add-GUC-stream_serialize_threshold-and-test-seri.patch (12.5K, ../OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com/5-v78-0003-Add-GUC-stream_serialize_threshold-and-test-seri.patch)
download | inline diff:
From 02992050a5425cc055e3a1ec3c99f8fea847bea1 Mon Sep 17 00:00:00 2001
From: Amit Kapila <[email protected]>
Date: Mon, 2 Jan 2023 15:37:25 +0530
Subject: [PATCH v78 3/4] Add GUC stream_serialize_threshold and test
serializing messages to disk.
---
doc/src/sgml/config.sgml | 32 +++++
.../replication/logical/applyparallelworker.c | 12 ++
src/backend/replication/logical/worker.c | 9 ++
src/backend/utils/misc/guc_tables.c | 14 ++
src/include/replication/worker_internal.h | 4 +
src/test/subscription/t/015_stream.pl | 144 ++++++++++++++++++++-
6 files changed, 212 insertions(+), 3 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 2fec613..74bb623 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11650,6 +11650,38 @@ LOG: CleanUpLock: deleting: lock(0xb7acd844) id(24688,24696,0,0,0,1)
</listitem>
</varlistentry>
+ <varlistentry id="guc-stream-serialize-threshold" xreflabel="stream_serialize_threshold">
+ <term><varname>stream_serialize_threshold</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>stream_serialize_threshold</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Forces the leader apply worker to serialize messages to files after
+ sending specified amount of streaming chunks to the parallel apply
+ worker. Setting this to zero serialize all messages. A value of
+ <literal>-1</literal> (the default) disables this feature. This is
+ intended to test serialization to files with
+ <literal>streaming = parallel</literal>.
+ </para>
+
+ <para>
+ When logical replication subscription <literal>streaming</literal>
+ parameter is set to <literal>parallel</literal>, the leader apply worker
+ sends messages to parallel workers with a timeout. By default, the
+ leader apply worker will serialize the remaining messages to files if
+ the timeout is exceeded. If this option is set to any value other than
+ <literal>-1</literal>, serialize to files even without timeout.
+ </para>
+
+ <para>
+ This parameter can only be set in the <filename>postgresql.conf</filename>
+ file or on the server command line.
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</sect1>
<sect1 id="runtime-config-short">
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index ab758fe..a7ce57b 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -254,6 +254,9 @@ static ParallelApplyWorkerInfo *stream_apply_worker = NULL;
/* A list to maintain subtransactions, if any. */
static List *subxactlist = NIL;
+/* GUC variable */
+int stream_serialize_threshold = -1;
+
static void pa_free_worker_info(ParallelApplyWorkerInfo *winfo);
static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared);
static PartialFileSetState pa_get_fileset_state(void);
@@ -1187,6 +1190,15 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
Assert(!IsTransactionState());
Assert(!winfo->serialize_changes);
+ /* Force to serialize messages if stream_serialize_threshold is reached. */
+ if (stream_serialize_threshold != -1 &&
+ (stream_serialize_threshold == 0 ||
+ stream_serialize_threshold < parallel_stream_nchunks))
+ {
+ parallel_stream_nchunks = 0;
+ return false;
+ }
+
/*
* This timeout is a bit arbitrary but testing revealed that it is sufficient
* to send the message unless the parallel apply worker is waiting on some
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c7be76d..5c8ce97 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -328,6 +328,12 @@ static TransactionId stream_xid = InvalidTransactionId;
static uint32 parallel_stream_nchanges = 0;
/*
+ * The number of streaming chunks sent by leader apply worker during one
+ * streamed transaction. This is only used when stream_serialize_threshold > 0.
+ */
+uint32 parallel_stream_nchunks = 0;
+
+/*
* We enable skipping all data modification changes (INSERT, UPDATE, etc.) for
* the subscription if the remote transaction's finish LSN matches the subskiplsn.
* Once we start skipping changes, we don't stop it until we skip all changes of
@@ -1521,6 +1527,9 @@ apply_handle_stream_start(StringInfo s)
case TRANS_LEADER_SEND_TO_PARALLEL:
Assert(winfo);
+ if (stream_serialize_threshold > 0)
+ parallel_stream_nchunks++;
+
/*
* Once we start serializing the changes, the parallel apply
* worker will wait for the leader to release the stream lock
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 92545b4..302ceb7 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -61,6 +61,7 @@
#include "replication/logicallauncher.h"
#include "replication/slot.h"
#include "replication/syncrep.h"
+#include "replication/worker_internal.h"
#include "storage/bufmgr.h"
#include "storage/large_object.h"
#include "storage/pg_shmem.h"
@@ -3015,6 +3016,19 @@ struct config_int ConfigureNamesInt[] =
},
{
+ {"stream_serialize_threshold", PGC_SIGHUP, DEVELOPER_OPTIONS,
+ gettext_noop("Forces the leader apply worker to serialize messages "
+ "to files after sending specified amount of streaming "
+ "chunks in streaming parallel mode."),
+ gettext_noop("A value of -1 disables this feature."),
+ GUC_NOT_IN_SAMPLE
+ },
+ &stream_serialize_threshold,
+ -1, -1, INT_MAX,
+ NULL, NULL, NULL
+ },
+
+ {
{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
gettext_noop("Sets the amount of time to wait before forcing "
"log file rotation."),
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 1d9d7d6..f3b2f2d 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -225,6 +225,10 @@ extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
extern PGDLLIMPORT bool in_remote_transaction;
+extern PGDLLIMPORT int stream_serialize_threshold;
+
+extern PGDLLIMPORT uint32 parallel_stream_nchunks;
+
extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 91e8aa8..83d6956 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -133,13 +133,20 @@ sub test_streaming
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
-$node_publisher->append_conf('postgresql.conf',
- 'logical_decoding_work_mem = 64kB');
+$node_publisher->append_conf(
+ 'postgresql.conf', qq(
+max_prepared_transactions = 10
+logical_decoding_work_mem = 64kB
+));
$node_publisher->start;
# Create subscriber node
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->append_conf(
+ 'postgresql.conf', qq(
+max_prepared_transactions = 10
+));
$node_subscriber->start;
# Create some preexisting content on publisher
@@ -170,7 +177,7 @@ my $appname = 'tap_sub';
# Test using streaming mode 'on'
################################
$node_subscriber->safe_psql('postgres',
- "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on, two_phase = on)"
);
# Wait for initial table sync to finish
@@ -312,6 +319,137 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
is($result, qq(10000), 'data replicated to subscriber after dropping index');
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# Test serializing messages to disk
+
+# Set stream_serialize_threshold to zero, so the messages will be serialized to disk.
+$node_subscriber->safe_psql('postgres',
+ 'ALTER SYSTEM SET stream_serialize_threshold = 0;');
+$node_subscriber->reload;
+
+# Run a query to make sure that the reload has taken effect.
+$node_subscriber->safe_psql('postgres', q{SELECT 1});
+
+# Serialize the COMMIT transaction.
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i)");
+
+# Ensure that the messages are serialized.
+$node_subscriber->wait_for_log(
+ qr/DEBUG: ( [A-Z0-9]+:)? opening file ".*\.changes" for streamed changes/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is committed on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(5000),
+ 'data replicated to subscriber by serializing messages to disk');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# Serialize the PREPARE transaction.
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
+ PREPARE TRANSACTION 'xact';
+ });
+
+# Ensure that the messages are serialized.
+$node_subscriber->wait_for_log(
+ qr/DEBUG: ( [A-Z0-9]+:)? opening file ".*\.changes" for streamed changes/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is in prepared state on subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(1), 'transaction is prepared on subscriber');
+
+# Check that 2PC gets committed on subscriber
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'xact';");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is committed on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(5000),
+ 'data replicated to subscriber by serializing messages to disk');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# Serialize the ABORT top-transaction.
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
+ ROLLBACK;
+ });
+
+# Ensure that the messages are serialized.
+$node_subscriber->wait_for_log(
+ qr/DEBUG: ( [A-Z0-9]+:)? opening file ".*\.changes" for streamed changes/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is aborted on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(0),
+ 'data replicated to subscriber by serializing messages to disk');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# Serialize the ABORT sub-transaction.
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
+ SAVEPOINT sp;
+ INSERT INTO test_tab_2 SELECT i FROM generate_series(5001, 10000) s(i);
+ ROLLBACK TO sp;
+ COMMIT;
+ });
+
+# Ensure that the messages are serialized.
+$node_subscriber->wait_for_log(
+ qr/DEBUG: ( [A-Z0-9]+:)? opening file ".*\.changes" for streamed changes/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that only sub-transaction is aborted on subscriber.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(5000),
+ 'data replicated to subscriber by serializing messages to disk');
+
$node_subscriber->stop;
$node_publisher->stop;
--
2.7.2.windows.1
view thread (105+ 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]
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.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