From: Alexey Kondratov <a.kondratov@postgrespro.ru>
To: Tomas Vondra <tomas.vondra@2ndquadrant.com>
To: Peter Eisentraut <peter.eisentraut@2ndquadrant.com>
To: Erik Rijkers <er@xs4all.nl>
Cc: Masahiko Sawada <sawada.mshk@gmail.com>
Cc: PostgreSQL Hackers <pgsql-hackers@postgresql.org>
Cc: Konstantin Knizhnik <k.knizhnik@postgrespro.ru>
Subject: Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions
Date: Mon, 17 Dec 2018 19:23:44 +0300
Message-ID: <40c38758-04b5-74f4-c963-cf300f9e5dff@postgrespro.ru> (raw)
In-Reply-To: <8940f374-6651-26d0-cefe-6c719454b6fe@2ndquadrant.com>
References: <688b0b7f-2f6c-d827-c27b-216a8e3ea700@2ndquadrant.com>
<bbe1333f-17ba-1156-0713-7cc0bfe2d842@2ndquadrant.com>
<bc83b847-2fc7-6561-4ef1-a3c9d154cc31@2ndquadrant.com>
<47223121-222d-8e04-4928-a28ecdc5327d@2ndquadrant.com>
<CAD21AoBa2mhbhZW6hnDRKs1ggycJOBE69eHk3KgCJ5Cwj9-ZkA@mail.gmail.com>
<b9b306a3-c0a2-7eb9-92eb-a3ca111bd1bc@2ndquadrant.com>
<4762f6bd-b147-cd06-f286-22f54790b77d@2ndquadrant.com>
<cb2530b102afad854e051cca44577c6a@xs4all.nl>
<d807dc44-55f9-5c23-97e5-a5881f4135e2@2ndquadrant.com>
<5eddec05-404f-3fc4-bafd-15e9e4787705@2ndquadrant.com>
<e84013b6-b890-1c4c-2b82-eca185a2197c@2ndquadrant.com>
<8940f374-6651-26d0-cefe-6c719454b6fe@2ndquadrant.com>
Hi Tomas,
> This new version is mostly just a rebase to current master (or almost,
> because 2pc decoding only applies to 29180e5d78 due to minor bitrot),
> but it also addresses the new stuff committed since last version (most
> importantly decoding of TRUNCATE). It also fixes a bug in WAL-logging of
> subxact assignments, where the assignment was included in records with
> XID=0, essentially failing to track the subxact properly.
I started reviewing your patch about a month ago and tried to do an
in-depth review, since I am very interested in this patch too. The new
version is not applicable to master 29180e5d78, but everything is OK
after applying 2pc patch before. Anyway, I guess it may complicate
further testing and review, since any potential reviewer has to take
into account both patches at once. Previous version was applicable to
master and was working fine for me separately (excepting a few
patch-specific issues, which I try to explain below).
Patch review
========
First of all, I want to say thank you for such a huge work done. Here
are some problems, which I have found and hopefully fixed with my
additional patch (please, find attached, it should be applicable to the
last commit of your newest patch version):
1) The most important issue is that your tap tests were broken—there was
missing option "WITH (streaming=true)" in the subscription creating
statement. Therefore, spilling mechanism has been tested rather than
streaming.
2) After fixing tests the first one with simple streaming is immediately
failed, because of logical replication worker segmentation fault. It
happens, since worker tries to call stream_cleanup_files inside
stream_open_file at the stream start, while nxids is zero, then it goes
to the negative value and everything crashes. Something similar may
happen with xids array, so I added two checks there.
3) The next problem is much more critical and is dedicated to historic
MVCC visibility rules. Previously, walsender was starting to decode
transaction on commit and we were able to resolve all xmin, xmax,
combocids to cmin/cmax, build tuplecids hash and so on, but now we start
doing all these things on the fly.
Thus, rather difficult situation arises: HeapTupleSatisfiesHistoricMVCC
is trying to validate catalog tuples, which are currently in the future
relatively to the current decoder position inside transaction, e.g. we
may want to resolve cmin/cmax of a tuple, which was created with cid 3
and deleted with cid 5, while we are currently at cid 4, so our
tuplecids hash is not complete to handle such a case.
I have updated HeapTupleSatisfiesHistoricMVCC visibility rules with two
options:
/*
* If we accidentally see a tuple from our transaction, but cannot
resolve its
* cmin, so probably it is from the future, thus drop it.
*/
if (!resolved)
return false;
and
/*
* If we accidentally see a tuple from our transaction, but cannot
resolve its
* cmax or cmax == InvalidCommandId, so probably it is still valid,
thus accept it.
*/
if (!resolved || cmax == InvalidCommandId)
return true;
4) There was a problem with marking top-level transaction as having
catalog changes if one of its subtransactions has. It was causing a
problem with DDL statements just after subtransaction start (savepoint),
so data from new columns is not replicated.
5) Similar issue with schema send. You send schema only once per each
sub/transaction (IIRC), while we have to update schema on each catalog
change: invalidation execution, snapshot rebuild, adding new tuple cids.
So I ended up with adding is_schema_send flag to ReorderBufferTXN, since
it is easy to set it inside RB and read in the output plugin. Probably,
we have to choose a better place for this flag.
6) To better handle all these tricky cases I added new tap
test—014_stream_tough_ddl.pl—which consist of really tough combination
of DDL, DML, savepoints and ROLLBACK/RELEASE in a single transaction.
I marked all my fixes and every questionable place with comment and
"TOCHECK:" label for easy search. Removing of pretty any of these fixes
leads to the tests fail due to the segmentation fault or replication
mismatch. Though I mostly read and tested old version of patch, but
after a quick look it seems that all these fixes are applicable to the
new version of patch as well.
Performance
========
I have also performed a series of performance tests, and found that
patch adds a huge overhead in the case of a large transaction consisting
of many small rows, e.g.:
CREATE TABLE large_test (num1 bigint, num2 double precision, num3 double
precision);
EXPLAIN (ANALYZE, BUFFERS) INSERT INTO large_test (num1, num2, num3)
SELECT round(random()*10), random(), random()*142
FROM generate_series(1, 1000000) s(i);
Execution Time: 2407.709 ms
Total Time: 11494,238 ms (00:11,494)
With synchronous_standby_names and 64 MB logical_work_mem it takes up to
x5 longer, while without patch it is about x2. Thus, logical replication
streaming is approximately x4 as slower for similar transactions.
However, dealing with large transactions consisting of a small number of
large rows is much better:
CREATE TABLE large_text (t TEXT);
EXPLAIN (ANALYZE, BUFFERS) INSERT INTO large_text
SELECT (SELECT string_agg('x', ',')
FROM generate_series(1, 1000000)) FROM generate_series(1, 125);
Execution Time: 3545.642 ms
Total Time: 7678,617 ms (00:07,679)
It is around the same x2 as without patch. If someone is interested I
also added flamegraphs of walsender and logical replication worker
during first numerical transaction processing.
Regards
--
Alexey Kondratov
Postgres Professional https://www.postgrespro.com
Russian Postgres Company
Attachments:
logical_repl_worker_new_perf.svg.zip (application/zip, 56.0K) download
walsender_new_perf.svg.zip (application/zip, 34.7K) download0009-Fix-worker-historic-MVCC-visibility-rules-subxacts-s.patch (text/x-patch, 20.4K) download
From 0a7cbaeaa179bde9dd5146e60f7b5c52b6c34899 Mon Sep 17 00:00:00 2001
From: Alexey Kondratov <alex.lumir@gmail.com>
Date: Mon, 17 Dec 2018 15:43:13 +0300
Subject: [PATCH] Fix worker, historic MVCC visibility rules, subxacts, schema
send, tests
---
doc/src/sgml/logicaldecoding.sgml | 2 +-
.../replication/logical/reorderbuffer.c | 61 +++++++++---
src/backend/replication/logical/worker.c | 25 ++---
src/backend/replication/pgoutput/pgoutput.c | 20 +++-
src/backend/replication/walsender.c | 2 +-
src/backend/utils/time/tqual.c | 16 ++-
src/include/replication/reorderbuffer.h | 5 +
..._stream_simple.pl => 011_stream_simple.pl} | 2 +-
...tream_subxact.pl => 012_stream_subxact.pl} | 2 +-
.../{011_stream_ddl.pl => 013_stream_ddl.pl} | 2 +-
.../subscription/t/014_stream_tough_ddl.pl | 98 +++++++++++++++++++
...t_abort.pl => 015_stream_subxact_abort.pl} | 2 +-
...ort.pl => 016_stream_subxact_ddl_abort.pl} | 2 +-
13 files changed, 200 insertions(+), 39 deletions(-)
rename src/test/subscription/t/{009_stream_simple.pl => 011_stream_simple.pl} (98%)
rename src/test/subscription/t/{010_stream_subxact.pl => 012_stream_subxact.pl} (98%)
rename src/test/subscription/t/{011_stream_ddl.pl => 013_stream_ddl.pl} (98%)
create mode 100644 src/test/subscription/t/014_stream_tough_ddl.pl
rename src/test/subscription/t/{012_stream_subxact_abort.pl => 015_stream_subxact_abort.pl} (97%)
rename src/test/subscription/t/{013_stream_subxact_ddl_abort.pl => 016_stream_subxact_ddl_abort.pl} (97%)
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgmlindex 3571f96a8d..dbec2e4ef7 100644--- a/doc/src/sgml/logicaldecoding.sgml+++ b/doc/src/sgml/logicaldecoding.sgml@@ -1057,7 +1057,7 @@ stream_commit_cb(...); <-- commit of the streamed transaction
</para>
<para>
- Similarly to spill-to-disk behavior, sStreaming is triggered when the total+ Similarly to spill-to-disk behavior, streaming is triggered when the total
amount of changes decoded from the WAL (for all in-progress transactions)
exceeds limit defined by <varname>logical_work_mem</varname> setting. At
that point the largest toplevel transaction (measured by amount of memory
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.cindex 1c394296ac..2200332999 100644--- a/src/backend/replication/logical/reorderbuffer.c+++ b/src/backend/replication/logical/reorderbuffer.c@@ -1744,9 +1744,10 @@ ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
}
else
{
+ // TOCHECK: Is the second assert actually necessary?
Assert(ent->cmin == change->data.tuplecid.cmin);
- Assert(ent->cmax == InvalidCommandId ||- ent->cmax == change->data.tuplecid.cmax);+ // Assert(ent->cmax == InvalidCommandId ||+ // ent->cmax == change->data.tuplecid.cmax);
/*
* if the tuple got valid in this transaction and now got deleted
@@ -2881,6 +2882,9 @@ ReorderBufferExecuteInvalidations(ReorderBuffer *rb, ReorderBufferTXN *txn)
for (i = 0; i < txn->ninvalidations; i++)
LocalExecuteInvalidationMessage(&txn->invalidations[i]);
++ /* Invalidate current schema as well */+ txn->is_schema_sent = false;
}
/*
@@ -2895,6 +2899,23 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
++ /*+ * We read catalog changes from WAL, which are not yet sent, so+ * invalidate current schema in order output plugin can resend+ * schema again.+ */+ txn->is_schema_sent = false;++ /*+ * TOCHECK: Mark toplevel transaction as having catalog changes too+ * if one of its children has.+ */+ if (txn->toptxn != NULL)+ {+ txn->toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;+ txn->toptxn->is_schema_sent = false;+ }
}
/*
@@ -3476,7 +3497,15 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
* using snapshot half-way through the subxact.
*/
command_id = txn->command_id;
- snapshot_now = txn->snapshot_now;++ /*+ * TOCHECK: We have to rebuild historic snapshot to be sure it includes all+ * information about subtransactions, which could arrive after streaming start.+ */+ if (!txn->is_schema_sent)+ snapshot_now = ReorderBufferCopySnap(rb, txn->base_snapshot,+ txn, command_id);+ // snapshot_now = txn->snapshot_now;
}
/*
@@ -3522,7 +3551,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
/*
* Enforce correct ordering of changes, merged from multiple
* subtransactions. The changes may have the same LSN due to
- * MULTI_INSERT xllog records.+ * MULTI_INSERT xlog records.
*/
if (prev_lsn != InvalidXLogRecPtr)
Assert(prev_lsn <= change->lsn);
@@ -3731,6 +3760,11 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
snapshot_now = change->data.snapshot;
}
+ /*+ * TOCHECK: Snapshot changed, then invalidate current schema to reflect+ * possible catalog changes.+ */+ txn->is_schema_sent = false;
/* and continue with the new one */
SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash,
@@ -3868,7 +3902,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
rb->streamTxns += (rbtxn_is_streamed(txn)) ? 1 : 0;
rb->streamBytes += txn->size;
- elog(WARNING, "updating stream stats %p %ld %ld %ld",+ elog(INFO, "updating stream stats %p %ld %ld %ld",
rb, rb->streamCount, rb->streamTxns, txn->size);
/*
@@ -4950,7 +4984,7 @@ ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data,
CommandId *cmin, CommandId *cmax)
{
ReorderBufferTupleCidKey key;
- ReorderBufferTupleCidEnt *ent;+ ReorderBufferTupleCidEnt *ent = NULL;
ForkNumber forkno;
BlockNumber blockno;
bool updated_mapping = false;
@@ -4974,11 +5008,16 @@ ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data,
&key.tid);
restart:
- ent = (ReorderBufferTupleCidEnt *)- hash_search(tuplecid_data,- (void *) &key,- HASH_FIND,- NULL);+ /*+ * TOCHECK: If tuplecid_data is NULL, then we are not able to resolve cmin/cmax,+ * so try to update mappings and return false.+ */+ if (tuplecid_data != NULL)+ ent = (ReorderBufferTupleCidEnt *)+ hash_search(tuplecid_data,+ (void *) &key,+ HASH_FIND,+ NULL);
/*
* failed to find a mapping, check whether the table was rewritten and
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.cindex eaefba2049..adf69a5a38 100644--- a/src/backend/replication/logical/worker.c+++ b/src/backend/replication/logical/worker.c@@ -2444,10 +2444,21 @@ stream_open_file(Oid subid, TransactionId xid, bool first_segment)
{
MemoryContext oldcxt;
- stream_cleanup_files(subid, xid);+ /*+ * TOCHECK: If nxids=0, then we have nothing to clean up.+ */+ if (nxids > 0)+ stream_cleanup_files(subid, xid);
oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+ /* TOCHECK: Initialize xids array if it is the first run. */+ if (xids == NULL)+ {+ maxnxids = 64;+ xids = palloc(maxnxids * sizeof(TransactionId));+ }+
/*
* We need to remember the XIDs we spilled to files, so that we can
* remove them at worker exit (e.g. after DROP SUBSCRIPTION).
@@ -2462,16 +2473,8 @@ stream_open_file(Oid subid, TransactionId xid, bool first_segment)
*/
if (nxids == maxnxids) /* array of XIDs is full */
{
- if (!xids)- {- maxnxids = 64;- xids = palloc(maxnxids * sizeof(TransactionId));- }- else- {- maxnxids = 2 * maxnxids;- xids = repalloc(xids, maxnxids * sizeof(TransactionId));- }+ maxnxids = 2 * maxnxids;+ xids = repalloc(xids, maxnxids * sizeof(TransactionId));
}
xids[nxids++] = xid;
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.cindex 04432ffb57..d5a3cf6308 100644--- a/src/backend/replication/pgoutput/pgoutput.c+++ b/src/backend/replication/pgoutput/pgoutput.c@@ -370,7 +370,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
static void
maybe_send_schema(LogicalDecodingContext *ctx,
TransactionId topxid, TransactionId xid,
- Relation relation, RelationSyncEntry *relentry)+ Relation relation, RelationSyncEntry *relentry, ReorderBufferTXN *txn)
{
bool schema_sent = relentry->schema_sent;
@@ -381,7 +381,15 @@ maybe_send_schema(LogicalDecodingContext *ctx,
* that we don't know at this point.
*/
if (in_streaming)
- schema_sent = get_schema_sent_in_streamed_txn(relentry, topxid);+ {+ /*+ * TOCHECK: We have to send schema after each catalog change and it may + * occur when streaming already started, so we have to track new catalog + * changes somehow.+ */+ schema_sent = txn->is_schema_sent;+ // schema_sent = get_schema_sent_in_streamed_txn(relentry, topxid);+ }
if (!schema_sent)
{
@@ -415,7 +423,9 @@ maybe_send_schema(LogicalDecodingContext *ctx,
relentry->xid = xid;
if (in_streaming)
- set_schema_sent_in_streamed_txn(relentry, topxid);+ /* TOCHECK: Maybe change flag location? */+ txn->is_schema_sent = true;+ // set_schema_sent_in_streamed_txn(relentry, topxid);
else
relentry->schema_sent = true;
}
@@ -479,7 +489,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
- maybe_send_schema(ctx, topxid, xid, relation, relentry);+ maybe_send_schema(ctx, topxid, xid, relation, relentry, txn);
/* Send the data */
switch (change->action)
@@ -569,7 +579,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
continue;
relids[nrelids++] = relid;
- maybe_send_schema(ctx, topxid, xid, relation, relentry);+ maybe_send_schema(ctx, topxid, xid, relation, relentry, txn);
}
if (nrelids > 0)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.cindex be92f3e5f5..de709250c0 100644--- a/src/backend/replication/walsender.c+++ b/src/backend/replication/walsender.c@@ -3621,7 +3621,7 @@ UpdateSpillStats(LogicalDecodingContext *ctx)
MyWalSnd->streamCount = rb->streamCount;
MyWalSnd->streamBytes = rb->streamBytes;
- elog(WARNING, "UpdateSpillStats: updating stats %p %ld %ld %ld %ld %ld %ld",+ elog(INFO, "UpdateSpillStats: updating stats %p %ld %ld %ld %ld %ld %ld",
rb, rb->spillTxns, rb->spillCount, rb->spillBytes,
rb->streamTxns, rb->streamCount, rb->streamBytes);
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.cindex f7c4c9188c..b905077164 100644--- a/src/backend/utils/time/tqual.c+++ b/src/backend/utils/time/tqual.c@@ -1692,8 +1692,12 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
htup, buffer,
&cmin, &cmax);
+ /*+ * TOCHECK: If we accidentally see a tuple from our transaction, but cannot resolve its+ * cmin, so probably it is from the future, thus drop it.+ */
if (!resolved)
- elog(ERROR, "could not resolve cmin/cmax of catalog tuple");+ return false;
Assert(cmin != InvalidCommandId);
@@ -1763,10 +1767,12 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
htup, buffer,
&cmin, &cmax);
- if (!resolved)- elog(ERROR, "could not resolve combocid to cmax");-- Assert(cmax != InvalidCommandId);+ /*+ * TOCHECK: If we accidentally see a tuple from our transaction, but cannot resolve its+ * cmax or cmax == InvalidCommandId, so probably it is still valid, thus accept it.+ */+ if (!resolved || cmax == InvalidCommandId)+ return true;
if (cmax >= snapshot->curcid)
return true; /* deleted after scan started */
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.hindex bf5766c1e4..29dba6673c 100644--- a/src/include/replication/reorderbuffer.h+++ b/src/include/replication/reorderbuffer.h@@ -220,6 +220,11 @@ typedef struct ReorderBufferTXN
/* In case of 2PC we need to pass GID to output plugin */
char *gid;
+ /*+ * Do we need to send schema for this transaction in output plugin?+ */+ bool is_schema_sent;+
/*
* Toplevel transaction for this subxact (NULL for top-level).
*/
diff --git a/src/test/subscription/t/009_stream_simple.pl b/src/test/subscription/t/011_stream_simple.plsimilarity index 98%rename from src/test/subscription/t/009_stream_simple.plrename to src/test/subscription/t/011_stream_simple.plindex 4d01f7e5ec..f0aae1041a 100644--- a/src/test/subscription/t/009_stream_simple.pl+++ b/src/test/subscription/t/011_stream_simple.pl@@ -40,7 +40,7 @@ $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tes
my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
-"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"
);
wait_for_caught_up($node_publisher, $appname);
diff --git a/src/test/subscription/t/010_stream_subxact.pl b/src/test/subscription/t/012_stream_subxact.plsimilarity index 98%rename from src/test/subscription/t/010_stream_subxact.plrename to src/test/subscription/t/012_stream_subxact.plindex 1a8b8ffe9e..00dd60b91f 100644--- a/src/test/subscription/t/010_stream_subxact.pl+++ b/src/test/subscription/t/012_stream_subxact.pl@@ -40,7 +40,7 @@ $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tes
my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
-"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"
);
wait_for_caught_up($node_publisher, $appname);
diff --git a/src/test/subscription/t/011_stream_ddl.pl b/src/test/subscription/t/013_stream_ddl.plsimilarity index 98%rename from src/test/subscription/t/011_stream_ddl.plrename to src/test/subscription/t/013_stream_ddl.plindex 04af0900ac..ecaf4383b1 100644--- a/src/test/subscription/t/011_stream_ddl.pl+++ b/src/test/subscription/t/013_stream_ddl.pl@@ -40,7 +40,7 @@ $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tes
my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
-"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"
);
wait_for_caught_up($node_publisher, $appname);
diff --git a/src/test/subscription/t/014_stream_tough_ddl.pl b/src/test/subscription/t/014_stream_tough_ddl.plnew file mode 100644index 0000000000..02969c7260--- /dev/null+++ b/src/test/subscription/t/014_stream_tough_ddl.pl@@ -0,0 +1,98 @@+# Test streaming of large transaction with DDL, subtransactions and rollbacks.+use strict;+use warnings;+use PostgresNode;+use TestLib;+use Test::More tests => 2;++sub wait_for_caught_up+{+ my ($node, $appname) = @_;++ $node->poll_query_until('postgres',+"SELECT pg_current_wal_lsn() <= replay_lsn FROM pg_stat_replication WHERE application_name = '$appname';"+ ) or die "Timed out while waiting for subscriber to catch up";+}++# Create publisher node+my $node_publisher = get_new_node('publisher');+$node_publisher->init(allows_streaming => 'logical');+$node_publisher->append_conf('postgresql.conf', 'logical_work_mem = 64kB');+$node_publisher->start;++# Create subscriber node+my $node_subscriber = get_new_node('subscriber');+$node_subscriber->init(allows_streaming => 'logical');+$node_subscriber->start;++# Create some preexisting content on publisher+$node_publisher->safe_psql('postgres',+ "CREATE TABLE test_tab (a int primary key, b varchar)");+$node_publisher->safe_psql('postgres',+ "INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar')");++# Setup structure on subscriber+$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab (a int primary key, b text, c INT, d text, e INT)");++# Setup logical replication+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';+$node_subscriber->safe_psql('postgres',+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"+);++wait_for_caught_up($node_publisher, $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) FROM test_tab");+is($result, qq(2|0|0), 'check initial data was copied to subscriber');+++# large (streamed) transaction with DDL and DML+$node_publisher->safe_psql('postgres', q{+BEGIN;+SAVEPOINT s1;+INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 1000) s(i);+SAVEPOINT s2;+ALTER TABLE test_tab ADD COLUMN c INT;+INSERT INTO test_tab SELECT i, md5(i::text), i FROM generate_series(1001, 2000) s(i);+SAVEPOINT s3;+ALTER TABLE test_tab ADD COLUMN d text;+SAVEPOINT s4;+SAVEPOINT s5;+INSERT INTO test_tab SELECT i, md5(i::text), i, md5(i::text) FROM generate_series(2001, 3000) s(i);+ALTER TABLE test_tab ADD COLUMN e INT;+INSERT INTO test_tab SELECT i, md5(i::text), i, md5(i::text), i FROM generate_series(3001, 4000) s(i);+SAVEPOINT s10;+ALTER TABLE test_tab DROP d;+INSERT INTO test_tab SELECT i, md5(i::text), i, i FROM generate_series(4001, 5000) s(i);+ALTER TABLE test_tab ADD COLUMN d text;+ROLLBACK TO SAVEPOINT s10;+RELEASE SAVEPOINT s10;+SAVEPOINT s10;+INSERT INTO test_tab SELECT i, md5(i::text), i, md5(i::text), i FROM generate_series(5001, 6000) s(i);+SAVEPOINT s6;+ALTER TABLE test_tab DROP d;+INSERT INTO test_tab SELECT i, md5(i::text), i, i FROM generate_series(6001, 7000) s(i);+SAVEPOINT s7;+ALTER TABLE test_tab ADD COLUMN d text;+INSERT INTO test_tab (a, b, c, d, e) SELECT i, md5(i::text), i, md5(i::text), i FROM generate_series(7001, 8000) s(i);+COMMIT;+});++wait_for_caught_up($node_publisher, $appname);++$result =+ $node_subscriber->safe_psql('postgres', "SELECT count(*), count(a), count(b), count(c), count(d), count(e) FROM test_tab");+is($result, qq(7000|7000|7000|6000|4000|4000), 'check extra columns contain local defaults');++$node_subscriber->stop;+$node_publisher->stop;diff --git a/src/test/subscription/t/012_stream_subxact_abort.pl b/src/test/subscription/t/015_stream_subxact_abort.plsimilarity index 97%rename from src/test/subscription/t/012_stream_subxact_abort.plrename to src/test/subscription/t/015_stream_subxact_abort.plindex 6fecfe6fe7..dcec081fa1 100644--- a/src/test/subscription/t/012_stream_subxact_abort.pl+++ b/src/test/subscription/t/015_stream_subxact_abort.pl@@ -40,7 +40,7 @@ $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tes
my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
-"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"
);
wait_for_caught_up($node_publisher, $appname);
diff --git a/src/test/subscription/t/013_stream_subxact_ddl_abort.pl b/src/test/subscription/t/016_stream_subxact_ddl_abort.plsimilarity index 97%rename from src/test/subscription/t/013_stream_subxact_ddl_abort.plrename to src/test/subscription/t/016_stream_subxact_ddl_abort.plindex 50990c170c..41ad2b668c 100644--- a/src/test/subscription/t/013_stream_subxact_ddl_abort.pl+++ b/src/test/subscription/t/016_stream_subxact_ddl_abort.pl@@ -40,7 +40,7 @@ $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tes
my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
-"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"+"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming=true)"
);
wait_for_caught_up($node_publisher, $appname);
--
2.17.1
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: pgsql-hackers@postgresql.org
Cc: a.kondratov@postgrespro.ru, tomas.vondra@2ndquadrant.com, peter.eisentraut@2ndquadrant.com, er@xs4all.nl, sawada.mshk@gmail.com, k.knizhnik@postgrespro.ru
Subject: Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions
In-Reply-To: <40c38758-04b5-74f4-c963-cf300f9e5dff@postgrespro.ru>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox