public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: [email protected] <[email protected]>
To: Peter Smith <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Fri, 29 Apr 2022 05:22:41 +0000
Message-ID: <OSZPR01MB6310F0FABB05F8E5BB31A5B0FDFC9@OSZPR01MB6310.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OS0PR01MB5716E8D536552467EFB512EF94FC9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
References: <CAA4eK1+wyN6zpaHUkCLorEWNx75MG0xhMwcFhvjqm2KURZEAGw@mail.gmail.com>
<OS0PR01MB57160DFDEF5B1C4668346B7D94E99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57166BC3CAA873364A2CC07A94EF9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1Kc0VAhJAQe4_5YGrnTpfKYA9=F6sxqEKPrrax5Hy0+jw@mail.gmail.com>
<OS0PR01MB571616192F70A18AF88F3B2894F59@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716A6C18E69FB1EA9A27EED94F59@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+PuqYP5eD5wcSCtk=a6KuMjat2UCzqyGoE7sieCaBsVskQ@mail.gmail.com>
<OS0PR01MB57160D1B0D29DC66ED035F5694F89@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716E8D536552467EFB512EF94FC9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Fri, Apr 29, 2022 10:07 AM Hou, Zhijie/侯 志杰 <[email protected]> wrote:
>
> I addressed the rest of Peter's comments and here is a new version patch.
>
Thanks for your patch.
The patch modified streaming option in logical replication, it can be set to
'on', 'off' and 'apply'. The new option 'apply' haven't been tested in the tap test.
Attach a patch which modified the subscription tap test to cover both 'on' and
'apply' option. (The main patch is also attached to make cfbot happy.)
Besides, I noticed that for two-phase commit transactions, if the transaction is
prepared by a background worker, the background worker would be asked to handle
the message about commit/rollback this transaction. Is it possible that the
messages about commit/rollback prepared transaction are handled by apply worker
directly?
Regards,
Shi yu
Attachments:
[application/octet-stream] v5-0002-Test-streaming-apply-option-in-tap-test.patch (61.9K, ../OSZPR01MB6310F0FABB05F8E5BB31A5B0FDFC9@OSZPR01MB6310.jpnprd01.prod.outlook.com/2-v5-0002-Test-streaming-apply-option-in-tap-test.patch)
download | inline diff:
From 680631bd58b9b565a94887f709af81f8e1ba2713 Mon Sep 17 00:00:00 2001
From: "shiy.fnst" <[email protected]>
Date: Fri, 29 Apr 2022 11:01:20 +0800
Subject: [PATCH v5 2/2] Test streaming apply option in tap test
For the tap tests about streaming option in logical replication, test both
'on' and 'apply' option.
---
src/test/subscription/t/015_stream.pl | 168 ++++----
src/test/subscription/t/016_stream_subxact.pl | 84 ++--
src/test/subscription/t/017_stream_ddl.pl | 152 ++++---
.../t/018_stream_subxact_abort.pl | 165 +++++---
.../t/019_stream_subxact_ddl_abort.pl | 71 +++-
.../subscription/t/022_twophase_cascade.pl | 278 ++++++------
.../subscription/t/023_twophase_stream.pl | 398 ++++++++++--------
7 files changed, 740 insertions(+), 576 deletions(-)
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 6561b189de..0985a3b2bf 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -8,6 +8,88 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+
+ # Interleave a pair of transactions, each exceeding the 64kB limit.
+ my $in = '';
+ my $out = '';
+
+ my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
+
+ my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+ on_error_stop => 0);
+
+ $in .= q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ };
+ $h->pump_nb;
+
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 9999) s(i);
+ DELETE FROM test_tab WHERE a > 5000;
+ COMMIT;
+ });
+
+ $in .= q{
+ COMMIT;
+ \q
+ };
+ $h->finish; # errors make the next test fail, so ignore them here
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334), 'check extra columns contain local defaults');
+
+ # Test the streaming in binary mode
+ $node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET (binary = on)");
+
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 10000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(6667|6667|6667), 'check extra columns contain local defaults');
+
+ # Change the local values of the extra columns on the subscriber,
+ # update publisher, and check that subscriber retains the expected
+ # values. This is to ensure that non-streaming transactions behave
+ # properly after a streaming transaction.
+ $node_subscriber->safe_psql('postgres',
+ "UPDATE test_tab SET c = 'epoch'::timestamptz + 987654321 * interval '1s'"
+ );
+ $node_publisher->safe_psql('postgres',
+ "UPDATE test_tab SET b = md5(a::text)");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(extract(epoch from c) = 987654321), count(d = 999) FROM test_tab"
+ );
+ is($result, qq(6667|6667|6667),
+ 'check extra columns contain locally changed data');
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,99 +119,39 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
-
$node_publisher->wait_for_catchup($appname);
-
# Also wait for initial table sync to finish
my $synced_query =
"SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
$node_subscriber->poll_query_until('postgres', $synced_query)
or die "Timed out while waiting for subscriber to synchronize data";
-
my $result =
$node_subscriber->safe_psql('postgres',
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-# Interleave a pair of transactions, each exceeding the 64kB limit.
-my $in = '';
-my $out = '';
-
-my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
- on_error_stop => 0);
-
-$in .= q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-};
-$h->pump_nb;
-
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 9999) s(i);
-DELETE FROM test_tab WHERE a > 5000;
-COMMIT;
-});
-
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish; # errors make the next test fail, so ignore them here
-
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'check extra columns contain local defaults');
-
-# Test the streaming in binary mode
-$node_subscriber->safe_psql('postgres',
- "ALTER SUBSCRIPTION tap_sub SET (binary = on)");
-
-# Insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 10000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname);
+# Test streaming mode apply
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE (a > 2)");
$node_publisher->wait_for_catchup($appname);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(6667|6667|6667), 'check extra columns contain local defaults');
-
-# Change the local values of the extra columns on the subscriber,
-# update publisher, and check that subscriber retains the expected
-# values. This is to ensure that non-streaming transactions behave
-# properly after a streaming transaction.
+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',
- "UPDATE test_tab SET c = 'epoch'::timestamptz + 987654321 * interval '1s'"
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply, binary = off)"
);
-$node_publisher->safe_psql('postgres',
- "UPDATE test_tab SET b = md5(a::text)");
+$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 PUBLICATION";
-$node_publisher->wait_for_catchup($appname);
-
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(extract(epoch from c) = 987654321), count(d = 999) FROM test_tab"
-);
-is($result, qq(6667|6667|6667),
- 'check extra columns contain locally changed data');
+test_streaming($node_publisher, $node_subscriber, $appname);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/016_stream_subxact.pl b/src/test/subscription/t/016_stream_subxact.pl
index f27f1694f2..bc49bf9e40 100644
--- a/src/test/subscription/t/016_stream_subxact.pl
+++ b/src/test/subscription/t/016_stream_subxact.pl
@@ -8,6 +8,46 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+
+ # Insert, update and delete enough rows to exceed 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series( 3, 500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501, 1000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001, 1500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501, 2000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s4;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001, 2500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(1667|1667|1667),
+ 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+ );
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +77,8 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,40 +96,22 @@ my $result =
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-# Insert, update and delete enough rows to exceed 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series( 3, 500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501, 1000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001, 1500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501, 2000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s4;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001, 2500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname);
+# Test streaming mode apply
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE (a > 2)");
$node_publisher->wait_for_catchup($appname);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(1667|1667|1667),
- 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+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(streaming = apply)"
);
+$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 PUBLICATION";
+test_streaming($node_publisher, $node_subscriber, $appname);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/017_stream_ddl.pl b/src/test/subscription/t/017_stream_ddl.pl
index 0bce63b716..fccac95f69 100644
--- a/src/test/subscription/t/017_stream_ddl.pl
+++ b/src/test/subscription/t/017_stream_ddl.pl
@@ -8,6 +8,82 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+
+ # a small (non-streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab VALUES (3, md5(3::text));
+ ALTER TABLE test_tab ADD COLUMN c INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab VALUES (4, md5(4::text), -4);
+ COMMIT;
+ });
+
+ # large (streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(5, 1000) s(i);
+ ALTER TABLE test_tab ADD COLUMN d INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001, 2000) s(i);
+ COMMIT;
+ });
+
+ # a small (non-streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab VALUES (2001, md5(2001::text), -2001, 2*2001);
+ ALTER TABLE test_tab ADD COLUMN e INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab VALUES (2002, md5(2002::text), -2002, 2*2002, -3*2002);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d), count(e) FROM test_tab");
+ is($result, qq(2002|1999|1002|1),
+ 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+ );
+
+ # A large (streamed) transaction with DDL and DML. One of the DDL is performed
+ # after DML to ensure that we invalidate the schema sent for test_tab so that
+ # the next transaction has to send the schema again.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(2003,5000) s(i);
+ ALTER TABLE test_tab ADD COLUMN f INT;
+ COMMIT;
+ });
+
+ # A small transaction that won't get streamed. This is just to ensure that we
+ # send the schema again to reflect the last column added in the previous test.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i, 4*i FROM generate_series(5001,5005) s (i);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d), count(e), count(f) FROM test_tab");
+ is($result, qq(5005|5002|4005|3004|5),
+ 'check data was copied to subscriber for both streaming and non-streaming transactions'
+ );
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +113,8 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,76 +132,28 @@ my $result =
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|0|0), 'check initial data was copied to subscriber');
-# a small (non-streamed) transaction with DDL and DML
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab VALUES (3, md5(3::text));
-ALTER TABLE test_tab ADD COLUMN c INT;
-SAVEPOINT s1;
-INSERT INTO test_tab VALUES (4, md5(4::text), -4);
-COMMIT;
-});
-
-# large (streamed) transaction with DDL and DML
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(5, 1000) s(i);
-ALTER TABLE test_tab ADD COLUMN d INT;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001, 2000) s(i);
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname);
-# a small (non-streamed) transaction with DDL and DML
+# Test streaming mode apply
$node_publisher->safe_psql(
'postgres', q{
-BEGIN;
-INSERT INTO test_tab VALUES (2001, md5(2001::text), -2001, 2*2001);
-ALTER TABLE test_tab ADD COLUMN e INT;
-SAVEPOINT s1;
-INSERT INTO test_tab VALUES (2002, md5(2002::text), -2002, 2*2002, -3*2002);
-COMMIT;
+DELETE FROM test_tab WHERE (a > 2);
+ALTER TABLE test_tab DROP COLUMN c, DROP COLUMN d, DROP COLUMN e, DROP COLUMN f;
});
$node_publisher->wait_for_catchup($appname);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d), count(e) FROM test_tab");
-is($result, qq(2002|1999|1002|1),
- 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
-
-# A large (streamed) transaction with DDL and DML. One of the DDL is performed
-# after DML to ensure that we invalidate the schema sent for test_tab so that
-# the next transaction has to send the schema again.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(2003,5000) s(i);
-ALTER TABLE test_tab ADD COLUMN f INT;
-COMMIT;
-});
-
-# A small transaction that won't get streamed. This is just to ensure that we
-# send the schema again to reflect the last column added in the previous test.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i, 4*i FROM generate_series(5001,5005) s(i);
-COMMIT;
-});
-
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d), count(e), count(f) FROM test_tab");
-is($result, qq(5005|5002|4005|3004|5),
- 'check data was copied to subscriber for both streaming and non-streaming transactions'
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)"
);
+$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 PUBLICATION";
+
+test_streaming($node_publisher, $node_subscriber, $appname);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/018_stream_subxact_abort.pl b/src/test/subscription/t/018_stream_subxact_abort.pl
index 7155442e76..8ba38d40b3 100644
--- a/src/test/subscription/t/018_stream_subxact_abort.pl
+++ b/src/test/subscription/t/018_stream_subxact_abort.pl
@@ -8,6 +8,87 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+
+ # large (streamed) transaction with DDL, DML and ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501,1000) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001,1500) s(i);
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501,2000) s(i);
+ ROLLBACK TO s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001,2500) s(i);
+ ROLLBACK TO s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2501,3000) s(i);
+ SAVEPOINT s4;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3001,3500) s(i);
+ SAVEPOINT s5;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3501,4000) s(i);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2000|0),
+ 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+ );
+
+ # large (streamed) transaction with subscriber receiving out of order
+ # subtransaction ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(4001,4500) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001,5500) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(6001,6500) s(i);
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(7001,7500) s(i);
+ RELEASE s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8001,8500) s(i);
+ ROLLBACK TO s1;
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2500|0),
+ 'check rollback to savepoint was reflected on subscriber');
+
+ # large (streamed) transaction with subscriber receiving rollback
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8501,9000) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9001,9500) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9501,10000) s(i);
+ ROLLBACK;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2500|0), 'check rollback was reflected on subscriber');
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -36,6 +117,8 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -53,81 +136,23 @@ my $result =
"SELECT count(*), count(c) FROM test_tab");
is($result, qq(2|0), 'check initial data was copied to subscriber');
-# large (streamed) transaction with DDL, DML and ROLLBACKs
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501,1000) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001,1500) s(i);
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501,2000) s(i);
-ROLLBACK TO s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001,2500) s(i);
-ROLLBACK TO s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2501,3000) s(i);
-SAVEPOINT s4;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3001,3500) s(i);
-SAVEPOINT s5;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3501,4000) s(i);
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname);
+# Test streaming mode apply
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE (a > 2)");
$node_publisher->wait_for_catchup($appname);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2000|0),
- 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+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(streaming = apply)"
+);
+$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 PUBLICATION";
-# large (streamed) transaction with subscriber receiving out of order
-# subtransaction ROLLBACKs
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(4001,4500) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001,5500) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(6001,6500) s(i);
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(7001,7500) s(i);
-RELEASE s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8001,8500) s(i);
-ROLLBACK TO s1;
-COMMIT;
-});
-
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2500|0),
- 'check rollback to savepoint was reflected on subscriber');
-
-# large (streamed) transaction with subscriber receiving rollback
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8501,9000) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9001,9500) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9501,10000) s(i);
-ROLLBACK;
-});
-
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2500|0), 'check rollback was reflected on subscriber');
+test_streaming($node_publisher, $node_subscriber, $appname);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/019_stream_subxact_ddl_abort.pl b/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
index dbd0fca4d1..c6150d3155 100644
--- a/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
+++ b/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
@@ -9,6 +9,40 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+
+ # large (streamed) transaction with DDL, DML and ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
+ ALTER TABLE test_tab ADD COLUMN c INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(501,1000) s(i);
+ ALTER TABLE test_tab ADD COLUMN d INT;
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001,1500) s(i);
+ ALTER TABLE test_tab ADD COLUMN e INT;
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(1501,2000) s(i);
+ ALTER TABLE test_tab DROP COLUMN c;
+ ROLLBACK TO s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), i FROM generate_series(501,1000) s(i);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(1000|500),
+ 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+ );
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +71,8 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,34 +90,27 @@ my $result =
"SELECT count(*), count(c) FROM test_tab");
is($result, qq(2|0), 'check initial data was copied to subscriber');
-# large (streamed) transaction with DDL, DML and ROLLBACKs
+test_streaming($node_publisher, $node_subscriber, $appname);
+
+# Test streaming mode apply
$node_publisher->safe_psql(
'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
-ALTER TABLE test_tab ADD COLUMN c INT;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(501,1000) s(i);
-ALTER TABLE test_tab ADD COLUMN d INT;
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001,1500) s(i);
-ALTER TABLE test_tab ADD COLUMN e INT;
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(1501,2000) s(i);
+DELETE FROM test_tab WHERE (a > 2);
ALTER TABLE test_tab DROP COLUMN c;
-ROLLBACK TO s1;
-INSERT INTO test_tab SELECT i, md5(i::text), i FROM generate_series(501,1000) s(i);
-COMMIT;
});
-
$node_publisher->wait_for_catchup($appname);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(1000|500),
- 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+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(streaming = apply)"
);
+$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 PUBLICATION";
+
+test_streaming($node_publisher, $node_subscriber, $appname);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/022_twophase_cascade.pl b/src/test/subscription/t/022_twophase_cascade.pl
index 900c25d5ce..bb20a3640c 100644
--- a/src/test/subscription/t/022_twophase_cascade.pl
+++ b/src/test/subscription/t/022_twophase_cascade.pl
@@ -11,6 +11,148 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# ---------------------
+# 2PC + STREAMING TESTS
+# ---------------------
+sub test_streaming
+{
+ my ($node_A, $node_B, $node_C, $appname_B, $appname_C, $streaming) = @_;
+
+ my $oldpid_B = $node_A->safe_psql('postgres', "
+ SELECT pid FROM pg_stat_replication
+ WHERE application_name = '$appname_B' AND state = 'streaming';");
+ my $oldpid_C = $node_B->safe_psql('postgres', "
+ SELECT pid FROM pg_stat_replication
+ WHERE application_name = '$appname_C' AND state = 'streaming';");
+
+ # Setup logical replication streaming mode
+
+ $node_B->safe_psql('postgres', "
+ ALTER SUBSCRIPTION tap_sub_B
+ SET (streaming = $streaming);");
+ $node_C->safe_psql('postgres', "
+ ALTER SUBSCRIPTION tap_sub_C
+ SET (streaming = $streaming)");
+
+ # Wait for subscribers to finish initialization
+
+ $node_A->poll_query_until('postgres', "
+ SELECT pid != $oldpid_B FROM pg_stat_replication
+ WHERE application_name = '$appname_B' AND state = 'streaming';"
+ ) or die "Timed out while waiting for apply to restart";
+ $node_B->poll_query_until('postgres', "
+ SELECT pid != $oldpid_C FROM pg_stat_replication
+ WHERE application_name = '$appname_C' AND state = 'streaming';"
+ ) or die "Timed out while waiting for apply to restart";
+
+ ###############################
+ # Test 2PC PREPARE / COMMIT PREPARED.
+ # 1. Data is streamed as a 2PC transaction.
+ # 2. Then do commit prepared.
+ #
+ # Expect all data is replicated on subscriber(s) after the commit.
+ ###############################
+
+ # First, delete the data except for 2 rows (delete will be replicated)
+ $node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ # Then 2PC PREPARE
+ $node_A->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state is prepared on subscriber(s)
+ my $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber B');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber C');
+
+ # 2PC COMMIT
+ $node_A->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check that transaction was committed on subscriber(s)
+ $result = $node_B->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber B, and extra columns have local defaults');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber C, and extra columns have local defaults');
+
+ # check the transaction state is ended on subscriber(s)
+ $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber B');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber C');
+
+ ###############################
+ # Test 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT.
+ # 0. Cleanup from previous test leaving only 2 rows.
+ # 1. Insert one more row.
+ # 2. Record a SAVEPOINT.
+ # 3. Data is streamed using 2PC.
+ # 4. Do rollback to SAVEPOINT prior to the streamed inserts.
+ # 5. Then COMMIT PREPARED.
+ #
+ # Expect data after the SAVEPOINT is aborted leaving only 3 rows (= 2 original + 1 from step 1).
+ ###############################
+
+ # First, delete the data except for 2 rows (delete will be replicated)
+ $node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT
+ $node_A->safe_psql('postgres', "
+ BEGIN;
+ INSERT INTO test_tab VALUES (9999, 'foobar');
+ SAVEPOINT sp_inner;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ ROLLBACK TO SAVEPOINT sp_inner;
+ PREPARE TRANSACTION 'outer';
+ ");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state prepared on subscriber(s)
+ $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber B');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber C');
+
+ # 2PC COMMIT
+ $node_A->safe_psql('postgres', "COMMIT PREPARED 'outer';");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state is ended on subscriber
+ $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is ended on subscriber B');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is ended on subscriber C');
+
+ # check inserts are visible at subscriber(s).
+ # All the streamed data (prior to the SAVEPOINT) should be rolled back.
+ # (9999, 'foobar') should be committed.
+ $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab where b = 'foobar';");
+ is($result, qq(1), 'Rows committed are present on subscriber B');
+ $result = $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
+ is($result, qq(3), 'Rows committed are present on subscriber B');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab where b = 'foobar';");
+ is($result, qq(1), 'Rows committed are present on subscriber C');
+ $result = $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
+ is($result, qq(3), 'Rows committed are present on subscriber C');
+}
+
###############################
# Setup a cascade of pub/sub nodes.
# node_A -> node_B -> node_C
@@ -229,140 +371,8 @@ is($result, qq(21), 'Rows committed are present on subscriber B');
$result = $node_C->safe_psql('postgres', "SELECT a FROM tab_full where a IN (21,22);");
is($result, qq(21), 'Rows committed are present on subscriber C');
-# ---------------------
-# 2PC + STREAMING TESTS
-# ---------------------
-
-my $oldpid_B = $node_A->safe_psql('postgres', "
- SELECT pid FROM pg_stat_replication
- WHERE application_name = '$appname_B' AND state = 'streaming';");
-my $oldpid_C = $node_B->safe_psql('postgres', "
- SELECT pid FROM pg_stat_replication
- WHERE application_name = '$appname_C' AND state = 'streaming';");
-
-# Setup logical replication (streaming = on)
-
-$node_B->safe_psql('postgres', "
- ALTER SUBSCRIPTION tap_sub_B
- SET (streaming = on);");
-$node_C->safe_psql('postgres', "
- ALTER SUBSCRIPTION tap_sub_C
- SET (streaming = on)");
-
-# Wait for subscribers to finish initialization
-
-$node_A->poll_query_until('postgres', "
- SELECT pid != $oldpid_B FROM pg_stat_replication
- WHERE application_name = '$appname_B' AND state = 'streaming';"
-) or die "Timed out while waiting for apply to restart";
-$node_B->poll_query_until('postgres', "
- SELECT pid != $oldpid_C FROM pg_stat_replication
- WHERE application_name = '$appname_C' AND state = 'streaming';"
-) or die "Timed out while waiting for apply to restart";
-
-###############################
-# Test 2PC PREPARE / COMMIT PREPARED.
-# 1. Data is streamed as a 2PC transaction.
-# 2. Then do commit prepared.
-#
-# Expect all data is replicated on subscriber(s) after the commit.
-###############################
-
-# Insert, update and delete enough rows to exceed the 64kB limit.
-# Then 2PC PREPARE
-$node_A->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check the transaction state is prepared on subscriber(s)
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber B');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber C');
-
-# 2PC COMMIT
-$node_A->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check that transaction was committed on subscriber(s)
-$result = $node_B->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber B, and extra columns have local defaults');
-$result = $node_C->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber C, and extra columns have local defaults');
-
-# check the transaction state is ended on subscriber(s)
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber B');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber C');
-
-###############################
-# Test 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT.
-# 0. Cleanup from previous test leaving only 2 rows.
-# 1. Insert one more row.
-# 2. Record a SAVEPOINT.
-# 3. Data is streamed using 2PC.
-# 4. Do rollback to SAVEPOINT prior to the streamed inserts.
-# 5. Then COMMIT PREPARED.
-#
-# Expect data after the SAVEPOINT is aborted leaving only 3 rows (= 2 original + 1 from step 1).
-###############################
-
-# First, delete the data except for 2 rows (delete will be replicated)
-$node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT
-$node_A->safe_psql('postgres', "
- BEGIN;
- INSERT INTO test_tab VALUES (9999, 'foobar');
- SAVEPOINT sp_inner;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- ROLLBACK TO SAVEPOINT sp_inner;
- PREPARE TRANSACTION 'outer';
- ");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check the transaction state prepared on subscriber(s)
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber B');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber C');
-
-# 2PC COMMIT
-$node_A->safe_psql('postgres', "COMMIT PREPARED 'outer';");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check the transaction state is ended on subscriber
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is ended on subscriber B');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is ended on subscriber C');
-
-# check inserts are visible at subscriber(s).
-# All the streamed data (prior to the SAVEPOINT) should be rolled back.
-# (9999, 'foobar') should be committed.
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab where b = 'foobar';");
-is($result, qq(1), 'Rows committed are present on subscriber B');
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
-is($result, qq(3), 'Rows committed are present on subscriber B');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab where b = 'foobar';");
-is($result, qq(1), 'Rows committed are present on subscriber C');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
-is($result, qq(3), 'Rows committed are present on subscriber C');
+test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'on');
+test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'apply');
###############################
# check all the cleanup
diff --git a/src/test/subscription/t/023_twophase_stream.pl b/src/test/subscription/t/023_twophase_stream.pl
index 93ce3ef132..f5c4016480 100644
--- a/src/test/subscription/t/023_twophase_stream.pl
+++ b/src/test/subscription/t/023_twophase_stream.pl
@@ -8,6 +8,203 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname) = @_;
+ ###############################
+ # Test 2PC PREPARE / COMMIT PREPARED.
+ # 1. Data is streamed as a 2PC transaction.
+ # 2. Then do commit prepared.
+ #
+ # Expect all data is replicated on subscriber side after the commit.
+ ###############################
+
+ # check that 2PC gets replicated to subscriber
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is in prepared state on subscriber
+ my $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber');
+
+ # 2PC transaction gets committed
+ $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is committed on subscriber
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab") ;
+ is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults');
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber');
+
+ ###############################
+ # Test 2PC PREPARE / ROLLBACK PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. Do rollback prepared.
+ #
+ # Expect data rolls back leaving only the original 2 rows.
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # 2PC transaction gets aborted
+ $node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is aborted on subscriber
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab") ;
+ is($result, qq(2|2|2), 'Rows inserted by 2PC are rolled back, leaving only the original 2 rows');
+
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is aborted on subscriber');
+
+ ###############################
+ # Check that 2PC COMMIT PREPARED is decoded properly on crash restart.
+ # 1. insert, update and delete enough rows to exceed the 64kB limit.
+ # 2. Then server crashes before the 2PC transaction is committed.
+ # 3. After servers are restarted the pending transaction is committed.
+ #
+ # Expect all data is replicated on subscriber side after the commit.
+ # Note: both publisher and subscriber do crash/restart.
+ ###############################
+
+ $node_publisher->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $node_subscriber->stop('immediate');
+ $node_publisher->stop('immediate');
+
+ $node_publisher->start;
+ $node_subscriber->start;
+
+ # commit post the restart
+ $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+ $node_publisher->wait_for_catchup($appname);
+
+ # check inserts are visible
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab") ;
+ is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults');
+
+ ###############################
+ # Do INSERT after the PREPARE but before ROLLBACK PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. A single row INSERT is done which is after the PREPARE.
+ # 4. Then do a ROLLBACK PREPARED.
+ #
+ # Expect the 2PC data rolls back leaving only 3 rows on the subscriber
+ # (the original 2 + inserted 1).
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # Insert a different record (now we are outside of the 2PC transaction)
+ # Note: the 2PC transaction still holds row locks so make sure this insert is for a separate primary key
+ $node_publisher->safe_psql('postgres', "INSERT INTO test_tab VALUES (99999, 'foobar')");
+
+ # 2PC transaction gets aborted
+ $node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is aborted on subscriber,
+ # but the extra INSERT outside of the 2PC still was replicated
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab") ;
+ is($result, qq(3|3|3), 'check the outside insert was copied to subscriber');
+
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is aborted on subscriber');
+
+ ###############################
+ # Do INSERT after the PREPARE but before COMMIT PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. A single row INSERT is done which is after the PREPARE.
+ # 4. Then do a COMMIT PREPARED.
+ #
+ # Expect 2PC data + the extra row are on the subscriber
+ # (the 3334 + inserted 1 = 3335).
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # Insert a different record (now we are outside of the 2PC transaction)
+ # Note: the 2PC transaction still holds row locks so make sure this insert is for a separare primary key
+ $node_publisher->safe_psql('postgres', "INSERT INTO test_tab VALUES (99999, 'foobar')");
+
+ # 2PC transaction gets committed
+ $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is committed on subscriber
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3335|3335|3335), 'Rows inserted by 2PC (as well as outside insert) have committed on subscriber, and extra columns contain local defaults');
+
+ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber');
+}
+
###############################
# Setup
###############################
@@ -42,6 +239,8 @@ my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+# Test streaming mode on
$node_subscriber->safe_psql('postgres', "
CREATE SUBSCRIPTION tap_sub
CONNECTION '$publisher_connstr application_name=$appname'
@@ -69,198 +268,23 @@ $node_subscriber->poll_query_until('postgres', $twophase_query)
my $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-###############################
-# Test 2PC PREPARE / COMMIT PREPARED.
-# 1. Data is streamed as a 2PC transaction.
-# 2. Then do commit prepared.
-#
-# Expect all data is replicated on subscriber side after the commit.
-###############################
-
-# check that 2PC gets replicated to subscriber
-# Insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
+test_streaming($node_publisher, $node_subscriber, $appname);
-# 2PC transaction gets committed
-$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is committed on subscriber
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults');
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber');
-
-###############################
-# Test 2PC PREPARE / ROLLBACK PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. Do rollback prepared.
-#
-# Expect data rolls back leaving only the original 2 rows.
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
+# Test streaming mode apply
$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# 2PC transaction gets aborted
-$node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'test_prepared_tab';");
-
$node_publisher->wait_for_catchup($appname);
-# check that transaction is aborted on subscriber
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(2|2|2), 'Rows inserted by 2PC are rolled back, leaving only the original 2 rows');
-
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is aborted on subscriber');
-
-###############################
-# Check that 2PC COMMIT PREPARED is decoded properly on crash restart.
-# 1. insert, update and delete enough rows to exceed the 64kB limit.
-# 2. Then server crashes before the 2PC transaction is committed.
-# 3. After servers are restarted the pending transaction is committed.
-#
-# Expect all data is replicated on subscriber side after the commit.
-# Note: both publisher and subscriber do crash/restart.
-###############################
-
-$node_publisher->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$node_subscriber->stop('immediate');
-$node_publisher->stop('immediate');
-
-$node_publisher->start;
-$node_subscriber->start;
-
-# commit post the restart
-$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
-$node_publisher->wait_for_catchup($appname);
-
-# check inserts are visible
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults');
-
-###############################
-# Do INSERT after the PREPARE but before ROLLBACK PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. A single row INSERT is done which is after the PREPARE.
-# 4. Then do a ROLLBACK PREPARED.
-#
-# Expect the 2PC data rolls back leaving only 3 rows on the subscriber
-# (the original 2 + inserted 1).
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
-$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# Insert a different record (now we are outside of the 2PC transaction)
-# Note: the 2PC transaction still holds row locks so make sure this insert is for a separate primary key
-$node_publisher->safe_psql('postgres', "INSERT INTO test_tab VALUES (99999, 'foobar')");
-
-# 2PC transaction gets aborted
-$node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is aborted on subscriber,
-# but the extra INSERT outside of the 2PC still was replicated
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3|3|3), 'check the outside insert was copied to subscriber');
-
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is aborted on subscriber');
-
-###############################
-# Do INSERT after the PREPARE but before COMMIT PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. A single row INSERT is done which is after the PREPARE.
-# 4. Then do a COMMIT PREPARED.
-#
-# Expect 2PC data + the extra row are on the subscriber
-# (the 3334 + inserted 1 = 3335).
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
-$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql('postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# Insert a different record (now we are outside of the 2PC transaction)
-# Note: the 2PC transaction still holds row locks so make sure this insert is for a separare primary key
-$node_publisher->safe_psql('postgres', "INSERT INTO test_tab VALUES (99999, 'foobar')");
-
-# 2PC transaction gets committed
-$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is committed on subscriber
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3335|3335|3335), 'Rows inserted by 2PC (as well as outside insert) have committed on subscriber, and extra columns contain local defaults');
+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(streaming = apply)"
+);
+$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 PUBLICATION";
-$result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber');
+test_streaming($node_publisher, $node_subscriber, $appname);
###############################
# check all the cleanup
--
2.18.4
[application/octet-stream] v5-0001-Perform-streaming-logical-transactions-by-background.patch (81.8K, ../OSZPR01MB6310F0FABB05F8E5BB31A5B0FDFC9@OSZPR01MB6310.jpnprd01.prod.outlook.com/3-v5-0001-Perform-streaming-logical-transactions-by-background.patch)
download | inline diff:
From ec33a270eb80330b0103d6051ae9bea0b2d6f7fd Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <[email protected]>
Date: Wed, 20 Apr 2022 16:45:07 +0800
Subject: [PATCH] Perform streaming logical transactions by background workers
Currently, for large transactions, the publisher sends the data in multiple
streams (changes divided into chunks depending upon logical_decoding_work_mem),
and then on the subscriber-side, the apply worker writes the changes into
temporary files and once it receives the commit, it reads from the file and
applies the entire transaction. To improve the performance of such
transactions, we can instead allow them to be applied via background workers.
In this approach, we assign a new apply background (if available) as soon as the
xact's first stream is received and the main apply worker will send changes to
this new worker via shared memory. The apply background will directly apply the
change instead of writing it to temporary files. We keep this worker assigned
till the transaction commit is received and also wait for the worker to finish
at commit. This preserves commit ordering and avoids writing to and reading
from file in most cases. We still need to spill if there is no worker
available. We also need to allow stream_stop to complete by the apply background
to finish it to avoid deadlocks because T-1's current stream of changes can
update rows in conflicting order with T-2's next stream of changes.
This patch also extends the subscription streaming option so that user can
control whether apply the streaming transaction in a apply background or spill
the change to disk. User can set the streaming option to 'on/off', 'apply'. For
now, 'apply' means the streaming will be applied via a apply background if
available. 'on' means the streaming transaction will be spilled to disk.
---
doc/src/sgml/catalogs.sgml | 10 +-
doc/src/sgml/ref/create_subscription.sgml | 24 +-
src/backend/commands/subscriptioncmds.c | 66 +-
src/backend/postmaster/bgworker.c | 3 +
src/backend/replication/logical/launcher.c | 84 +-
src/backend/replication/logical/origin.c | 10 +-
src/backend/replication/logical/proto.c | 7 +-
src/backend/replication/logical/tablesync.c | 10 +-
src/backend/replication/logical/worker.c | 1454 ++++++++++++++++++++++++---
src/backend/utils/activity/wait_event.c | 3 +
src/bin/pg_dump/pg_dump.c | 6 +-
src/include/catalog/pg_subscription.h | 16 +-
src/include/replication/logicalproto.h | 4 +-
src/include/replication/logicalworker.h | 1 +
src/include/replication/origin.h | 2 +-
src/include/replication/worker_internal.h | 7 +-
src/include/utils/wait_event.h | 1 +
src/test/regress/expected/subscription.out | 8 +-
src/tools/pgindent/typedefs.list | 2 +
19 files changed, 1506 insertions(+), 212 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index a533a21..ed61f99 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7863,11 +7863,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>substream</structfield> <type>bool</type>
+ <structfield>substream</structfield> <type>char</type>
</para>
<para>
- If true, the subscription will allow streaming of in-progress
- transactions
+ Controls how to handle the streaming of in-progress transactions.
+ <literal>f</literal> = disallow streaming of in-progress transactions
+ <literal>o</literal> = spill the changes of in-progress transactions to
+ disk and apply at once after the transaction is committed on the
+ publisher.
+ <literal>a</literal> = apply changes directly using a background worker
</para></entry>
</row>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 203bb41..8441aa0 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -217,13 +217,29 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</varlistentry>
<varlistentry>
- <term><literal>streaming</literal> (<type>boolean</type>)</term>
+ <term><literal>streaming</literal> (<type>enum</type>)</term>
<listitem>
<para>
Specifies whether to enable streaming of in-progress transactions
- for this subscription. By default, all transactions
- are fully decoded on the publisher and only then sent to the
- subscriber as a whole.
+ for this subscription. The default value is <literal>off</literal>,
+ all transactions are fully decoded on the publisher and only then
+ sent to the subscriber as a whole.
+ </para>
+
+ <para>
+ If set to <literal>on</literal>, the changes of transaction are
+ written to temporary files and then applied at once after the
+ transaction is committed on the publisher.
+ </para>
+
+ <para>
+ If set to <literal>apply</literal> incoming
+ changes are directly applied via one of the background worker, if
+ available. If no background worker is free to handle streaming
+ transaction then the changes are written to a file and applied after
+ the transaction is committed. Note that if error happen when applying
+ changes in background worker, it might not report the finish LSN of
+ the remote transaction in server log.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b94236f..fe1d373 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -83,7 +83,7 @@ typedef struct SubOpts
bool copy_data;
bool refresh;
bool binary;
- bool streaming;
+ char streaming;
bool twophase;
bool disableonerr;
XLogRecPtr lsn;
@@ -96,6 +96,62 @@ static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname,
/*
+ * Extract the streaming mode value from a DefElem. This is like
+ * defGetBoolean() but also accepts the special value and "apply".
+ */
+static char
+defGetStreamingMode(DefElem *def)
+{
+ /*
+ * If no parameter given, assume "true" is meant.
+ */
+ if (def->arg == NULL)
+ return SUBSTREAM_ON;
+
+ /*
+ * Allow 0, 1, "true", "false", "on", "off" or "apply".
+ */
+ switch (nodeTag(def->arg))
+ {
+ case T_Integer:
+ switch (intVal(def->arg))
+ {
+ case 0:
+ return SUBSTREAM_OFF;
+ case 1:
+ return SUBSTREAM_ON;
+ default:
+ /* otherwise, error out below */
+ break;
+ }
+ break;
+ default:
+ {
+ char *sval = defGetString(def);
+
+ /*
+ * The set of strings accepted here should match up with the
+ * grammar's opt_boolean_or_string production.
+ */
+ if (pg_strcasecmp(sval, "true") == 0 ||
+ pg_strcasecmp(sval, "on") == 0)
+ return SUBSTREAM_ON;
+ if (pg_strcasecmp(sval, "apply") == 0)
+ return SUBSTREAM_APPLY;
+ if (pg_strcasecmp(sval, "false") == 0 ||
+ pg_strcasecmp(sval, "off") == 0)
+ return SUBSTREAM_OFF;
+ }
+ break;
+ }
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s requires a Boolean value or \"apply\"",
+ def->defname)));
+ return SUBSTREAM_OFF; /* keep compiler quiet */
+}
+
+/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
*
* Since not all options can be specified in both commands, this function
@@ -132,7 +188,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
if (IsSet(supported_opts, SUBOPT_BINARY))
opts->binary = false;
if (IsSet(supported_opts, SUBOPT_STREAMING))
- opts->streaming = false;
+ opts->streaming = SUBSTREAM_OFF;
if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
opts->twophase = false;
if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
@@ -233,7 +289,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
errorConflictingDefElem(defel, pstate);
opts->specified_opts |= SUBOPT_STREAMING;
- opts->streaming = defGetBoolean(defel);
+ opts->streaming = defGetStreamingMode(defel);
}
else if (strcmp(defel->defname, "two_phase") == 0)
{
@@ -601,7 +657,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
- values[Anum_pg_subscription_substream - 1] = BoolGetDatum(opts.streaming);
+ values[Anum_pg_subscription_substream - 1] = CharGetDatum(opts.streaming);
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
LOGICALREP_TWOPHASE_STATE_PENDING :
@@ -1060,7 +1116,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
{
values[Anum_pg_subscription_substream - 1] =
- BoolGetDatum(opts.streaming);
+ CharGetDatum(opts.streaming);
replaces[Anum_pg_subscription_substream - 1] = true;
}
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 30682b6..070d778 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -128,6 +128,9 @@ static const struct
},
{
"ApplyWorkerMain", ApplyWorkerMain
+ },
+ {
+ "ApplyBgworkerMain", ApplyBgworkerMain
}
};
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 0adb2d1..299634c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -72,6 +72,7 @@ static void logicalrep_launcher_onexit(int code, Datum arg);
static void logicalrep_worker_onexit(int code, Datum arg);
static void logicalrep_worker_detach(void);
static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
+static void logicalrep_worker_stop_internal(LogicalRepWorker *worker);
static bool on_commit_launcher_wakeup = false;
@@ -224,6 +225,10 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+ /* We only need main apply worker or table sync worker here */
+ if (w->subworker)
+ continue;
+
if (w->in_use && w->subid == subid && w->relid == relid &&
(!only_running || w->proc))
{
@@ -262,9 +267,9 @@ logicalrep_workers_find(Oid subid, bool only_running)
/*
* Start new apply background worker, if possible.
*/
-void
+bool
logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
- Oid relid)
+ Oid relid, dsm_handle subworker_dsm)
{
BackgroundWorker bgw;
BackgroundWorkerHandle *bgw_handle;
@@ -275,6 +280,9 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
int nsyncworkers;
TimestampTz now;
+ /* We don't support table sync in subworker */
+ Assert(!((subworker_dsm != DSM_HANDLE_INVALID) && OidIsValid(relid)));
+
ereport(DEBUG1,
(errmsg_internal("starting logical replication worker for subscription \"%s\"",
subname)));
@@ -351,7 +359,7 @@ retry:
if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
{
LWLockRelease(LogicalRepWorkerLock);
- return;
+ return false;
}
/*
@@ -365,7 +373,7 @@ retry:
(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
errmsg("out of logical replication worker slots"),
errhint("You might need to increase max_logical_replication_workers.")));
- return;
+ return false;
}
/* Prepare the worker slot. */
@@ -380,6 +388,7 @@ retry:
worker->relstate = SUBREL_STATE_UNKNOWN;
worker->relstate_lsn = InvalidXLogRecPtr;
worker->stream_fileset = NULL;
+ worker->subworker = (subworker_dsm != DSM_HANDLE_INVALID);
worker->last_lsn = InvalidXLogRecPtr;
TIMESTAMP_NOBEGIN(worker->last_send_time);
TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -397,19 +406,31 @@ retry:
BGWORKER_BACKEND_DATABASE_CONNECTION;
bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
- snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+
+ if (subworker_dsm == DSM_HANDLE_INVALID)
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+ else
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyBgworkerMain");
+
+
if (OidIsValid(relid))
snprintf(bgw.bgw_name, BGW_MAXLEN,
"logical replication worker for subscription %u sync %u", subid, relid);
- else
+ else if (subworker_dsm == DSM_HANDLE_INVALID)
snprintf(bgw.bgw_name, BGW_MAXLEN,
"logical replication worker for subscription %u", subid);
+ else
+ snprintf(bgw.bgw_name, BGW_MAXLEN,
+ "logical replication apply worker for subscription %u", subid);
snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
bgw.bgw_restart_time = BGW_NEVER_RESTART;
bgw.bgw_notify_pid = MyProcPid;
bgw.bgw_main_arg = Int32GetDatum(slot);
+ if (subworker_dsm != DSM_HANDLE_INVALID)
+ memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
+
if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
{
/* Failed to start worker, so clean up the worker slot. */
@@ -422,11 +443,13 @@ retry:
(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
errmsg("out of background worker slots"),
errhint("You might need to increase max_worker_processes.")));
- return;
+ return false;
}
/* Now wait until it attaches. */
WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
+
+ return true;
}
/*
@@ -437,7 +460,6 @@ void
logicalrep_worker_stop(Oid subid, Oid relid)
{
LogicalRepWorker *worker;
- uint16 generation;
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
@@ -450,6 +472,22 @@ logicalrep_worker_stop(Oid subid, Oid relid)
return;
}
+ logicalrep_worker_stop_internal(worker);
+
+ LWLockRelease(LogicalRepWorkerLock);
+}
+
+/*
+ * Workhorse for logicalrep_worker_stop() and logicalrep_worker_detach(). Stop
+ * the worker and wait for wait for it to die.
+ */
+static void
+logicalrep_worker_stop_internal(LogicalRepWorker *worker)
+{
+ uint16 generation;
+
+ Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
/*
* Remember which generation was our worker so we can check if what we see
* is still the same one.
@@ -486,10 +524,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
* different, meaning that a different worker has taken the slot.
*/
if (!worker->in_use || worker->generation != generation)
- {
- LWLockRelease(LogicalRepWorkerLock);
return;
- }
/* Worker has assigned proc, so it has started. */
if (worker->proc)
@@ -523,8 +558,6 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
}
-
- LWLockRelease(LogicalRepWorkerLock);
}
/*
@@ -600,6 +633,28 @@ logicalrep_worker_attach(int slot)
static void
logicalrep_worker_detach(void)
{
+ /*
+ * If we are the main apply worker, stop all the apply background workers
+ * we started before.
+ */
+ if (!MyLogicalRepWorker->subworker)
+ {
+ List *workers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+ workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true);
+ foreach(lc, workers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+ if (w->subworker)
+ logicalrep_worker_stop_internal(w);
+ }
+
+ LWLockRelease(LogicalRepWorkerLock);
+ }
+
/* Block concurrent access. */
LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
@@ -622,6 +677,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
worker->userid = InvalidOid;
worker->subid = InvalidOid;
worker->relid = InvalidOid;
+ worker->subworker = false;
}
/*
@@ -869,7 +925,7 @@ ApplyLauncherMain(Datum main_arg)
wait_time = wal_retrieve_retry_interval;
logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
- sub->owner, InvalidOid);
+ sub->owner, InvalidOid, DSM_HANDLE_INVALID);
}
}
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index b0c8b6c..c6ea68a 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1068,7 +1068,7 @@ ReplicationOriginExitCleanup(int code, Datum arg)
* with replorigin_session_reset().
*/
void
-replorigin_session_setup(RepOriginId node)
+replorigin_session_setup(RepOriginId node, bool acquire)
{
static bool registered_cleanup;
int i;
@@ -1110,7 +1110,11 @@ replorigin_session_setup(RepOriginId node)
if (curstate->roident != node)
continue;
- else if (curstate->acquired_by != 0)
+ /*
+ * We allow the apply worker to get the slot which is acquired by its
+ * leader process.
+ */
+ else if (curstate->acquired_by != 0 && acquire)
{
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
@@ -1321,7 +1325,7 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
origin = replorigin_by_name(name, false);
- replorigin_session_setup(origin);
+ replorigin_session_setup(origin, true);
replorigin_session_origin = origin;
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index ff8513e..0409fde 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -1138,14 +1138,11 @@ logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn,
/*
* Read STREAM COMMIT from the output stream.
*/
-TransactionId
+void
logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
{
- TransactionId xid;
uint8 flags;
- xid = pq_getmsgint(in, 4);
-
/* read flags (unused for now) */
flags = pq_getmsgbyte(in);
@@ -1156,8 +1153,6 @@ logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
commit_data->commit_lsn = pq_getmsgint64(in);
commit_data->end_lsn = pq_getmsgint64(in);
commit_data->committime = pq_getmsgint64(in);
-
- return xid;
}
/*
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 49ceec3..04c9f96 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -568,7 +568,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
MySubscription->oid,
MySubscription->name,
MyLogicalRepWorker->userid,
- rstate->relid);
+ rstate->relid,
+ DSM_HANDLE_INVALID);
hentry->last_start_time = now;
}
}
@@ -589,6 +590,9 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
void
process_syncing_tables(XLogRecPtr current_lsn)
{
+ if (MyLogicalRepWorker->subworker)
+ return;
+
if (am_tablesync_worker())
process_syncing_tables_for_sync(current_lsn);
else
@@ -1275,7 +1279,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
* time this tablesync was launched.
*/
originid = replorigin_by_name(originname, false);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
*origin_startpos = replorigin_session_get_progress(false);
@@ -1386,7 +1390,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
true /* go backward */ , true /* WAL log */ );
UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
}
else
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 7da7823..a317f72 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -22,8 +22,25 @@
* STREAMED TRANSACTIONS
* ---------------------
* Streamed transactions (large transactions exceeding a memory limit on the
- * upstream) are not applied immediately, but instead, the data is written
- * to temporary files and then applied at once when the final commit arrives.
+ * upstream) are applied via one of two approaches.
+ *
+ * 1) Separate background workers
+ *
+ * Assign a new bgworker (if available) as soon as the xact's first stream is
+ * received and the main apply worker will send changes to this new worker via
+ * shared memory. We keep this worker assigned till the transaction commit is
+ * received and also wait for the worker to finish at commit. This preserves
+ * commit ordering and avoids writing to and reading from file in most cases.
+ * We still need to spill if there is no worker available. We also need to
+ * allow stream_stop to complete by the background worker to finish it to avoid
+ * deadlocks because T-1's current stream of changes can update rows in
+ * conflicting order with T-2's next stream of changes.
+ *
+ * 2) Write to temporary files and apply when the final commit arrives
+ *
+ * If no worker is available to handle streamed transaction, the data is
+ * written to temporary files and then applied at once when the final commit
+ * arrives.
*
* Unlike the regular (non-streamed) case, handling streamed transactions has
* to handle aborts of both the toplevel transaction and subtransactions. This
@@ -174,11 +191,15 @@
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
#include "storage/bufmgr.h"
+#include "storage/dsm.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
+#include "storage/spin.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
@@ -198,6 +219,75 @@
#define NAPTIME_PER_CYCLE 1000 /* max sleep time between cycles (1s) */
+#define PG_LOGICAL_APPLY_SHM_MAGIC 0x79fb2447 // TODO Consider change
+
+/*
+ * Shared information among apply workers.
+ */
+typedef struct ParallelState
+{
+ slock_t mutex;
+
+ /* State for apply background worker. */
+ char state;
+
+ TransactionId stream_xid;
+ uint32 n; /* id of apply background worker */
+} ParallelState;
+
+/*
+ * States for apply background worker.
+ */
+#define APPLY_BGWORKER_ATTACHED 'a'
+#define APPLY_BGWORKER_READY 'r'
+#define APPLY_BGWORKER_BUSY 'b'
+#define APPLY_BGWORKER_FINISHED 'f'
+#define APPLY_BGWORKER_EXIT 'e'
+
+/*
+ * Struct for maintaining a apply background worker.
+ */
+typedef struct WorkerState
+{
+ shm_mq_handle *mq_handle;
+ dsm_segment *dsm_seg;
+ ParallelState volatile *pstate;
+} WorkerState;
+
+typedef struct WorkerEntry
+{
+ TransactionId xid;
+ WorkerState *wstate;
+} WorkerEntry;
+
+/* Apply background workers hash table (initialized on first use) */
+static HTAB *ApplyWorkersHash = NULL;
+static List *ApplyWorkersIdleList = NIL;
+static List *ApplyWorkersList = NIL;
+
+/* Fields valid only for apply background workers */
+volatile ParallelState *MyParallelState = NULL;
+static List *subxactlist = NIL;
+
+/* The number of changes during one streaming block */
+static uint32 nchanges = 0;
+
+/* Worker setup and interactions */
+static WorkerState *apply_bgworker_setup(void);
+static WorkerState *find_or_start_apply_bgworker(TransactionId xid,
+ bool start);
+static void apply_bgworker_setup_dsm(WorkerState *wstate);
+static void apply_bgworker_wait_for(WorkerState *wstate,
+ char wait_for_state);
+static void apply_bgworker_send_data(WorkerState *wstate, Size nbytes,
+ const void *data);
+static void apply_bgworker_free(WorkerState *wstate);
+static void apply_bgworker_check_status(void);
+static void apply_bgworker_set_state(char state);
+
+#define am_apply_bgworker() (MyLogicalRepWorker->subworker)
+#define applying_changes_in_bgworker() (in_streamed_transaction && stream_apply_worker != NULL)
+
typedef struct FlushPosition
{
dlist_node node;
@@ -260,18 +350,20 @@ static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
static bool in_streamed_transaction = false;
static TransactionId stream_xid = InvalidTransactionId;
+static WorkerState *stream_apply_worker = NULL;
/*
* 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
* the transaction even if pg_subscription is updated and MySubscription->skiplsn
- * gets changed or reset during that. Also, in streaming transaction cases, we
- * don't skip receiving and spooling the changes since we decide whether or not
+ * gets changed or reset during that. Also, in streaming transaction cases (streaming = on),
+ * we don't skip receiving and spooling the changes since we decide whether or not
* to skip applying the changes when starting to apply changes. The subskiplsn is
* cleared after successfully skipping the transaction or applying non-empty
* transaction. The latter prevents the mistakenly specified subskiplsn from
- * being left.
+ * being left. Note that we cannot skip the streaming transaction in apply mode,
+ * because we cannot get the finish LSN before applying the changes.
*/
static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
#define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
@@ -426,43 +518,103 @@ end_replication_step(void)
}
/*
- * Handle streamed transactions.
+ * Handle streamed transactions for both main apply worker and apply background
+ * worker.
*
- * If in streaming mode (receiving a block of streamed transaction), we
- * simply redirect it to a file for the proper toplevel transaction.
+ * For the main apply worker, if in streaming mode (receiving a block of
+ * streamed transaction), we send the data to the apply background worker.
+ *
+ * For the apply background worker, define a savepoint if new subtransaction
+ * was started.
*
* Returns true for streamed transactions, false otherwise (regular mode).
*/
static bool
handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
{
- TransactionId xid;
+ TransactionId current_xid = InvalidTransactionId;
- /* not in streaming mode */
- if (!in_streamed_transaction)
+ /*
+ * Return if we are not in streaming mode and are not in a apply background
+ * worker.
+ */
+ if (!in_streamed_transaction && !am_apply_bgworker())
return false;
- Assert(stream_fd != NULL);
Assert(TransactionIdIsValid(stream_xid));
/*
* We should have received XID of the subxact as the first part of the
* message, so extract it.
*/
- xid = pq_getmsgint(s, 4);
+ current_xid = pq_getmsgint(s, 4);
- if (!TransactionIdIsValid(xid))
+ if (!TransactionIdIsValid(current_xid))
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
- /* Add the new subxact to the array (unless already there). */
- subxact_info_add(xid);
+ if (am_apply_bgworker())
+ {
+ /*
+ * Inside apply background worker we can figure out that new
+ * subtransaction was started if new change arrived with different xid.
+ * In that case we can define named savepoint, so that we were able to
+ * commit/rollback it separately later.
+ *
+ * Special case is if the first change comes from subtransaction, then
+ * we check that current_xid differs from stream_xid.
+ */
+ if (current_xid != stream_xid &&
+ !list_member_int(subxactlist, (int) current_xid))
+ {
+ MemoryContext oldctx;
+ char spname[MAXPGPATH];
+
+ snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", current_xid);
+
+ elog(LOG, "[Apply BGW #%u] defining savepoint %s",
+ MyParallelState->n, spname);
+
+ DefineSavepoint(spname);
+ CommitTransactionCommand();
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+ subxactlist = lappend_int(subxactlist, (int) current_xid);
+ MemoryContextSwitchTo(oldctx);
+ }
+ }
+ else if (applying_changes_in_bgworker())
+ {
+ /*
+ * If we decided to apply the changes of this transaction in a apply
+ * background worker, pass the data to the worker.
+ */
+ apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
+ nchanges += 1;
+
+ /*
+ * XXX The publisher side doesn't always send relation update message
+ * after the streaming transaction, so update the relation in main
+ * apply worker here.
+ */
+ if (action == LOGICAL_REP_MSG_RELATION)
+ {
+ LogicalRepRelation *rel = logicalrep_read_rel(s);
+ logicalrep_relmap_update(rel);
+ }
+
+ }
+ else
+ {
+ /* Add the new subxact to the array (unless already there). */
+ subxact_info_add(current_xid);
- /* write the change to the current file */
- stream_write_change(action, s);
+ /* write the change to the current file */
+ stream_write_change(action, s);
+ }
- return true;
+ return !am_apply_bgworker();
}
/*
@@ -844,6 +996,9 @@ apply_handle_commit(StringInfo s)
apply_handle_commit_internal(&commit_data);
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
@@ -898,7 +1053,9 @@ apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
* BeginTransactionBlock is necessary to balance the EndTransactionBlock
* called within the PrepareTransactionBlock below.
*/
- BeginTransactionBlock();
+ if (!IsTransactionBlock())
+ BeginTransactionBlock();
+
CommitTransactionCommand(); /* Completes the preceding Begin command. */
/*
@@ -950,6 +1107,9 @@ apply_handle_prepare(StringInfo s)
in_remote_transaction = false;
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -974,32 +1134,57 @@ apply_handle_commit_prepared(StringInfo s)
{
LogicalRepCommitPreparedTxnData prepare_data;
char gid[GIDSIZE];
+ WorkerState *wstate;
logicalrep_read_commit_prepared(s, &prepare_data);
set_apply_error_context_xact(prepare_data.xid, prepare_data.commit_lsn);
- /* Compute GID for two_phase transactions. */
- TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid,
- gid, sizeof(gid));
+ /* Check if we have prepared transaction in another bgworker */
+ wstate = find_or_start_apply_bgworker(prepare_data.xid, false);
- /* There is no transaction when COMMIT PREPARED is called */
- begin_replication_step();
+ if (wstate)
+ {
+ elog(DEBUG1, "received commit for streamed transaction %u", prepare_data.xid);
- /*
- * Update origin state so we can restart streaming from correct position
- * in case of crash.
- */
- replorigin_session_origin_lsn = prepare_data.end_lsn;
- replorigin_session_origin_timestamp = prepare_data.commit_time;
+ /* Send commit message */
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ /* Wait for bgworker to finish */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+
+ apply_bgworker_free(wstate);
+ }
+ else
+ {
+ /* Compute GID for two_phase transactions. */
+ TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid,
+ gid, sizeof(gid));
+
+ /* There is no transaction when COMMIT PREPARED is called */
+ begin_replication_step();
+
+ /*
+ * Update origin state so we can restart streaming from correct position
+ * in case of crash.
+ */
+ replorigin_session_origin_lsn = prepare_data.end_lsn;
+ replorigin_session_origin_timestamp = prepare_data.commit_time;
+
+ FinishPreparedTransaction(gid, true);
+ end_replication_step();
+ CommitTransactionCommand();
+
+ apply_bgworker_set_state(APPLY_BGWORKER_FINISHED);
+ }
- FinishPreparedTransaction(gid, true);
- end_replication_step();
- CommitTransactionCommand();
pgstat_report_stat(false);
store_flush_position(prepare_data.end_lsn);
in_remote_transaction = false;
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1017,37 +1202,57 @@ apply_handle_rollback_prepared(StringInfo s)
{
LogicalRepRollbackPreparedTxnData rollback_data;
char gid[GIDSIZE];
+ WorkerState *wstate;
logicalrep_read_rollback_prepared(s, &rollback_data);
set_apply_error_context_xact(rollback_data.xid, rollback_data.rollback_end_lsn);
- /* Compute GID for two_phase transactions. */
- TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
- gid, sizeof(gid));
+ /* Check if we are processing the prepared transaction in a bgworker */
+ wstate = find_or_start_apply_bgworker(rollback_data.xid, false);
- /*
- * It is possible that we haven't received prepare because it occurred
- * before walsender reached a consistent point or the two_phase was still
- * not enabled by that time, so in such cases, we need to skip rollback
- * prepared.
- */
- if (LookupGXact(gid, rollback_data.prepare_end_lsn,
- rollback_data.prepare_time))
+ if (wstate)
+ {
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ /* Wait for bgworker to finish */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+
+ elog(LOG, "rollback prepared streaming of xid %u", rollback_data.xid);
+
+ apply_bgworker_free(wstate);
+ }
+ else
{
+ /* Compute GID for two_phase transactions. */
+ TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
+ gid, sizeof(gid));
+
/*
- * Update origin state so we can restart streaming from correct
- * position in case of crash.
+ * It is possible that we haven't received prepare because it occurred
+ * before walsender reached a consistent point or the two_phase was still
+ * not enabled by that time, so in such cases, we need to skip rollback
+ * prepared.
*/
- replorigin_session_origin_lsn = rollback_data.rollback_end_lsn;
- replorigin_session_origin_timestamp = rollback_data.rollback_time;
+ if (LookupGXact(gid, rollback_data.prepare_end_lsn,
+ rollback_data.prepare_time))
+ {
+ /*
+ * Update origin state so we can restart streaming from correct
+ * position in case of crash.
+ */
+ replorigin_session_origin_lsn = rollback_data.rollback_end_lsn;
+ replorigin_session_origin_timestamp = rollback_data.rollback_time;
- /* There is no transaction when ABORT/ROLLBACK PREPARED is called */
- begin_replication_step();
- FinishPreparedTransaction(gid, false);
- end_replication_step();
- CommitTransactionCommand();
+ /* There is no transaction when ABORT/ROLLBACK PREPARED is called */
+ begin_replication_step();
+ FinishPreparedTransaction(gid, false);
+ end_replication_step();
+ CommitTransactionCommand();
+
+ clear_subscription_skip_lsn(rollback_data.rollback_end_lsn);
+ }
- clear_subscription_skip_lsn(rollback_data.rollback_end_lsn);
+ apply_bgworker_set_state(APPLY_BGWORKER_FINISHED);
}
pgstat_report_stat(false);
@@ -1055,6 +1260,9 @@ apply_handle_rollback_prepared(StringInfo s)
store_flush_position(rollback_data.rollback_end_lsn);
in_remote_transaction = false;
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(rollback_data.rollback_end_lsn);
@@ -1063,11 +1271,131 @@ apply_handle_rollback_prepared(StringInfo s)
}
/*
- * Handle STREAM PREPARE.
+ * Look up worker inside ApplyWorkersHash for requested xid.
*
- * Logic is in two parts:
- * 1. Replay all the spooled operations
- * 2. Mark the transaction as prepared
+ * If start flag is true, try to start a new worker if not found in hash table.
+ */
+static WorkerState *
+find_or_start_apply_bgworker(TransactionId xid, bool start)
+{
+ bool found;
+ WorkerState *wstate;
+ WorkerEntry *entry = NULL;
+
+ if (!TransactionIdIsValid(xid))
+ return NULL;
+
+ if (!start && ApplyWorkersHash == NULL)
+ return NULL;
+
+ /*
+ * We don't start new background worker if we are not in streaming apply
+ * mode.
+ */
+ if (MySubscription->stream != SUBSTREAM_APPLY)
+ return NULL;
+
+ /*
+ * We don't start new background worker if user has set skiplsn as it's
+ * possible that user want to skip the streaming transaction. For streaming
+ * transaction, we need to spill the transaction to disk so that we can get
+ * the last LSN of the transaction to judge whether to skip before starting
+ * to apply the change.
+ */
+ if (start && !XLogRecPtrIsInvalid(MySubscription->skiplsn))
+ return NULL;
+
+ /*
+ * For streaming transactions that are being applied in bgworker, we cannot
+ * decide whether to apply the change for a relation that is not in the
+ * READY state (see should_apply_changes_for_rel) as we won't know
+ * remote_final_lsn by that time. So, we don't start new bgworker in this
+ * case.
+ */
+ if (start && !AllTablesyncsReady())
+ return NULL;
+
+ /* First time through, initialize apply workers hashtable */
+ if (ApplyWorkersHash == NULL)
+ {
+ HASHCTL ctl;
+
+ MemSet(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(TransactionId);
+ ctl.entrysize = sizeof(WorkerEntry);
+ ctl.hcxt = ApplyContext;
+
+ ApplyWorkersHash = hash_create("logical apply workers hash", 8, &ctl,
+ HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+ }
+
+ /*
+ * Find entry for requested transaction.
+ */
+ entry = hash_search(ApplyWorkersHash, &xid, start ? HASH_ENTER : HASH_FIND,
+ &found);
+ if (found)
+ {
+ entry->wstate->pstate->state = APPLY_BGWORKER_BUSY;
+ return entry->wstate;
+ }
+ else if (!start)
+ return NULL;
+
+ /* If there is at least one worker in the idle list, then take one. */
+ if (list_length(ApplyWorkersIdleList) > 0)
+ {
+ wstate = (WorkerState *) llast(ApplyWorkersIdleList);
+ ApplyWorkersIdleList = list_delete_last(ApplyWorkersIdleList);
+ }
+ else
+ {
+ wstate = apply_bgworker_setup();
+
+ if (wstate == NULL)
+ {
+ /*
+ * If the bgworker cannot be launched, remove entry in hash table.
+ */
+ hash_search(ApplyWorkersHash, &xid, HASH_REMOVE, &found);
+ return NULL;
+ }
+ }
+
+ /* Fill up the hash entry */
+ wstate->pstate->state = APPLY_BGWORKER_BUSY;
+ wstate->pstate->stream_xid = xid;
+ entry->wstate = wstate;
+ entry->xid = xid;
+
+ return wstate;
+}
+
+/*
+ * Add the worker to the freelist and remove the entry from hash table.
+ */
+static void
+apply_bgworker_free(WorkerState *wstate)
+{
+ bool found;
+ MemoryContext oldctx;
+ TransactionId xid = wstate->pstate->stream_xid;
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ hash_search(ApplyWorkersHash, &xid,
+ HASH_REMOVE, &found);
+
+ elog(LOG, "adding finished apply worker #%u for xid %u to the idle list",
+ wstate->pstate->n, wstate->pstate->stream_xid);
+
+ ApplyWorkersIdleList = lappend(ApplyWorkersIdleList, wstate);
+
+ MemoryContextSwitchTo(oldctx);
+}
+
+/*
+ * Handle STREAM PREPARE.
*/
static void
apply_handle_stream_prepare(StringInfo s)
@@ -1088,24 +1416,71 @@ apply_handle_stream_prepare(StringInfo s)
logicalrep_read_stream_prepare(s, &prepare_data);
set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
- elog(DEBUG1, "received prepare for streamed transaction %u", prepare_data.xid);
+ /*
+ * If we are in a bgworker, just prepare the transaction.
+ */
+ if (am_apply_bgworker())
+ {
+ elog(LOG, "received prepare for streamed transaction %u", prepare_data.xid);
- /* Replay all the spooled operations. */
- apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+ /* Mark the transaction as prepared. */
+ apply_handle_prepare_internal(&prepare_data);
- /* Mark the transaction as prepared. */
- apply_handle_prepare_internal(&prepare_data);
+ CommitTransactionCommand();
- CommitTransactionCommand();
+ pgstat_report_stat(false);
- pgstat_report_stat(false);
+ list_free(subxactlist);
+ subxactlist = NIL;
- store_flush_position(prepare_data.end_lsn);
+ apply_bgworker_set_state(APPLY_BGWORKER_READY);
+ }
+ else
+ {
+ WorkerState *wstate = find_or_start_apply_bgworker(prepare_data.xid, false);
+
+ /*
+ * If we are in main apply worker, check if we are processing this
+ * transaction in a bgworker.
+ */
+ if (wstate)
+ {
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_READY);
+
+ pgstat_report_stat(false);
+ store_flush_position(prepare_data.end_lsn);
+ }
+ else
+ {
+ /*
+ * If we are in main apply worker and the transaction has been
+ * serialized to file, replay all the spooled operations.
+ */
+ apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+
+ /* Mark the transaction as prepared. */
+ apply_handle_prepare_internal(&prepare_data);
+
+ CommitTransactionCommand();
+
+ pgstat_report_stat(false);
+
+ store_flush_position(prepare_data.end_lsn);
+
+ in_remote_transaction = false;
+
+ /* unlink the files with serialized changes and subxact info. */
+ stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+ }
+ }
in_remote_transaction = false;
+ stream_apply_worker = NULL;
- /* unlink the files with serialized changes and subxact info. */
- stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1155,15 +1530,6 @@ apply_handle_stream_start(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("duplicate STREAM START message")));
- /*
- * Start a transaction on stream start, this transaction will be committed
- * on the stream stop unless it is a tablesync worker in which case it
- * will be committed after processing all the messages. We need the
- * transaction for handling the buffile, used for serializing the
- * streaming data and subxact info.
- */
- begin_replication_step();
-
/* notify handle methods we're processing a remote transaction */
in_streamed_transaction = true;
@@ -1177,36 +1543,90 @@ apply_handle_stream_start(StringInfo s)
set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
+ if (am_apply_bgworker())
+ {
+ /*
+ * Make sure the handle apply_dispatch methods are aware we're in a
+ * remote transaction.
+ */
+ in_remote_transaction = true;
+
+ /* If we are in a bgworker, begin the transaction */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+
+ StartTransactionCommand();
+ BeginTransactionBlock();
+ CommitTransactionCommand();
+
+ return;
+ }
+
/*
- * Initialize the worker's stream_fileset if we haven't yet. This will be
- * used for the entire duration of the worker so create it in a permanent
- * context. We create this on the very first streaming message from any
- * transaction and then use it for this and other streaming transactions.
- * Now, we could create a fileset at the start of the worker as well but
- * then we won't be sure that it will ever be used.
+ * If we are in main apply worker, check if there is any free bgworker
+ * we can use to process this transaction.
*/
- if (MyLogicalRepWorker->stream_fileset == NULL)
+ stream_apply_worker = find_or_start_apply_bgworker(stream_xid, first_segment);
+
+ if (applying_changes_in_bgworker())
{
- MemoryContext oldctx;
+ /*
+ * If we have free worker or we already started to apply this
+ * transaction in bgworker, we pass the data to worker.
+ */
+ if (first_segment)
+ apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
- oldctx = MemoryContextSwitchTo(ApplyContext);
+ nchanges = 0;
+ elog(LOG, "starting streaming of xid %u", stream_xid);
+ }
- MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
- FileSetInit(MyLogicalRepWorker->stream_fileset);
+ /*
+ * If no worker is available for the first stream start, we start to
+ * serialize all the changes of the transaction.
+ */
+ else
+ {
+ /*
+ * Start a transaction on stream start, this transaction will be committed
+ * on the stream stop unless it is a tablesync worker in which case it
+ * will be committed after processing all the messages. We need the
+ * transaction for handling the buffile, used for serializing the
+ * streaming data and subxact info.
+ */
+ begin_replication_step();
- MemoryContextSwitchTo(oldctx);
- }
+ /*
+ * Initialize the worker's stream_fileset if we haven't yet. This will be
+ * used for the entire duration of the worker so create it in a permanent
+ * context. We create this on the very first streaming message from any
+ * transaction and then use it for this and other streaming transactions.
+ * Now, we could create a fileset at the start of the worker as well but
+ * then we won't be sure that it will ever be used.
+ */
+ if (MyLogicalRepWorker->stream_fileset == NULL)
+ {
+ MemoryContext oldctx;
- /* open the spool file for this transaction */
- stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+ oldctx = MemoryContextSwitchTo(ApplyContext);
- /* if this is not the first segment, open existing subxact file */
- if (!first_segment)
- subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+ MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
+ FileSetInit(MyLogicalRepWorker->stream_fileset);
- pgstat_report_activity(STATE_RUNNING, NULL);
+ MemoryContextSwitchTo(oldctx);
+ }
- end_replication_step();
+ /* open the spool file for this transaction */
+ stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+
+ /* if this is not the first segment, open existing subxact file */
+ if (!first_segment)
+ subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+
+ end_replication_step();
+ }
+
+ pgstat_report_activity(STATE_RUNNING, NULL);
}
/*
@@ -1220,44 +1640,47 @@ apply_handle_stream_stop(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("STREAM STOP message without STREAM START")));
- /*
- * Close the file with serialized changes, and serialize information about
- * subxacts for the toplevel transaction.
- */
- subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
- stream_close_file();
+ if (applying_changes_in_bgworker())
+ {
+ char action = LOGICAL_REP_MSG_STREAM_STOP;
- /* We must be in a valid transaction state */
- Assert(IsTransactionState());
+ apply_bgworker_send_data(stream_apply_worker, 1, &action);
+ apply_bgworker_wait_for(stream_apply_worker, APPLY_BGWORKER_READY);
- /* Commit the per-stream transaction */
- CommitTransactionCommand();
+ elog(LOG, "stopped streaming of xid %u, %u changes streamed", stream_xid, nchanges);
+ }
+ else
+ {
+ /*
+ * Close the file with serialized changes, and serialize information
+ * about subxacts for the toplevel transaction.
+ */
+ subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
+ stream_close_file();
- in_streamed_transaction = false;
+ /* We must be in a valid transaction state */
+ Assert(IsTransactionState());
+
+ /* Commit the per-stream transaction */
+ CommitTransactionCommand();
- /* Reset per-stream context */
- MemoryContextReset(LogicalStreamingContext);
+ /* Reset per-stream context */
+ MemoryContextReset(LogicalStreamingContext);
+ }
+
+ in_streamed_transaction = false;
+ stream_apply_worker = NULL;
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
/*
- * Handle STREAM abort message.
+ * Handle STREAM ABORT message when the transaction was spilled to disk.
*/
static void
-apply_handle_stream_abort(StringInfo s)
+serialize_stream_abort(TransactionId xid, TransactionId subxid)
{
- TransactionId xid;
- TransactionId subxid;
-
- if (in_streamed_transaction)
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("STREAM ABORT message without STREAM STOP")));
-
- logicalrep_read_stream_abort(s, &xid, &subxid);
-
/*
* If the two XIDs are the same, it's in fact abort of toplevel xact, so
* just delete the files with serialized info.
@@ -1339,8 +1762,120 @@ apply_handle_stream_abort(StringInfo s)
end_replication_step();
CommitTransactionCommand();
}
+}
- reset_apply_error_context_info();
+/*
+ * Handle STREAM ABORT message.
+ */
+static void
+apply_handle_stream_abort(StringInfo s)
+{
+ TransactionId xid;
+ TransactionId subxid;
+
+ if (in_streamed_transaction)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("STREAM COMMIT message without STREAM STOP")));
+
+ logicalrep_read_stream_abort(s, &xid, &subxid);
+
+ if (am_apply_bgworker())
+ {
+ ereport(LOG,
+ (errmsg("[Apply BGW #%u] aborting current transaction xid=%u, subxid=%u",
+ MyParallelState->n, GetCurrentTransactionIdIfAny(), GetCurrentSubTransactionId())));
+
+ /*
+ * If the two XIDs are the same, it's in fact abort of toplevel xact,
+ * so just free the subxactlist.
+ */
+ if (subxid == xid)
+ {
+ set_apply_error_context_xact(subxid, InvalidXLogRecPtr);
+
+ AbortCurrentTransaction();
+
+ EndTransactionBlock(false);
+ CommitTransactionCommand();
+
+ in_remote_transaction = false;
+
+ list_free(subxactlist);
+ subxactlist = NIL;
+
+ apply_bgworker_set_state(APPLY_BGWORKER_FINISHED);
+ }
+ else
+ {
+ /*
+ * OK, so it's a subxact. Rollback to the savepoint.
+ *
+ * We also need to read the subxactlist, determine the offset
+ * tracked for the subxact, and truncate the list.
+ */
+ int i;
+ bool found = false;
+ char spname[MAXPGPATH];
+
+ set_apply_error_context_xact(subxid, InvalidXLogRecPtr);
+ snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", subxid);
+
+ ereport(LOG,
+ (errmsg("[Apply BGW #%u] rolling back to savepoint %s",
+ MyParallelState->n, spname)));
+
+ for(i = list_length(subxactlist) - 1; i >= 0; i--)
+ {
+ xid = (TransactionId) list_nth_int(subxactlist, i);
+ if (xid == subxid)
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (found)
+ {
+ elog(LOG, "rolled back to savepoint %s", spname);
+ RollbackToSavepoint(spname);
+ CommitTransactionCommand();
+ subxactlist = list_truncate(subxactlist, i + 1);
+ }
+
+ apply_bgworker_set_state(APPLY_BGWORKER_READY);
+ }
+
+ reset_apply_error_context_info();
+ }
+ else
+ {
+ WorkerState *wstate = find_or_start_apply_bgworker(xid, false);
+
+ /*
+ * If we are in main apply worker, check if we are processing this
+ * transaction in a bgworker.
+ */
+ if (wstate)
+ {
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ if (subxid == xid)
+ {
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+ apply_bgworker_free(wstate);
+ }
+ else
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_READY);
+ }
+
+ /*
+ * We are in main apply worker and the transaction has been serialized
+ * to file.
+ */
+ else
+ serialize_stream_abort(xid, subxid);
+ }
}
/*
@@ -1463,40 +1998,6 @@ apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)
}
/*
- * Handle STREAM COMMIT message.
- */
-static void
-apply_handle_stream_commit(StringInfo s)
-{
- TransactionId xid;
- LogicalRepCommitData commit_data;
-
- if (in_streamed_transaction)
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("STREAM COMMIT message without STREAM STOP")));
-
- xid = logicalrep_read_stream_commit(s, &commit_data);
- set_apply_error_context_xact(xid, commit_data.commit_lsn);
-
- elog(DEBUG1, "received commit for streamed transaction %u", xid);
-
- apply_spooled_messages(xid, commit_data.commit_lsn);
-
- apply_handle_commit_internal(&commit_data);
-
- /* unlink the files with serialized changes and subxact info */
- stream_cleanup_files(MyLogicalRepWorker->subid, xid);
-
- /* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(commit_data.end_lsn);
-
- pgstat_report_activity(STATE_IDLE, NULL);
-
- reset_apply_error_context_info();
-}
-
-/*
* Helper function for apply_handle_commit and apply_handle_stream_commit.
*/
static void
@@ -2445,6 +2946,101 @@ apply_handle_truncate(StringInfo s)
end_replication_step();
}
+/*
+ * Handle STREAM COMMIT message.
+ */
+static void
+apply_handle_stream_commit(StringInfo s)
+{
+ LogicalRepCommitData commit_data;
+ TransactionId xid;
+
+ if (in_streamed_transaction)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("STREAM COMMIT message without STREAM STOP")));
+
+ xid = pq_getmsgint(s, 4);
+ logicalrep_read_stream_commit(s, &commit_data);
+ set_apply_error_context_xact(xid, commit_data.commit_lsn);
+
+ elog(DEBUG1, "received commit for streamed transaction %u", xid);
+
+ if (am_apply_bgworker())
+ {
+ /*
+ * Update origin state so we can restart streaming from correct
+ * position in case of crash.
+ */
+ replorigin_session_origin_lsn = commit_data.end_lsn;
+ replorigin_session_origin_timestamp = commit_data.committime;
+
+ CommitTransactionCommand();
+ EndTransactionBlock(false);
+ CommitTransactionCommand();
+
+ in_remote_transaction = false;
+
+ pgstat_report_stat(false);
+
+ list_free(subxactlist);
+ subxactlist = NIL;
+
+ apply_bgworker_set_state(APPLY_BGWORKER_FINISHED);
+ }
+ else
+ {
+ /*
+ * If we are in main apply worker, check if we are processing this
+ * transaction in a apply background worker.
+ */
+ WorkerState *wstate = find_or_start_apply_bgworker(xid, false);
+
+ if (wstate)
+ {
+ /* Send commit message */
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ /* Wait for apply background worker to finish */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+
+ pgstat_report_stat(false);
+ store_flush_position(commit_data.end_lsn);
+ stop_skipping_changes();
+
+ apply_bgworker_free(wstate);
+
+ /*
+ * The transaction is either non-empty or skipped, so we clear the
+ * subskiplsn.
+ */
+ clear_subscription_skip_lsn(commit_data.commit_lsn);
+ }
+ else
+ {
+ /*
+ * If we are in main apply worker and the transaction has been
+ * serialized to file, replay all the spooled operations.
+ */
+ apply_spooled_messages(xid, commit_data.commit_lsn);
+
+ apply_handle_commit_internal(&commit_data);
+
+ /* unlink the files with serialized changes and subxact info */
+ stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+ }
+ }
+
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
+ /* Process any tables that are being synchronized in parallel. */
+ process_syncing_tables(commit_data.end_lsn);
+
+ pgstat_report_activity(STATE_IDLE, NULL);
+
+ reset_apply_error_context_info();
+}
/*
* Logical replication protocol message dispatcher.
@@ -2511,6 +3107,8 @@ apply_dispatch(StringInfo s)
break;
case LOGICAL_REP_MSG_STREAM_START:
+ /* FIXME : log for debugging here. */
+ elog(LOG, "LOGICAL_REP_MSG_STREAM_START");
apply_handle_stream_start(s);
break;
@@ -2618,6 +3216,10 @@ store_flush_position(XLogRecPtr remote_lsn)
{
FlushPosition *flushpos;
+ /* We only need to collect the LSN in main apply worker */
+ if (am_apply_bgworker())
+ return;
+
/* Need to do this in permanent context */
MemoryContextSwitchTo(ApplyContext);
@@ -2794,6 +3396,9 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
AcceptInvalidationMessages();
maybe_reread_subscription();
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any table synchronization changes. */
process_syncing_tables(last_received);
}
@@ -3686,7 +4291,7 @@ ApplyWorkerMain(Datum main_arg)
originid = replorigin_by_name(originname, true);
if (!OidIsValid(originid))
originid = replorigin_create(originname);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -3733,7 +4338,7 @@ ApplyWorkerMain(Datum main_arg)
options.proto.logical.publication_names = MySubscription->publications;
options.proto.logical.binary = MySubscription->binary;
- options.proto.logical.streaming = MySubscription->stream;
+ options.proto.logical.streaming = (MySubscription->stream != SUBSTREAM_OFF);
options.proto.logical.twophase = false;
if (!am_tablesync_worker())
@@ -3891,7 +4496,8 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
XLogRecPtr myskiplsn = MySubscription->skiplsn;
bool started_tx = false;
- if (likely(XLogRecPtrIsInvalid(myskiplsn)))
+ if (likely(XLogRecPtrIsInvalid(myskiplsn)) ||
+ am_apply_bgworker())
return;
if (!IsTransactionState())
@@ -4027,3 +4633,529 @@ reset_apply_error_context_info(void)
apply_error_callback_arg.remote_attnum = -1;
set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
}
+
+/* Apply Background Worker main loop */
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ParallelState *pst)
+{
+ shm_mq_result shmq_res;
+ PGPROC *registrant;
+ ErrorContextCallback errcallback;
+ XLogRecPtr last_received = InvalidXLogRecPtr;
+
+ registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
+ SetLatch(®istrant->procLatch);
+
+ /*
+ * Push apply error context callback. Fields will be filled during
+ * applying a change.
+ */
+ errcallback.callback = apply_error_callback;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ for (;;)
+ {
+ void *data;
+ Size len;
+ StringInfoData s;
+ MemoryContext oldctx;
+ XLogRecPtr start_lsn;
+ XLogRecPtr end_lsn;
+ TimestampTz send_time;
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* Ensure we are reading the data into our memory context. */
+ oldctx = MemoryContextSwitchTo(ApplyMessageContext);
+
+ shmq_res = shm_mq_receive(mqh, &len, &data, false);
+
+ if (shmq_res != SHM_MQ_SUCCESS)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("lost connection to the main apply worker")));
+
+ if (len == 0)
+ {
+ elog(LOG, "[Apply BGW #%u] got zero-length message, stopping", pst->n);
+ break;
+ }
+
+ s.cursor = 0;
+ s.maxlen = -1;
+ s.data = (char *) data;
+ s.len = len;
+
+ /*
+ * We use first byte of message for additional communication between
+ * main Logical replication worker and apply bgworkers, so if it
+ * differs from 'w', then process it first.
+ */
+ switch (pq_getmsgbyte(&s))
+ {
+ /* End message of streaming chunk */
+ case LOGICAL_REP_MSG_STREAM_STOP:
+ elog(LOG, "[Apply BGW #%u] ended processing streaming chunk,"
+ "waiting on shm_mq_receive", pst->n);
+
+ apply_bgworker_set_state(APPLY_BGWORKER_READY);
+
+ SetLatch(®istrant->procLatch);
+
+ in_streamed_transaction = false;
+ pgstat_report_activity(STATE_IDLE, NULL);
+ continue;
+ case 'w':
+ break;
+ default:
+ elog(ERROR, "unexpected message");
+ break;
+ }
+
+ start_lsn = pq_getmsgint64(&s);
+ end_lsn = pq_getmsgint64(&s);
+ send_time = pq_getmsgint64(&s);
+
+ if (last_received < start_lsn)
+ last_received = start_lsn;
+
+ if (last_received < end_lsn)
+ last_received = end_lsn;
+
+ /*
+ * TO IMPROVE: Do we need to display the bgworker's information in
+ * pg_stat_replication ?
+ */
+ UpdateWorkerStats(last_received, send_time, false);
+
+ apply_dispatch(&s);
+
+ if (ConfigReloadPending)
+ {
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+ MemoryContextReset(ApplyMessageContext);
+ }
+
+ MemoryContextSwitchTo(TopMemoryContext);
+ MemoryContextReset(ApplyContext);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+
+ elog(LOG, "[Apply BGW #%u] exiting", pst->n);
+
+ /* Signal main process that we are done. */
+ SetLatch(®istrant->procLatch);
+}
+
+/*
+ * Set the failed flag so that the main apply worker can realize we have
+ * shutdown.
+ */
+static void
+ApplyBgwShutdown(int code, Datum arg)
+{
+ SpinLockAcquire(&MyParallelState->mutex);
+ MyParallelState->state = APPLY_BGWORKER_EXIT;
+ SpinLockRelease(&MyParallelState->mutex);
+
+ dsm_detach((dsm_segment *) DatumGetPointer(arg));
+}
+
+/*
+ * Apply Background Worker entry point
+ */
+void
+ApplyBgworkerMain(Datum main_arg)
+{
+ volatile ParallelState *pst;
+
+ dsm_handle handle;
+ dsm_segment *seg;
+ shm_toc *toc;
+ shm_mq *mq;
+ shm_mq_handle *mqh;
+ MemoryContext oldcontext;
+ RepOriginId originid;
+ int worker_slot = DatumGetInt32(main_arg);
+ char originname[NAMEDATALEN];
+
+ /* Attach to slot */
+ logicalrep_worker_attach(worker_slot);
+
+ MemoryContextSwitchTo(TopMemoryContext);
+
+ /* Load the subscription into persistent memory context. */
+ ApplyContext = AllocSetContextCreate(TopMemoryContext,
+ "ApplyContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /*
+ * Init the ApplyMessageContext which we clean up after each replication
+ * protocol message.
+ */
+ ApplyMessageContext = AllocSetContextCreate(ApplyContext,
+ "ApplyMessageContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /* Setup signal handling */
+ pqsignal(SIGHUP, SignalHandlerForConfigReload);
+ pqsignal(SIGTERM, die);
+ BackgroundWorkerUnblockSignals();
+
+ /*
+ * Connect to the dynamic shared memory segment.
+ *
+ * The backend that registered this worker passed us the ID of a shared
+ * memory segment to which we must attach for further instructions. In
+ * order to attach to dynamic shared memory, we need a resource owner.
+ * Once we've mapped the segment in our address space, attach to the table
+ * of contents so we can locate the various data structures we'll need to
+ * find within the segment.
+ */
+ CurrentResourceOwner = ResourceOwnerCreate(NULL, "Logical apply worker");
+ memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
+ seg = dsm_attach(handle);
+ if (seg == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("unable to map dynamic shared memory segment")));
+ toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
+ if (toc == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("bad magic number in dynamic shared memory segment")));
+
+ before_shmem_exit(ApplyBgwShutdown, PointerGetDatum(seg));
+
+ /* Look up the parallel state. */
+ pst = shm_toc_lookup(toc, 0, false);
+ MyParallelState = pst;
+
+ /*
+ * Attach to the message queue.
+ */
+ mq = shm_toc_lookup(toc, 1, false);
+ shm_mq_set_receiver(mq, MyProc);
+ mqh = shm_mq_attach(mq, seg, NULL);
+
+ /* Connect to our database. */
+ BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+ MyLogicalRepWorker->userid,
+ 0);
+
+ /*
+ * Set the client encoding to the database encoding, since that is what
+ * the leader will expect.
+ */
+ SetClientEncoding(GetDatabaseEncoding());
+
+ stream_xid = pst->stream_xid;
+
+ StartTransactionCommand();
+ oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+ MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
+ if (!MySubscription)
+ {
+ ereport(LOG,
+ (errmsg("logical replication apply worker for subscription %u will not "
+ "start because the subscription was removed during startup",
+ MyLogicalRepWorker->subid)));
+ proc_exit(0);
+ }
+
+ MySubscriptionValid = true;
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Setup synchronous commit according to the user's wishes */
+ SetConfigOption("synchronous_commit", MySubscription->synccommit,
+ PGC_BACKEND, PGC_S_OVERRIDE);
+
+ /* Keep us informed about subscription changes. */
+ CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
+ subscription_change_cb,
+ (Datum) 0);
+
+ CommitTransactionCommand();
+
+ /* Setup replication origin tracking. */
+ StartTransactionCommand();
+ snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
+ originid = replorigin_by_name(originname, true);
+ if (!OidIsValid(originid))
+ originid = replorigin_create(originname);
+ replorigin_session_setup(originid, false);
+ replorigin_session_origin = originid;
+ CommitTransactionCommand();
+
+ /*
+ * Allocate the origin name in long-lived context for error context
+ * message.
+ */
+ apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+ originname);
+
+ /*
+ * Indicate that we're fully initialized and ready to begin the main part
+ * of the apply operation.
+ */
+ apply_bgworker_set_state(APPLY_BGWORKER_ATTACHED);
+
+ elog(LOG, "[Apply BGW #%u] started", pst->n);
+
+ PG_TRY();
+ {
+ LogicalApplyBgwLoop(mqh, pst);
+ }
+ PG_CATCH();
+ {
+ /*
+ * Report the worker failed while applying streaming transaction
+ * changes. Abort the current transaction so that the stats message is
+ * sent in an idle state.
+ */
+ AbortOutOfAnyTransaction();
+ pgstat_report_subscription_error(MySubscription->oid, false);
+
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+
+ /*
+ * We're done. Explicitly detach the shared memory segment so that we
+ * don't get a resource leak warning at commit time. This will fire any
+ * on_dsm_detach callbacks we've registered, as well. Once that's done,
+ * we can go ahead and exit.
+ */
+ dsm_detach(seg);
+ proc_exit(0);
+}
+
+/*
+ * Set up a dynamic shared memory segment.
+ *
+ * We set up a control region that contains a ParallelState,
+ * plus one region per message queue. There are as many message queues as
+ * the number of workers.
+ */
+static void
+apply_bgworker_setup_dsm(WorkerState *wstate)
+{
+ shm_toc_estimator e;
+ int toc_key = 0;
+ Size segsize;
+ dsm_segment *seg;
+ shm_toc *toc;
+ ParallelState *pst;
+ shm_mq *mq;
+ int64 queue_size = 160000000; /* 16 MB for now */
+
+ /*
+ * Estimate how much shared memory we need.
+ *
+ * Because the TOC machinery may choose to insert padding of oddly-sized
+ * requests, we must estimate each chunk separately.
+ *
+ * We need one key to register the location of the header, and we need
+ * nworkers keys to track the locations of the message queues.
+ */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(ParallelState));
+ shm_toc_estimate_chunk(&e, (Size) queue_size);
+
+ shm_toc_estimate_keys(&e, 1 + 1);
+ segsize = shm_toc_estimate(&e);
+
+ /* Create the shared memory segment and establish a table of contents. */
+ seg = dsm_create(shm_toc_estimate(&e), 0);
+ toc = shm_toc_create(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg),
+ segsize);
+
+ /* Set up the header region. */
+ pst = shm_toc_allocate(toc, sizeof(ParallelState));
+ SpinLockInit(&pst->mutex);
+ pst->state = APPLY_BGWORKER_BUSY;
+ pst->stream_xid = stream_xid;
+ pst->n = list_length(ApplyWorkersList) + 1;
+
+ shm_toc_insert(toc, toc_key++, pst);
+
+ /* Set up one message queue per worker, plus one. */
+ mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
+ (Size) queue_size);
+ shm_toc_insert(toc, toc_key++, mq);
+ shm_mq_set_sender(mq, MyProc);
+
+ /* Attach the queues. */
+ wstate->mq_handle = shm_mq_attach(mq, seg, NULL);
+
+ /* Return results to caller. */
+ wstate->dsm_seg = seg;
+ wstate->pstate = pst;
+}
+
+/*
+ * Start apply worker background worker process and allocat shared memory for
+ * it.
+ */
+static WorkerState *
+apply_bgworker_setup(void)
+{
+ MemoryContext oldcontext;
+ bool launched;
+ WorkerState *wstate;
+
+ elog(LOG, "setting up apply worker #%u", list_length(ApplyWorkersList) + 1);
+
+ oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+ wstate = (WorkerState *) palloc0(sizeof(WorkerState));
+
+ /* Setup shared memory */
+ apply_bgworker_setup_dsm(wstate);
+
+ launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+ MySubscription->oid,
+ MySubscription->name,
+ MyLogicalRepWorker->userid,
+ InvalidOid,
+ dsm_segment_handle(wstate->dsm_seg));
+
+ if (launched)
+ {
+ /* Wait for worker to become ready. */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_ATTACHED);
+
+ ApplyWorkersList = lappend(ApplyWorkersList, wstate);
+ }
+ else
+ {
+ pfree(wstate);
+ wstate = NULL;
+ }
+
+ MemoryContextSwitchTo(oldcontext);
+
+ return wstate;
+}
+
+/*
+ * Send the data to the specified apply bgworker via shared-memory queue.
+ */
+static void
+apply_bgworker_send_data(WorkerState *wstate, Size nbytes, const void *data)
+{
+ shm_mq_result result;
+
+ result = shm_mq_send(wstate->mq_handle, nbytes, data, false, true);
+
+ if (result != SHM_MQ_SUCCESS)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not send tuples to shared-memory queue")));
+}
+
+/*
+ * Wait until the state of apply background worker reach the 'wait_for_state'
+ */
+static void
+apply_bgworker_wait_for(WorkerState *wstate, char wait_for_state)
+{
+ for (;;)
+ {
+ char status;
+
+ /* If the worker is ready, we have succeeded. */
+ SpinLockAcquire(&wstate->pstate->mutex);
+ status = wstate->pstate->state;
+ SpinLockRelease(&wstate->pstate->mutex);
+
+ if (status == wait_for_state)
+ break;
+
+ /* If any workers (or the postmaster) have died, we have failed. */
+ if (status == APPLY_BGWORKER_EXIT)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("Background worker %u failed to apply transaction %u",
+ wstate->pstate->n, wstate->pstate->stream_xid)));
+
+ /* Wait to be signalled. */
+ WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_LOGICAL_APPLY_WORKER_READY);
+
+ /* Reset the latch so we don't spin. */
+ ResetLatch(MyLatch);
+
+ /* An interrupt may have occurred while we were waiting. */
+ CHECK_FOR_INTERRUPTS();
+ }
+}
+
+/*
+ * Check the status of workers and report an error if any apply background
+ * worker has exited unexpectedly.
+ *
+ * Exit if any relation is not in the READY state and if any worker is handling
+ * the streaming transaction at the same time. Because for streaming
+ * transactions that is being applied in apply bgworker, we cannot decide whether to
+ * apply the change for a relation that is not in the READY state (see
+ * should_apply_changes_for_rel) as we won't know remote_final_lsn by that
+ * time.
+ */
+static void
+apply_bgworker_check_status(void)
+{
+ ListCell *lc;
+
+ if (am_apply_bgworker() || MySubscription->stream != SUBSTREAM_APPLY)
+ return;
+
+ foreach(lc, ApplyWorkersList)
+ {
+ WorkerState *wstate = (WorkerState *) lfirst(lc);
+
+ /*
+ * We don't lock here as in the worst case we will just detect the
+ * failure of worker a bit later.
+ */
+ if (wstate->pstate->state == APPLY_BGWORKER_EXIT)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("Background worker %u exited unexpectedly",
+ wstate->pstate->n)));
+ }
+
+ if (list_length(ApplyWorkersIdleList) != list_length(ApplyWorkersList) &&
+ !AllTablesyncsReady())
+ {
+ ereport(LOG,
+ (errmsg("logical replication apply workers for subscription \"%s\" will restart",
+ MySubscription->name),
+ errdetail("Cannot start table synchronization while bgworkers are "
+ "handling streamed replication transaction")));
+
+ proc_exit(0);
+ }
+}
+
+/* Set the state of apply background worker */
+static void
+apply_bgworker_set_state(char state)
+{
+ if (!am_apply_bgworker())
+ return;
+
+ elog(LOG, "[Apply BGW #%u] set state to %c",
+ MyParallelState->n, state);
+
+ SpinLockAcquire(&MyParallelState->mutex);
+ MyParallelState->state = state;
+ SpinLockRelease(&MyParallelState->mutex);
+}
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 87c15b9..24a6fac 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -388,6 +388,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT:
event_name = "HashGrowBucketsReinsert";
break;
+ case WAIT_EVENT_LOGICAL_APPLY_WORKER_READY:
+ event_name = "LogicalApplyWorkerReady";
+ break;
case WAIT_EVENT_LOGICAL_SYNC_DATA:
event_name = "LogicalSyncData";
break;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 786d592..a476885 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4449,7 +4449,7 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 140000)
appendPQExpBufferStr(query, " s.substream,\n");
else
- appendPQExpBufferStr(query, " false AS substream,\n");
+ appendPQExpBufferStr(query, " 'f' AS substream,\n");
if (fout->remoteVersion >= 150000)
appendPQExpBufferStr(query,
@@ -4579,8 +4579,10 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subbinary, "t") == 0)
appendPQExpBufferStr(query, ", binary = true");
- if (strcmp(subinfo->substream, "f") != 0)
+ if (strcmp(subinfo->substream, "o") == 0)
appendPQExpBufferStr(query, ", streaming = on");
+ else if (strcmp(subinfo->substream, "a") == 0)
+ appendPQExpBufferStr(query, ", streaming = apply");
if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
appendPQExpBufferStr(query, ", two_phase = on");
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index d1260f5..b5f270c 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -68,7 +68,7 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subbinary; /* True if the subscription wants the
* publisher to send data in binary */
- bool substream; /* Stream in-progress transactions. */
+ char substream; /* Stream in-progress transactions. */
char subtwophasestate; /* Stream two-phase transactions */
@@ -109,7 +109,7 @@ typedef struct Subscription
bool enabled; /* Indicates if the subscription is enabled */
bool binary; /* Indicates if the subscription wants data in
* binary format */
- bool stream; /* Allow streaming in-progress transactions. */
+ char stream; /* Allow streaming in-progress transactions. */
char twophasestate; /* Allow streaming two-phase transactions */
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
@@ -120,6 +120,18 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
} Subscription;
+/* Disallow streaming in-progress transactions */
+#define SUBSTREAM_OFF 'f'
+
+/*
+ * Streaming transactions are written to a temporary file and applied only
+ * after the transaction is committed on upstream.
+ */
+#define SUBSTREAM_ON 'o'
+
+/* Streaming transactions are appied immediately via a background worker */
+#define SUBSTREAM_APPLY 'a'
+
extern Subscription *GetSubscription(Oid subid, bool missing_ok);
extern void FreeSubscription(Subscription *sub);
extern void DisableSubscription(Oid subid);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index a771ab8..1eb8e12 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -243,8 +243,8 @@ extern TransactionId logicalrep_read_stream_start(StringInfo in,
extern void logicalrep_write_stream_stop(StringInfo out);
extern void logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);
-extern TransactionId logicalrep_read_stream_commit(StringInfo out,
- LogicalRepCommitData *commit_data);
+extern void logicalrep_read_stream_commit(StringInfo out,
+ LogicalRepCommitData *commit_data);
extern void logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
TransactionId subxid);
extern void logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index cd1b6e8..6a1af7f 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -13,6 +13,7 @@
#define LOGICALWORKER_H
extern void ApplyWorkerMain(Datum main_arg);
+extern void ApplyBgworkerMain(Datum main_arg);
extern bool IsLogicalWorker(void);
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 14d5c49..1ed51bc 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -53,7 +53,7 @@ extern XLogRecPtr replorigin_get_progress(RepOriginId node, bool flush);
extern void replorigin_session_advance(XLogRecPtr remote_commit,
XLogRecPtr local_commit);
-extern void replorigin_session_setup(RepOriginId node);
+extern void replorigin_session_setup(RepOriginId node, bool acquire);
extern void replorigin_session_reset(void);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 4485d4e..4d11a13 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -60,6 +60,8 @@ typedef struct LogicalRepWorker
*/
FileSet *stream_fileset;
+ bool subworker;
+
/* Stats. */
XLogRecPtr last_lsn;
TimestampTz last_send_time;
@@ -84,8 +86,9 @@ extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
- Oid userid, Oid relid);
+extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+ Oid userid, Oid relid,
+ dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index b578e2e..d5081d9 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -105,6 +105,7 @@ typedef enum
WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE,
WAIT_EVENT_HASH_GROW_BUCKETS_ELECT,
WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT,
+ WAIT_EVENT_LOGICAL_APPLY_WORKER_READY,
WAIT_EVENT_LOGICAL_SYNC_DATA,
WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE,
WAIT_EVENT_MQ_INTERNAL,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 7fcfad1..e08e4d4 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -197,7 +197,7 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
-- fail - streaming must be boolean
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = foo);
-ERROR: streaming requires a Boolean value
+ERROR: streaming requires a Boolean value or "apply"
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
@@ -205,7 +205,7 @@ WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ..
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | f | off | dbname=regress_doesnotexist | 0/0
+ regress_testsub | regress_subscription_user | f | {testpub} | f | o | d | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
@@ -299,7 +299,7 @@ ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist | 0/0
+ regress_testsub | regress_subscription_user | f | {testpub} | f | o | p | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -311,7 +311,7 @@ WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ..
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist | 0/0
+ regress_testsub | regress_subscription_user | f | {testpub} | f | o | p | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 87ee7bf..9c036d1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2918,11 +2918,13 @@ WordEntryPosVector
WordEntryPosVector1
WorkTableScan
WorkTableScanState
+WorkerEntry
WorkerInfo
WorkerInfoData
WorkerInstrumentation
WorkerJobDumpPtrType
WorkerJobRestorePtrType
+WorkerState
Working_State
WriteBufPtrType
WriteBytePtrType
--
2.7.2.windows.1
view thread (66+ 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]
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OSZPR01MB6310F0FABB05F8E5BB31A5B0FDFC9@OSZPR01MB6310.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