On Tue, May 25, 2021 at 2:34 PM Dilip Kumar <dilipbal...@gmail.com> wrote:
>
> > When the transaction is streamed, I see:
> > ```
> > + opening a streamed block for transaction
> > + table public.toasted: INSERT: id[integer]:9001 other[text]:'bbb'
> > data[text]:'ccc'
> > + table public.toasted: INSERT: id[integer]:9002 other[text]:'ddd'
> > data[text]:'eee'
> > + table public.toasted: INSERT: id[integer]:9003 other[text]:'bar'
> > data[text]:unchanged-toast-datum
> > <snipped>
> > ```
> >
> > For a non-streamed case, the `data[text]` column shows the actual data.
> > That probably manifests into NULL data when downstream handles it.
>
> Yes, I am able to reproduce this, basically, until we get the last
> tuple of the multi insert we can not clear the toast data otherwise we
> can never form a complete tuple. So the only possible fix I can think
> of is to consider the multi-insert WAL without the final multi-insert
> tuple as partial data then we will avoid streaming until we get the
> complete WAL of one multi-insert. I am working on the patch to fix
> this, I will share that in some time.
The attached patch should fix the issue, now the output is like below
===
opening a streamed block for transaction
table public.toasted: INSERT: id[integer]:9001 other[text]:'bbb'
data[text]:'ccc'
table public.toasted: INSERT: id[integer]:9002 other[text]:'ddd'
data[text]:'eee'
table public.toasted: INSERT: id[integer]:9003 other[text]:'bar'
data[text]:'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
<repeat >
===
--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com
From fc58da41802f966dd269679921a5daa33d21a291 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumar@localhost.localdomain>
Date: Tue, 25 May 2021 14:51:45 +0530
Subject: [PATCH v1] Fix bug while streaming the multi-insert toast changes
While processing the multi-insert we can not clean the toast untill
we get the last insert of the multi-insert. So mark the transaction
incomplete for streaming if we get any multi-insert change, and keep
it set until we get the last tuple of the multi-insert.
---
src/backend/replication/logical/reorderbuffer.c | 11 +++++++++++
src/include/replication/reorderbuffer.h | 11 ++++++++++-
2 files changed, 21 insertions(+), 1 deletion(-)
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index b0ab91c..b715992 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -732,6 +732,17 @@ ReorderBufferProcessPartialChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
}
/*
+ * Set the multi insert bit whenever we get the clear_toast_afterwards
+ * as false otherwise reset it if this is set.
+ */
+ if (!change->data.tp.clear_toast_afterwards)
+ {
+ toptxn->txn_flags |= RBTXN_HAS_MULTI_INSERT;
+ }
+ else if rbtxn_has_multi_insert(toptxn)
+ toptxn->txn_flags &= ~RBTXN_HAS_MULTI_INSERT;
+
+ /*
* Stream the transaction if it is serialized before and the changes are
* now complete in the top-level transaction.
*
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 53cdfa5..cd36a8d 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -176,6 +176,7 @@ typedef struct ReorderBufferChange
#define RBTXN_HAS_SPEC_INSERT 0x0040
#define RBTXN_PREPARE 0x0080
#define RBTXN_SKIPPED_PREPARE 0x0100
+#define RBTXN_HAS_MULTI_INSERT 0x0200
/* Does the transaction have catalog changes? */
#define rbtxn_has_catalog_changes(txn) \
@@ -214,11 +215,19 @@ typedef struct ReorderBufferChange
( \
((txn)->txn_flags & RBTXN_HAS_SPEC_INSERT) != 0 \
)
+/*
+ * This transaction's changes has multi insert, without last
+ * multi insert tuple.
+ */
+#define rbtxn_has_multi_insert(txn) \
+( \
+ ((txn)->txn_flags & RBTXN_HAS_MULTI_INSERT) != 0 \
+)
/* Check whether this transaction has an incomplete change. */
#define rbtxn_has_incomplete_tuple(txn) \
( \
- rbtxn_has_toast_insert(txn) || rbtxn_has_spec_insert(txn) \
+ rbtxn_has_toast_insert(txn) || rbtxn_has_spec_insert(txn) || rbtxn_has_multi_insert(txn) \
)
/*
--
1.8.3.1
