Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions

On Mon, Dec 2, 2019 at 2:02 PM Dilip Kumar <dilipbalaut(at)gmail(dot)com> wrote:
>
> On Sun, Dec 1, 2019 at 7:58 AM Michael Paquier <michael(at)paquier(dot)xyz> wrote:
> >
> > On Fri, Nov 22, 2019 at 01:18:11PM +0530, Dilip Kumar wrote:
> > > I have rebased the patch on the latest head and also fix the issue of
> > > "concurrent abort handling of the (sub)transaction." and attached as
> > > (v1-0013-Extend-handling-of-concurrent-aborts-for-streamin) along with
> > > the complete patch set. I have added the version number so that we
> > > can track the changes.
> >
> > The patch has rotten a bit and does not apply anymore. Could you
> > please send a rebased version? I have moved it to next CF, waiting on
> > author.
>
> I have rebased the patch set on the latest head.
>

Few comments:
assert variable should be within #ifdef USE_ASSERT_CHECKING in patch
v2-0008-Add-support-for-streaming-to-built-in-replication.patch:
+ int64 subidx;
+ bool found = false;
+ char path[MAXPGPATH];
+
+ subidx = -1;
+ subxact_info_read(MyLogicalRepWorker->subid, xid);
+
+ /* FIXME optimize the search by bsearch on sorted data */
+ for (i = nsubxacts; i > 0; i--)
+ {
+ if (subxacts[i - 1].xid == subxid)
+ {
+ subidx = (i - 1);
+ found = true;
+ break;
+ }
+ }
+
+ /* We should not receive aborts for unknown subtransactions. */
+ Assert(found);

Add the typedefs like below in typedefs.lst common across the patches:
xl_xact_invalidations, ReorderBufferStreamIterTXNEntry,
ReorderBufferStreamIterTXNState, SubXactInfo

"are written" appears twice in commit message of
v2-0002-Issue-individual-invalidations-with-wal_level-log.patch:
The individual invalidations are written are written using a new
xlog record type XLOG_XACT_INVALIDATIONS, from RM_XACT_ID resource
manager. See LogLogicalInvalidations for details.

v2-0002-Issue-individual-invalidations-with-wal_level-log.patch patch
does not compile by itself:
reorderbuffer.c:1822:9: error: ‘ReorderBufferTXN’ has no member named
‘is_schema_sent’
+
LocalExecuteInvalidationMessage(&change->data.inval.msg);
+ txn->is_schema_sent = false;
+ break;

Should we include printing of id here like in earlier cases in
v2-0002-Issue-individual-invalidations-with-wal_level-log.patch:
+ appendStringInfo(buf, " relcache %u", msg->rc.relId);
+ /* not expected, but print something anyway */
+ else if (msg->id == SHAREDINVALSMGR_ID)
+ appendStringInfoString(buf, " smgr");
+ /* not expected, but print something anyway */
+ else if (msg->id == SHAREDINVALRELMAP_ID)
+ appendStringInfo(buf, " relmap db %u", msg->rm.dbId);

There is some code duplication in stream_change_cb_wrapper,
stream_truncate_cb_wrapper, stream_message_cb_wrapper,
stream_abort_cb_wrapper, stream_commit_cb_wrapper,
stream_start_cb_wrapper and stream_stop_cb_wrapper functions in
v2-0003-Extend-the-output-plugin-API-with-stream-methods.patch patch.
Should we have a separate function for common code?

Should we can add function header for AssertChangeLsnOrder in
v2-0006-Implement-streaming-mode-in-ReorderBuffer.patch:
+static void
+AssertChangeLsnOrder(ReorderBuffer *rb, ReorderBufferTXN *txn)
+{

This "Assert(txn->first_lsn != InvalidXLogRecPtr)"can be before the
loop, can be checked only once:
+ dlist_foreach(iter, &txn->changes)
+ {
+ ReorderBufferChange *cur_change;
+
+ cur_change = dlist_container(ReorderBufferChange,
node, iter.cur);
+
+ Assert(txn->first_lsn != InvalidXLogRecPtr);
+ Assert(cur_change->lsn != InvalidXLogRecPtr);
+ Assert(txn->first_lsn <= cur_change->lsn);

Should we add function header for ReorderBufferDestroyTupleCidHash in
v2-0006-Implement-streaming-mode-in-ReorderBuffer.patch:
+static void
+ReorderBufferDestroyTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
+{
+ if (txn->tuplecid_hash != NULL)
+ {
+ hash_destroy(txn->tuplecid_hash);
+ txn->tuplecid_hash = NULL;
+ }
+}
+

Should we add function header for ReorderBufferStreamCommit in
v2-0006-Implement-streaming-mode-in-ReorderBuffer.patch:
+static void
+ReorderBufferStreamCommit(ReorderBuffer *rb, ReorderBufferTXN *txn)
+{
+ /* we should only call this for previously streamed transactions */
+ Assert(rbtxn_is_streamed(txn));
+
+ ReorderBufferStreamTXN(rb, txn);
+
+ rb->stream_commit(rb, txn, txn->final_lsn);
+
+ ReorderBufferCleanupTXN(rb, txn);
+}
+

Should we add function header for ReorderBufferCanStream in
v2-0006-Implement-streaming-mode-in-ReorderBuffer.patch:
+static bool
+ReorderBufferCanStream(ReorderBuffer *rb)
+{
+ LogicalDecodingContext *ctx = rb->private_data;
+
+ return ctx->streaming;
+}

patch v2-0008-Add-support-for-streaming-to-built-in-replication.patch
does not apply:
Hunk #18 FAILED at 2035.
Hunk #19 succeeded at 2199 (offset -16 lines).
1 out of 19 hunks FAILED -- saving rejects to file
src/backend/replication/logical/worker.c.rej

Header inclusion may not be required in patch
v2-0008-Add-support-for-streaming-to-built-in-replication.patch:
+++ b/src/backend/replication/logical/launcher.c
@@ -14,6 +14,8 @@
*
*-------------------------------------------------------------------------
*/
+#include <sys/types.h>
+#include <unistd.h>

Regards,
Vignesh
EnterpriseDB: http://www.enterprisedb.com