Subject: Re: PATCH: logical_work_mem and logical streaming of large
 in-progress transactions
To: Tomas Vondra <tomas.vondra@2ndquadrant.com>,
 Peter Eisentraut <peter.eisentraut@2ndquadrant.com>,
 Erik Rijkers <er@xs4all.nl>
Cc: Masahiko Sawada <sawada.mshk@gmail.com>,
 PostgreSQL Hackers <pgsql-hackers@postgresql.org>,
 Konstantin Knizhnik <k.knizhnik@postgrespro.ru>
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>
 <40c38758-04b5-74f4-c963-cf300f9e5dff@postgrespro.ru>
 <b95ba070-cd21-72b9-2979-8cdf4d3dd1c6@2ndquadrant.com>
 <126e3a30-a7f5-ede4-2bd0-b217749c8b33@postgrespro.ru>
 <c511125d-33d7-5209-98ce-5c9977f3e437@2ndquadrant.com>
From: Alexey Kondratov <a.kondratov@postgrespro.ru>
Message-ID: <f4cf3883-5ce2-6829-bb42-3038bfc7b4d9@postgrespro.ru>
Date: Wed, 19 Dec 2018 12:58:58 +0300
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:60.0) Gecko/20100101
 Thunderbird/60.2.1
MIME-Version: 1.0
In-Reply-To: <c511125d-33d7-5209-98ce-5c9977f3e437@2ndquadrant.com>
Content-Type: multipart/mixed;
 boundary="------------A2ACCDC71AFF7756709495A5"
Content-Language: en-US
Precedence: bulk

This is a multi-part message in MIME format.
--------------A2ACCDC71AFF7756709495A5
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit

Hi Tomas,

> I'm a bit confused by the changes to TAP tests. Per the patch summary,
> some .pl files get renamed (nor sure why), a new one is added, etc.

I added new tap test case, streaming=true option inside old stream_* 
ones and incremented streaming tests number (+2) because of the 
collision between 009_matviews.pl / 009_stream_simple.pl and 
010_truncate.pl / 010_stream_subxact.pl. At least in the previous 
version of the patch they were under the same numbers. Nothing special, 
but for simplicity, please, find attached my new tap test separately.

>   So
> I've instead enabled streaming subscriptions in all tests, which with
> this patch produces two failures:
>
> Test Summary Report
> -------------------
> t/004_sync.pl                    (Wstat: 7424 Tests: 1 Failed: 0)
>    Non-zero exit status: 29
>    Parse errors: Bad plan.  You planned 7 tests but ran 1.
> t/011_stream_ddl.pl              (Wstat: 256 Tests: 2 Failed: 1)
>    Failed test:  2
>    Non-zero exit status: 1
>
> So yeah, there's more stuff to fix. But I can't directly apply your
> fixes because the updated patches are somewhat different.

Fixes should apply clearly to the previous version of your patch. Also, 
I am not sure, that it is a good idea to simply enable streaming 
subscriptions in all tests (e.g. pre streaming patch t/004_sync.pl), 
since then they do not hit not streaming code.

>>> Interesting. Any idea where does the extra overhead in this particular
>>> case come from? It's hard to deduce that from the single flame graph,
>>> when I don't have anything to compare it with (i.e. the flame graph for
>>> the "normal" case).
>> I guess that bottleneck is in disk operations. You can check
>> logical_repl_worker_new_perf.svg flame graph: disk reads (~9%) and
>> writes (~26%) take around 35% of CPU time in summary. To compare,
>> please, see attached flame graph for the following transaction:
>>
>> INSERT INTO large_text
>> SELECT (SELECT string_agg('x', ',')
>> FROM generate_series(1, 2000)) FROM generate_series(1, 1000000);
>>
>> Execution Time: 44519.816 ms
>> Time: 98333,642 ms (01:38,334)
>>
>> where disk IO is only ~7-8% in total. So we get very roughly the same
>> ~x4-5 performance drop here. JFYI, I am using a machine with SSD for tests.
>>
>> Therefore, probably you may write changes on receiver in bigger chunks,
>> not each change separately.
>>
> Possibly, I/O is certainly a possible culprit, although we should be
> using buffered I/O and there certainly are not any fsyncs here. So I'm
> not sure why would it be cheaper to do the writes in batches.
>
> BTW does this mean you see the overhead on the apply side? Or are you
> running this on a single machine, and it's difficult to decide?

I run this on a single machine, but walsender and worker are utilizing 
almost 100% of CPU per each process all the time, and at apply side I/O 
syscalls take about 1/3 of CPU time. Though I am still not sure, but for 
me this result somehow links performance drop with problems at receiver 
side.

Writing in batches was just a hypothesis and to validate it I have 
performed test with large txn, but consisting of a smaller number of 
wide rows. This test does not exhibit any significant performance drop, 
while it was streamed too. So it seems to be valid. Anyway, I do not 
have other reasonable ideas beside that right now.


Regards

-- 
Alexey Kondratov

Postgres Professional https://www.postgrespro.com
Russian Postgres Company


--------------A2ACCDC71AFF7756709495A5
Content-Type: application/x-perl;
 name="0xx_stream_tough_ddl.pl"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="0xx_stream_tough_ddl.pl"

# 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;

--------------A2ACCDC71AFF7756709495A5--