public inbox for [email protected]
help / color / mirror / Atom feedFrom: Peter Smith <[email protected]>
To: [email protected] <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[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, 13 Jul 2022 14:33:18 +1000
Message-ID: <CAHut+PvN7fwtUE=bidzrsOUXSt+JpnkJztZ-Jn5t86moofaZ6g@mail.gmail.com> (raw)
In-Reply-To: <OS3PR01MB627594D75E870BBB45E2E80A9E839@OS3PR01MB6275.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>
Below are my review comments for the v16* patch set:
========
v16-0001
========
1.0 <general>
There are places (comments, docs, errmsgs, etc) in the patch referring
to "parallel mode". I think every one of those references should be
found and renamed to "parallel streaming mode" or "streaming=parallel"
or at the very least match sure that "streaming" is in the same
sentence. IMO it's too vague just saying "parallel" without also
saying the context is for the "streaming" parameter.
I have commented on some of those examples below, but please search
everything anyway (including the docs) to catch the ones I haven't
explicitly mentioned.
======
1.1 src/backend/commands/subscriptioncmds.c
+defGetStreamingMode(DefElem *def)
+{
+ /*
+ * If no value given, assume "true" is meant.
+ */
Please fix this comment to identical to this pushed patch [1]
======
1.2 .../replication/logical/applybgworker.c - apply_bgworker_start
+ if (list_length(ApplyWorkersFreeList) > 0)
+ {
+ wstate = (ApplyBgworkerState *) llast(ApplyWorkersFreeList);
+ ApplyWorkersFreeList = list_delete_last(ApplyWorkersFreeList);
+ Assert(wstate->pstate->status == APPLY_BGWORKER_FINISHED);
+ }
The Assert that the entries in the free-list are FINISHED seems like
unnecessary checking. IIUC, code is already doing the Assert that
entries are FINISHED before allowing them into the free-list in the
first place.
~~~
1.3 .../replication/logical/applybgworker.c - apply_bgworker_find
+ if (found)
+ {
+ char status = entry->wstate->pstate->status;
+
+ /* 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",
+ entry->wstate->pstate->n,
+ entry->wstate->pstate->stream_xid)));
+
+ Assert(status == APPLY_BGWORKER_BUSY);
+
+ return entry->wstate;
+ }
Why not remove that Assert but change the condition to be:
if (status != APPLY_BGWORKER_BUSY)
ereport(...)
======
1.4 src/backend/replication/logical/proto.c - logicalrep_write_stream_abort
@@ -1163,31 +1163,56 @@ logicalrep_read_stream_commit(StringInfo in,
LogicalRepCommitData *commit_data)
/*
* Write STREAM ABORT to the output stream. Note that xid and subxid will be
* same for the top-level transaction abort.
+ *
+ * If write_abort_lsn is true, send the abort_lsn and abort_time fields.
+ * Otherwise not.
*/
"Otherwise not." -> ", otherwise don't."
~~~
1.5 src/backend/replication/logical/proto.c - logicalrep_read_stream_abort
+ *
+ * If read_abort_lsn is true, try to read the abort_lsn and abort_time fields.
+ * Otherwise not.
*/
void
-logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
- TransactionId *subxid)
+logicalrep_read_stream_abort(StringInfo in,
+ LogicalRepStreamAbortData *abort_data,
+ bool read_abort_lsn)
"Otherwise not." -> ", otherwise don't."
======
1.6 src/backend/replication/logical/worker.c - file comment
+ * If streaming = parallel, We assign a new apply background worker (if
+ * available) as soon as the xact's first stream is received. The main apply
"We" -> "we" ... or maybe better just remove it completely.
~~~
1.7 src/backend/replication/logical/worker.c - apply_handle_stream_prepare
+ /*
+ * After sending the data to the apply background worker, wait for
+ * that worker to finish. This is necessary to maintain commit
+ * order which avoids failures due to transaction dependencies and
+ * deadlocks.
+ */
+ apply_bgworker_send_data(wstate, s->len, s->data);
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+ apply_bgworker_free(wstate);
The comment should be changed how you had suggested [2], so that it
will be formatted the same way as a couple of other similar comments.
~~~
1.8 src/backend/replication/logical/worker.c - apply_handle_stream_abort
+ /* Check whether the publisher sends abort_lsn and abort_time. */
+ if (am_apply_bgworker())
+ read_abort_lsn = MyParallelState->server_version >= 160000;
This is handling decisions about read/write of the protocol bytes. I
think feel like it will be better to be checking the server *protocol*
version (not the server postgres version) to make this decision – e.g.
this code should be using the new macro you introduced so it will end
up looking much like how the pgoutput_stream_abort code is doing it.
~~~
1.9 src/backend/replication/logical/worker.c - store_flush_position
@@ -2636,6 +2999,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;
+
SUGGESTION
/* Skip if not the main apply worker */
======
1.10 src/backend/replication/pgoutput/pgoutput.c
@@ -1820,6 +1820,8 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
XLogRecPtr abort_lsn)
{
ReorderBufferTXN *toptxn;
+ bool write_abort_lsn = false;
+ PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
/*
* The abort should happen outside streaming block, even for streamed
@@ -1832,8 +1834,13 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
Assert(rbtxn_is_streamed(toptxn));
+ /* We only send abort_lsn and abort_time if the subscriber needs them. */
+ if (data->protocol_version >= LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM)
+ write_abort_lsn = true;
+
IMO it's simpler to remove the declaration default assignment, and
instead this code can be written as:
write_abort_lsn = data->protocol_version >=
LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM;
======
1.11 src/include/replication/logicalproto.h
+ *
+ * LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM is the minimum protocol version
+ * with support for streaming large transactions in apply background worker.
+ * Introduced in PG16.
"in apply background worker" -> "using apply background workers"
~~~
1.12
+extern void logicalrep_read_stream_abort(StringInfo in,
+ LogicalRepStreamAbortData *abort_data,
+ bool include_abort_lsn);
I think the "include_abort_lsn" is now renamed to "include_abort_lsn".
========
v16-0002
========
No comments.
========
v16-0003
========
3.0 <general>
Same comment about "parallel mode" as in comment #1.0
======
3.1 doc/src/sgml/ref/create_subscription.sgml
+ the publisher-side; 2) there cannot be any non-immutable functions
+ in the subscriber-side replicated table.
The functions are not table data so maybe it's better to say
"functions in the ..." -> "functions used by the ...". If you change
this then there are equivalent comments and commit messages that
should change to match it.
======
3.2 .../replication/logical/applybgworker.c - apply_bgworker_relation_check
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot replicate target relation \"%s.%s\" in parallel "
+ "mode", rel->remoterel.nspname, rel->remoterel.relname),
+ errdetail("The unique column on subscriber is not the unique "
+ "column on publisher or there is at least one "
+ "non-immutable function."),
+ errhint("Please change the streaming option to 'on' instead of
'parallel'.")));
3.2a
SUGGESTED errmsg
"cannot replicate target relation \"%s.%s\" using subscription
parameter streaming=parallel"
3.2b
SUGGESTED errhint
"Please change to use subscription parameter streaming=on"
3.3
The errcode seems the wrong one. Perhaps it should be
ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE.
======
3.4 src/backend/replication/logical/proto.c - logicalrep_write_attrs
In [3] you wrote:
I think the file relcache.c should contain cache-build operations, and the code
I added doesn't have this operation. So I didn't change.
But I only gave relcache.c as an example. It can also be a new static
function in this same file, but anyway I still think this big slab of
code might be better if not done inline in logicalrep_write_attrs.
~~~
3.5 src/backend/replication/logical/proto.c - logicalrep_read_attrs
@@ -1012,11 +1062,14 @@ logicalrep_read_attrs(StringInfo in,
LogicalRepRelation *rel)
{
uint8 flags;
- /* Check for replica identity column */
+ /* Check for replica identity and unique column */
flags = pq_getmsgbyte(in);
- if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
+ if (flags & ATTR_IS_REPLICA_IDENTITY)
attkeys = bms_add_member(attkeys, i);
+ if (flags & ATTR_IS_UNIQUE)
+ attunique = bms_add_member(attunique, i);
The code comment really applies to all 3 statements so maybe better
not to have the blank line here.
======
3.6 src/backend/replication/logical/relation.c - logicalrep_rel_mark_parallel
3.6.a
+ /* Fast path if we marked 'parallel' flag. */
+ if (entry->parallel != PARALLEL_APPLY_UNKNOWN)
+ return;
SUGGESTED
Fast path if 'parallel' flag is already known.
~
3.6.b
+ /* Initialize the flag. */
+ entry->parallel = PARALLEL_APPLY_SAFE;
I think it makes more sense if assigning SAFE is the very *last* thing
this function does – not the first thing.
~
3.6.c
+ /*
+ * First, we check if the unique column in the relation on the
+ * subscriber-side is also the unique column on the publisher-side.
+ */
"First, we check..." -> "First, check..."
~
3.6.d
+ /*
+ * Then, We check if there is any non-immutable function in the local
+ * table. Look for functions in the following places:
"Then, We check..." -> "Then, check"
~~~
3.7 src/backend/replication/logical/relation.c - logicalrep_rel_mark_parallel
view thread (633+ 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]
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <CAHut+PvN7fwtUE=bidzrsOUXSt+JpnkJztZ-Jn5t86moofaZ6g@mail.gmail.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