public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Wed, 24 Aug 2022 15:05:29 +0000
Message-ID: <OS3PR01MB57185B6066FF5C6808CF896294739@OS3PR01MB5718.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OS0PR01MB57166F15F4542FAE69A74EC094739@OS0PR01MB5716.jpnprd01.prod.outlook.com>
References: <OS3PR01MB62758DBE8FA12BA72A43AC819EB89@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+Pu5ah+NhYMfb7KTbkYTxnuGY2j-7vT0G1MZZsS4SKiHew@mail.gmail.com>
<OS3PR01MB62755C6C9A75EB09F7218B589E839@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB627594D75E870BBB45E2E80A9E839@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PvN7fwtUE=bidzrsOUXSt+JpnkJztZ-Jn5t86moofaZ6g@mail.gmail.com>
<OS3PR01MB62758A6AAED27B3A848CEB7A9E8F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB62752B29CEC2592E18236A9A9E909@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAFiTN-uq5Mp+o+_u0iDc8OQV-q7CWf5W=dPgw2ORwxsi6yjVfw@mail.gmail.com>
<CAA4eK1LB1s=B+krpn9mGbZq46pVei0FD7WjhJORq4bD2GU4fng@mail.gmail.com>
<CAFiTN-sK3bVppVAioPfxGjZqY9sHKVM8NooK9Wuex2OCTK25wg@mail.gmail.com>
<OS0PR01MB571651D89C744F84E20AAFE394979@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1K0zkBJQtF1ZNhSvcwh_+pdCaUTf8+bBZR95yP92c0CuQ@mail.gmail.com>
<OS3PR01MB6275D64BE7726B0221B15F389E9F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PutJBXuoy3DB5S7X9sZitfZp4qsXfzgkt5C5=CTNtX-Ug@mail.gmail.com>
<OS3PR01MB6275739E73E8BEC5D13FB6739E6B9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275BA4CBB62CBE8C094E53D9E6A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1+_oHZHoDooAR7QcYD2CeTUWNSwkqVcLWC2iQijAJC4Cg@mail.gmail.com>
<CAA4eK1++B2PFQiin2ZGMs_tszsQ=NJeKuj18-M4PEgckfEYqNA@mail.gmail.com>
<OS0PR01MB57166F15F4542FAE69A74EC094739@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Wednesday, August 24, 2022 9:47 PM [email protected] wrote:
>
> On Friday, August 19, 2022 4:49 PM Amit Kapila <[email protected]>
> >
> > On Thu, Aug 18, 2022 at 5:14 PM Amit Kapila <[email protected]>
> > wrote:
> > >
> > > On Wed, Aug 17, 2022 at 11:58 AM [email protected]
> > > <[email protected]> wrote:
> > > >
> > > > Attach the new patches.
> > > >
> > >
> > > Few comments on v23-0001
> > > =======================
> > >
> >
> > Some more comments on v23-0001
> > ============================
> > 1.
> > static bool
> > handle_streamed_transaction(LogicalRepMsgType action, StringInfo s) { ...
> > - /* not in streaming mode */
> > - if (!in_streamed_transaction)
> > + /* Not in streaming mode and not in apply background worker. */ if
> > + (!(in_streamed_transaction || am_apply_bgworker()))
> > return false;
> >
> > This check appears a bit strange because ideally in bgworker
> > in_streamed_transaction should be false. I think we should set
> > in_streamed_transaction to true in apply_handle_stream_start() only
> > when we are going to write to file. Is there a reason for not doing the same?
>
> No, I removed this.
>
> > 2.
> > + {
> > + /* This is the main apply worker. */ ApplyBgworkerInfo *wstate =
> > + apply_bgworker_find(xid);
> > +
> > + /*
> > + * Check if we are processing this transaction using an apply
> > + * background worker and if so, send the changes to that worker.
> > + */
> > + if (wstate)
> > + {
> > + /* Send STREAM ABORT message to the apply background worker. */
> > + apply_bgworker_send_data(wstate, s->len, s->data);
> >
> > Why at some places the patch needs to separately fetch
> > ApplyBgworkerInfo whereas at other places it directly uses
> > stream_apply_worker to pass the data to bgworker.
> > 3. Why apply_handle_stream_abort() or apply_handle_stream_prepare()
> > doesn't use apply_bgworker_active() to identify whether it needs to
> > send the information to bgworker?
>
> I think stream_apply_worker is only valid between STREAM_START and
> STREAM_END, But it seems it's not clear from the code. So I added some
> comments and slightly refactor the code.
>
>
> > 4. In apply_handle_stream_prepare(), apply_handle_stream_abort(), and
> > some other similar functions, the patch handles three cases (a) apply
> > background worker, (b) sending data to bgworker, (c) handling for
> > streamed transaction in apply worker. I think the code will look
> > better if you move the respective code for all three cases into
> > separate functions. Surely, if the code to deal with each of the cases is less then
> we don't need to move it to a separate function.
>
> Refactored and simplified.
>
> > 5.
> > @@ -1088,24 +1177,78 @@ apply_handle_stream_prepare(StringInfo s) { ...
> > + 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;
> > ...
> >
> > We don't need to in_remote_transaction to false in multiple places.
>
> Removed.
>
> > 6.
> > @@ -1177,36 +1311,93 @@ apply_handle_stream_start(StringInfo s) { ...
> > ...
> > + if (am_apply_bgworker())
> > {
> > - MemoryContext oldctx;
> > -
> > - oldctx = MemoryContextSwitchTo(ApplyContext);
> > + /*
> > + * Make sure the handle apply_dispatch methods are aware we're in a
> > + * remote transaction.
> > + */
> > + in_remote_transaction = true;
> >
> > - MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
> > - FileSetInit(MyLogicalRepWorker->stream_fileset);
> > + /* Begin the transaction. */
> > + AcceptInvalidationMessages();
> > + maybe_reread_subscription();
> >
> > - MemoryContextSwitchTo(oldctx);
> > + StartTransactionCommand();
> > + BeginTransactionBlock();
> > + CommitTransactionCommand();
> > }
> > ...
> >
> > Why do we need to start a transaction here? Why can't it be done via
> > begin_replication_step() during the first operation apply? Is it
> > because we may need to define a save point in bgworker and we don't
> > that information beforehand? If so, then also, can't it be handled by
> > begin_replication_step() either by explicitly passing the information
> > or checking it there and then starting a transaction block? In any
> > case, please add a few comments to explain why this separate handling
> > is required for bgworker?
>
> The transaction block is used to define the savepoint and I moved these codes to
> the place where the savepoint is defined which looks better now.
>
> > 7. When we are already setting bgworker status as APPLY_BGWORKER_BUSY
> > in
> > apply_bgworker_setup_dsm() then why do we need to set it again in
> > apply_bgworker_start()?
>
> Removed.
>
> > 8. It is not clear to me how APPLY_BGWORKER_EXIT status is used. Is it
> > required for the cases where bgworker exists due to some error and
> > then apply worker uses it to detect that and exits? How other
> > bgworkers would notice this, is it done via apply_bgworker_check_status()?
>
> It was used to detect the unexpected exit of bgworker and I have changed the
> design of this which is now similar to what we have in parallel query.
>
> Attach the new version patch set(v24) which address above comments.
> Besides, I added some logic which try to stop the bgworker at transaction end if
> there are enough workers in the pool.
Also attach the result of performance test based on v23 patch.
This test used synchronous logical replication, and compared SQL execution
times before and after applying the patch. This is tested by varying
logical_decoding_work_mem.
The test was performed ten times, and the average of the middle eight was taken.
The results are as follows. The bar chart and the details of the test are attached.
RESULT - bulk insert (5kk)
----------------------------------
logical_decoding_work_mem 64kB 128kB 256kB 512kB 1MB 2MB 4MB 8MB 16MB 32MB 64MB
HEAD 46.940 46.428 46.663 46.373 46.339 46.838 50.346 50.536 50.452 50.582 47.491
patched 33.942 33.780 30.760 30.760 29.992 30.076 30.827 33.420 33.966 34.133 31.096
For different logical_decoding_work_mem size, it takes
about 30% ~ 40% less time, which looks good.
Some other tests are still in progress, might share them later.
Best regards,
Hou zj
Some parameters specified in postgresql.conf
=====================================================
shared_buffers = 8GB
checkpoint_timeout = 30min
max_wal_size = 20GB
min_wal_size = 10GB
autovacuum = off
Steps
=====================================================
-- create table in publisher and subscriber
CREATE TABLE large_test (
id int primary key,
num1 bigint,
num2 double precision,
num3 double precision
);
-- create publication
create publication pub for table large_test;
-- create subscription
-- HEAD:
CREATE SUBSCRIPTION sub CONNECTION 'port=5432 dbname=postgres' PUBLICATION pub with(streaming=on);
-- after applying patch:
CREATE SUBSCRIPTION sub CONNECTION 'port=5432 dbname=postgres' PUBLICATION pub with(streaming=parallel);
Then set synchronous replication and restart publisher server.
-- do bulk insert in publisher
\timing
EXPLAIN (ANALYZE, BUFFERS) INSERT INTO large_test (id, num1, num2, num3)
SELECT i, round(random()*10), random(), random()*142
FROM generate_series(1, 5000000) s(i);
Attachments:
[text/plain] details.txt (1.0K, ../OS3PR01MB57185B6066FF5C6808CF896294739@OS3PR01MB5718.jpnprd01.prod.outlook.com/2-details.txt)
download | inline:
Some parameters specified in postgresql.conf
=====================================================
shared_buffers = 8GB
checkpoint_timeout = 30min
max_wal_size = 20GB
min_wal_size = 10GB
autovacuum = off
Steps
=====================================================
-- create table in publisher and subscriber
CREATE TABLE large_test (
id int primary key,
num1 bigint,
num2 double precision,
num3 double precision
);
-- create publication
create publication pub for table large_test;
-- create subscription
-- HEAD:
CREATE SUBSCRIPTION sub CONNECTION 'port=5432 dbname=postgres' PUBLICATION pub with(streaming=on);
-- after applying patch:
CREATE SUBSCRIPTION sub CONNECTION 'port=5432 dbname=postgres' PUBLICATION pub with(streaming=parallel);
Then set synchronous replication and restart publisher server.
-- do bulk insert in publisher
\timing
EXPLAIN (ANALYZE, BUFFERS) INSERT INTO large_test (id, num1, num2, num3)
SELECT i, round(random()*10), random(), random()*142
FROM generate_series(1, 5000000) s(i);
[image/png] perfmance_test_bulk_insert.png (93.6K, ../OS3PR01MB57185B6066FF5C6808CF896294739@OS3PR01MB5718.jpnprd01.prod.outlook.com/3-perfmance_test_bulk_insert.png)
download | view image
view thread (625+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OS3PR01MB57185B6066FF5C6808CF896294739@OS3PR01MB5718.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