MIME-Version: 1.0
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>
In-Reply-To: 
 <OS3PR01MB627594D75E870BBB45E2E80A9E839@OS3PR01MB6275.jpnprd01.prod.outlook.com>
From: Peter Smith <smithpb2250@gmail.com>
Date: Wed, 13 Jul 2022 14:33:18 +1000
Message-ID: 
 <CAHut+PvN7fwtUE=bidzrsOUXSt+JpnkJztZ-Jn5t86moofaZ6g@mail.gmail.com>
Subject: Re: Perform streaming logical transactions by background workers and
 parallel apply
To: "wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>,
 Masahiko Sawada <sawada.mshk@gmail.com>,
	"shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>,
 "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>,
	PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Content-Type: text/plain; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable
Archived-At: 
 <https://www.postgresql.org/message-id/CAHut%2BPvN7fwtUE%3DbidzrsOUXSt%2BJpnkJztZ-Jn5t86moofaZ6g%40mail.gmail.com>
Precedence: bulk

Below are my review comments for the v16* patch set:

=3D=3D=3D=3D=3D=3D=3D=3D
v16-0001
=3D=3D=3D=3D=3D=3D=3D=3D

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=3Dparallel"
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.

=3D=3D=3D=3D=3D=3D

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]

=3D=3D=3D=3D=3D=3D

1.2 .../replication/logical/applybgworker.c - apply_bgworker_start

+ if (list_length(ApplyWorkersFreeList) > 0)
+ {
+ wstate =3D (ApplyBgworkerState *) llast(ApplyWorkersFreeList);
+ ApplyWorkersFreeList =3D list_delete_last(ApplyWorkersFreeList);
+ Assert(wstate->pstate->status =3D=3D 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 =3D entry->wstate->pstate->status;
+
+ /* If any workers (or the postmaster) have died, we have failed. */
+ if (status =3D=3D 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 =3D=3D APPLY_BGWORKER_BUSY);
+
+ return entry->wstate;
+ }

Why not remove that Assert but change the condition to be:

if (status !=3D APPLY_BGWORKER_BUSY)
ereport(...)

=3D=3D=3D=3D=3D=3D

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 fie=
lds.
+ * 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."

=3D=3D=3D=3D=3D=3D

1.6 src/backend/replication/logical/worker.c - file comment

+ * If streaming =3D parallel, We assign a new apply background worker (if
+ * available) as soon as the xact's first stream is received. The main app=
ly

"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 =3D MyParallelState->server_version >=3D 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 =E2=80=93 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 */

=3D=3D=3D=3D=3D=3D

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 =3D false;
+ PGOutputData *data =3D (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 >=3D LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_=
NUM)
+ write_abort_lsn =3D true;
+

IMO it's simpler to remove the declaration default assignment, and
instead this code can be written as:

write_abort_lsn =3D data->protocol_version >=3D
LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM;

=3D=3D=3D=3D=3D=3D

1.11 src/include/replication/logicalproto.h

+ *
+ * LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM is the minimum protocol ve=
rsion
+ * with support for streaming large transactions in apply background worke=
r.
+ * 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".


=3D=3D=3D=3D=3D=3D=3D=3D
v16-0002
=3D=3D=3D=3D=3D=3D=3D=3D

No comments.


=3D=3D=3D=3D=3D=3D=3D=3D
v16-0003
=3D=3D=3D=3D=3D=3D=3D=3D

3.0 <general>

Same comment about "parallel mode" as in comment #1.0

=3D=3D=3D=3D=3D=3D

3.1 doc/src/sgml/ref/create_subscription.sgml

+          the publisher-side; 2) there cannot be any non-immutable functio=
ns
+          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.

=3D=3D=3D=3D=3D=3D

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=3Dparallel"

3.2b
SUGGESTED errhint
"Please change to use subscription parameter streaming=3Don"

3.3
The errcode seems the wrong one. Perhaps it should be
ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE.

=3D=3D=3D=3D=3D=3D

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 =3D pq_getmsgbyte(in);
- if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
+ if (flags & ATTR_IS_REPLICA_IDENTITY)
  attkeys =3D bms_add_member(attkeys, i);

+ if (flags & ATTR_IS_UNIQUE)
+ attunique =3D bms_add_member(attunique, i);

The code comment really applies to all 3 statements so maybe better
not to have the blank line here.

=3D=3D=3D=3D=3D=3D

3.6 src/backend/replication/logical/relation.c - logicalrep_rel_mark_parall=
el

3.6.a
+ /* Fast path if we marked 'parallel' flag. */
+ if (entry->parallel !=3D PARALLEL_APPLY_UNKNOWN)
+ return;

SUGGESTED
Fast path if 'parallel' flag is already known.

~

3.6.b
+ /* Initialize the flag. */
+ entry->parallel =3D PARALLEL_APPLY_SAFE;

I think it makes more sense if assigning SAFE is the very *last* thing
this function does =E2=80=93 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_parall=
el