MIME-Version: 1.0
References: 
 <OS0PR01MB5716730C06159452335D870D947F9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAA4eK1Lu-6oXMk7ZaGYLwm3CRLBuzueGbasyHnNpJxu6Mq3mmg@mail.gmail.com>
 <OS3PR01MB6275F145878B4A44586C46CE9E499@OS3PR01MB6275.jpnprd01.prod.outlook.com>
In-Reply-To: 
 <OS3PR01MB6275F145878B4A44586C46CE9E499@OS3PR01MB6275.jpnprd01.prod.outlook.com>
From: Amit Kapila <amit.kapila16@gmail.com>
Date: Thu, 15 Sep 2022 17:09:45 +0530
Message-ID: 
 <CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g@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: "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>,
 Peter Smith <smithpb2250@gmail.com>,
	Dilip Kumar <dilipbalaut@gmail.com>, Masahiko Sawada <sawada.mshk@gmail.com>,
	"shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>,
	PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Content-Type: multipart/mixed; boundary="00000000000074d22a05e8b5b45f"
Archived-At: 
 <https://www.postgresql.org/message-id/CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g%40mail.gmail.com>
Precedence: bulk

--00000000000074d22a05e8b5b45f
Content-Type: text/plain; charset="UTF-8"

On Thu, Sep 15, 2022 at 10:45 AM wangw.fnst@fujitsu.com
<wangw.fnst@fujitsu.com> wrote:
>
> Attach the new patch set.
>

Review of v29-0001*
==================
1.
+parallel_apply_find_worker(TransactionId xid)
{
...
+ entry = hash_search(ParallelApplyWorkersHash, &xid, HASH_FIND, &found);
+ if (found)
+ {
+ /* If any workers (or the postmaster) have died, we have failed. */
+ if (entry->winfo->error_mq_handle == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("lost connection to parallel apply worker")));
...
}

I think the above comment is incorrect because if the postmaster would
have died then you wouldn't have found the entry in the hash table.
How about something like: "We can't proceed if the parallel streaming
worker has already exited."

2.
+/*
+ * Find the previously assigned worker for the given transaction, if any.
+ */
+ParallelApplyWorkerInfo *
+parallel_apply_find_worker(TransactionId xid)

No need to use word 'previously' in the above sentence.

3.
+ * We need one key to register the location of the header, and we need
+ * another key to track the location of the message queue.
+ */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(ParallelApplyWorkerShared));
+ shm_toc_estimate_chunk(&e, queue_size);
+ shm_toc_estimate_chunk(&e, error_queue_size);
+
+ shm_toc_estimate_keys(&e, 3);

Overall, three keys are used but the comment indicates two. You forgot
to mention about error_queue.

4.
+ if (launched)
+ ParallelApplyWorkersList = lappend(ParallelApplyWorkersList, winfo);
+ else
+ {
+ shm_mq_detach(winfo->mq_handle);
+ shm_mq_detach(winfo->error_mq_handle);
+ dsm_detach(winfo->dsm_seg);
+ pfree(winfo);
+
+ winfo = NULL;
+ }

A. The code used in the else part to free worker info is the same as
what is used in parallel_apply_free_worker. Can we move this to a
separate function say parallel_apply_free_worker_info()?
B. I think it will be better if you use {} for if branch to make it
look consistent with else branch.

5.
+ * case define a named savepoint, so that we are able to commit/rollback it
+ * separately later.
+ */
+void
+parallel_apply_subxact_info_add(TransactionId current_xid)

I don't see the need of commit in the above message. So, we can
slightly modify it to: "... so that we are able to rollback to it
separately later."

6.
+ for (i = list_length(subxactlist) - 1; i >= 0; i--)
+ {
+ xid = list_nth_xid(subxactlist, i);
...
...

+/*
+ * Return the TransactionId value contained in the n'th element of the
+ * specified list.
+ */
+static inline TransactionId
+list_nth_xid(const List *list, int n)
+{
+ Assert(IsA(list, XidList));
+ return lfirst_xid(list_nth_cell(list, n));
+}

I am not really sure that we need a new list function to use for this
place. Can't we directly use lfirst_xid(list_nth_cell) instead?

7.
+void
+parallel_apply_replorigin_setup(void)
+{
+ RepOriginId originid;
+ char originname[NAMEDATALEN];
+ bool started_tx = false;
+
+ /* This function might be called inside or outside of transaction. */
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }

Is there a place in the patch where this function will be called
without having an active transaction state? If so, then this coding is
fine but if not, then I suggest keeping an assert for transaction
state here. The same thing applies to
parallel_apply_replorigin_reset() as well.

8.
+ *
+ * If write_abort_lsn is true, send the abort_lsn and abort_time fields,
+ * otherwise don't.
  */
 void
 logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
-   TransactionId subxid)
+   TransactionId subxid, XLogRecPtr abort_lsn,
+   TimestampTz abort_time, bool abort_info)

In the comment, the name of the variable needs to be updated.

9.
+TransactionId stream_xid = InvalidTransactionId;

-static TransactionId stream_xid = InvalidTransactionId;
...
...
+void
+parallel_apply_subxact_info_add(TransactionId current_xid)
+{
+ if (current_xid != stream_xid &&
+ !list_member_xid(subxactlist, current_xid))

It seems you have changed the scope of stream_xid to use it in
parallel_apply_subxact_info_add(). Won't it be better to pass it as a
parameter (say top_xid)?

10.
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -20,6 +20,7 @@
 #include <sys/time.h>

 #include "access/xlog.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_type.h"
 #include "common/connect.h"
 #include "funcapi.h"
@@ -443,9 +444,14 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
  appendStringInfo(&cmd, "proto_version '%u'",
  options->proto.logical.proto_version);

- if (options->proto.logical.streaming &&
- PQserverVersion(conn->streamConn) >= 140000)
- appendStringInfoString(&cmd, ", streaming 'on'");
+ if (options->proto.logical.streaming != SUBSTREAM_OFF)
+ {
+ if (PQserverVersion(conn->streamConn) >= 160000 &&
+ options->proto.logical.streaming == SUBSTREAM_PARALLEL)
+ appendStringInfoString(&cmd, ", streaming 'parallel'");
+ else if (PQserverVersion(conn->streamConn) >= 140000)
+ appendStringInfoString(&cmd, ", streaming 'on'");
+ }

It doesn't seem like a good idea to expose subscription options here.
Can we think of having char *streaming_option instead of the current
streaming parameter which is filled by the caller and used here
directly?

11. The error message used in pgoutput_startup() seems to be better
than the current messages used in that function but it is better to be
consistent with other messages. There is a discussion in the email
thread [1] on improving those messages, so kindly suggest there.

12. In addition to the above, I have changed/added a few comments in
the attached patch.

[1] - https://www.postgresql.org/message-id/20220914.111507.13049297635620898.horikyota.ntt%40gmail.com

-- 
With Regards,
Amit Kapila.

--00000000000074d22a05e8b5b45f
Content-Type: application/octet-stream;
	name="changed_comments_amit_v29.patch"
Content-Disposition: attachment; filename="changed_comments_amit_v29.patch"
Content-Transfer-Encoding: base64
Content-ID: <f_l82zbn680>
X-Attachment-Id: f_l82zbn680

ZGlmZiAtLWdpdCBhL3NyYy9iYWNrZW5kL3JlcGxpY2F0aW9uL2xvZ2ljYWwvbGF1bmNoZXIuYyBi
L3NyYy9iYWNrZW5kL3JlcGxpY2F0aW9uL2xvZ2ljYWwvbGF1bmNoZXIuYwppbmRleCA4MzQ0ZTQ1
MDliLi5jMTU0ZWVkMzRjIDEwMDY0NAotLS0gYS9zcmMvYmFja2VuZC9yZXBsaWNhdGlvbi9sb2dp
Y2FsL2xhdW5jaGVyLmMKKysrIGIvc3JjL2JhY2tlbmQvcmVwbGljYXRpb24vbG9naWNhbC9sYXVu
Y2hlci5jCkBAIC02NjQsMTUgKzY2NCwxMyBAQCBsb2dpY2FscmVwX3dvcmtlcl9hdHRhY2goaW50
IHNsb3QpCiB9CiAKIC8qCi0gKiBEZXRhY2ggdGhlIHdvcmtlciAoY2xlYW5zIHVwIHRoZSB3b3Jr
ZXIgaW5mbykuCisgKiBTdG9wIHRoZSBwYXJhbGxlbCBhcHBseSB3b3JrZXJzIGlmIGFueSwgYW5k
IGRldGFjaCB0aGUgbGVhZGVyIGFwcGx5CisgKiB3b3JrZXIgKGNsZWFucyB1cCB0aGUgd29ya2Vy
IGluZm8pLgogICovCiBzdGF0aWMgdm9pZAogbG9naWNhbHJlcF93b3JrZXJfZGV0YWNoKHZvaWQp
CiB7Ci0JLyoKLQkgKiBJZiB0aGlzIGlzIHRoZSBsZWFkZXIgYXBwbHkgd29ya2VyIHRoZW4gc3Rv
cCBhbGwgdGhlIHBhcmFsbGVsIGFwcGx5Ci0JICogd29ya2VycyBwcmV2aW91c2x5IHN0YXJ0ZWQg
ZnJvbSBoZXJlLgotCSAqLworCS8qIFN0b3AgdGhlIHBhcmFsbGVsIGFwcGx5IHdvcmtlcnMuICov
CiAJaWYgKCFhbV9wYXJhbGxlbF9hcHBseV93b3JrZXIoKSAmJiAhYW1fdGFibGVzeW5jX3dvcmtl
cigpKQogCXsKIAkJTGlzdAkgICAqd29ya2VyczsKZGlmZiAtLWdpdCBhL3NyYy9iYWNrZW5kL3Jl
cGxpY2F0aW9uL2xvZ2ljYWwvd29ya2VyLmMgYi9zcmMvYmFja2VuZC9yZXBsaWNhdGlvbi9sb2dp
Y2FsL3dvcmtlci5jCmluZGV4IDJiZmYxZGJjNzQuLmU4YWE2NjE5NzEgMTAwNjQ0Ci0tLSBhL3Ny
Yy9iYWNrZW5kL3JlcGxpY2F0aW9uL2xvZ2ljYWwvd29ya2VyLmMKKysrIGIvc3JjL2JhY2tlbmQv
cmVwbGljYXRpb24vbG9naWNhbC93b3JrZXIuYwpAQCAtNTgsMjAgKzU4LDE4IEBACiAgKiBwYXJh
bGxlbCBhcHBseSB3b3JrZXJzIChtYXhfcGFyYWxsZWxfYXBwbHlfd29ya2Vyc19wZXJfc3Vic2Ny
aXB0aW9uKSwgYnV0IHRoaXMKICAqIG1heSB3YXN0ZSBzb21lIG1lbW9yeSBpZiBubyBwcm9jZXNz
IGlzIGFjdHVhbGx5IHN0YXJ0ZWQuCiAgKgotICogVGhlIGR5bmFtaWMgc2hhcmVkIG1lbW9yeSBz
ZWdtZW50IHdpbGwgY29udGFpbiAgKDEpIGEgc2htX21xIHRoYXQgY2FuIGJlCisgKiBUaGUgZHlu
YW1pYyBzaGFyZWQgbWVtb3J5IHNlZ21lbnQgd2lsbCBjb250YWluIChhKSBhIHNobV9tcSB0aGF0
IGNhbiBiZQogICogdXNlZCB0byBzZW5kIGNoYW5nZXMgaW4gdGhlIHRyYW5zYWN0aW9uIGZyb20g
bGVhZGVyIGFwcGx5IHdvcmtlciB0byBwYXJhbGxlbAotICogYXBwbHkgd29ya2VyICgyKSBhbm90
aGVyIHNobV9tcSB0aGF0IGNhbiBiZSB1c2VkIHRvIHNlbmQgZXJyb3JzIChhbmQgb3RoZXIKKyAq
IGFwcGx5IHdvcmtlciAoYikgYW5vdGhlciBzaG1fbXEgdGhhdCBjYW4gYmUgdXNlZCB0byBzZW5k
IGVycm9ycyAoYW5kIG90aGVyCiAgKiBtZXNzYWdlcyByZXBvcnRlZCB2aWEgZWxvZy9lcmVwb3J0
KSBmcm9tIHRoZSBwYXJhbGxlbCBhcHBseSB3b3JrZXIgdG8gbGVhZGVyCi0gKiBhcHBseSB3b3Jr
ZXIgKDMpIG5lY2Vzc2FyeSBpbmZvcm1hdGlvbiB0byBiZSBzaGFyZWQgYW1vbmcgcGFyYWxsZWwg
YXBwbHkKLSAqIHdvcmtlcnMgYW5kIGxlYWRlciBhcHBseSB3b3JrZXIoaS5lLiBpbl9wYXJhbGxl
bF9hcHBseV94YWN0IGZsYWcgYW5kIHRoZQorICogYXBwbHkgd29ya2VyIChjKSBuZWNlc3Nhcnkg
aW5mb3JtYXRpb24gdG8gYmUgc2hhcmVkIGFtb25nIHBhcmFsbGVsIGFwcGx5CisgKiB3b3JrZXJz
IGFuZCBsZWFkZXIgYXBwbHkgd29ya2VyIChpLmUuIGluX3BhcmFsbGVsX2FwcGx5X3hhY3QgZmxh
ZyBhbmQgdGhlCiAgKiBjb3JyZXNwb25kaW5nIExvZ2ljYWxSZXBXb3JrZXIgc2xvdCBpbmZvcm1h
dGlvbikuCiAgKgotICogMikgV3JpdGUgdG8gdGVtcG9yYXJ5IGZpbGVzIGFuZCBhcHBseSB3aGVu
IHRoZSBmaW5hbCBjb21taXQgYXJyaXZlcworICogSW4gY2FzZSwgbm8gd29ya2VyIGlzIGF2YWls
YWJsZSB0byBoYW5kbGUgdGhlIHN0cmVhbWVkIHRyYW5zYWN0aW9uLCB3ZQorICogZm9sbG93IGFw
cHJvYWNoIDIuCiAgKgotICogSWYgc3RyZWFtaW5nID0gdHJ1ZSwgb3IgaWYgc3RyZWFtaW5nID0g
cGFyYWxsZWwgYnV0IHRoZXJlIGFyZSBub3QgcGFyYWxsZWwKLSAqIGFwcGx5IHdvcmtlcnMgYXZh
aWxhYmxlIHRvIGhhbmRsZSB0aGUgc3RyZWFtZWQgdHJhbnNhY3Rpb24sIHRoZSBkYXRhIGlzCi0g
KiB3cml0dGVuIHRvIHRlbXBvcmFyeSBmaWxlcyBhbmQgdGhlbiBhcHBsaWVkIGF0IG9uY2Ugd2hl
biB0aGUgZmluYWwgY29tbWl0Ci0gKiBhcnJpdmVzLgorICogMikgV3JpdGUgdG8gdGVtcG9yYXJ5
IGZpbGVzIGFuZCBhcHBseSB3aGVuIHRoZSBmaW5hbCBjb21taXQgYXJyaXZlcwogICoKICAqIFVu
bGlrZSB0aGUgcmVndWxhciAobm9uLXN0cmVhbWVkKSBjYXNlLCBoYW5kbGluZyBzdHJlYW1lZCB0
cmFuc2FjdGlvbnMgaGFzCiAgKiB0byBoYW5kbGUgYWJvcnRzIG9mIGJvdGggdGhlIHRvcGxldmVs
IHRyYW5zYWN0aW9uIGFuZCBzdWJ0cmFuc2FjdGlvbnMuIFRoaXMK
--00000000000074d22a05e8b5b45f--