Re: Any objections to implementing LogicalDecodeMessageCB for pgoutput?

On Thu, Nov 5, 2020 at 9:16 AM David Pirotte <dpirotte(at)gmail(dot)com> wrote:
>
> On Tue, Nov 3, 2020 at 7:19 AM Dave Cramer <davecramer(at)gmail(dot)com> wrote:
>>
>> David,
>>
>> On Thu, 24 Sep 2020 at 00:22, Michael Paquier <michael(at)paquier(dot)xyz> wrote:
>>>
>>> On Tue, Sep 08, 2020 at 12:18:23PM -0700, Andres Freund wrote:
>>> > A test verifying that a non-transactional message in later aborted
>>> > transaction is handled correctly would be good.
>>>
>>> On top of that, the patch needs a rebase as it visibly fails to apply,
>>> per the CF bot.
>>> --
>>> Michael
>>
>>
>> Where are you with this? Are you able to work on it ?
>> Dave Cramer
>
>
> Apologies for the delay, here.
>
> I've attached v2 of this patch which applies cleanly to master. The patch also now includes a test demonstrating that pg_logical_emit_message correctly sends non-transactional messages when called inside a transaction that is rolled back. (Thank you, Andres, for this suggestion.) The only other change is that I added this new message type into the LogicalRepMsgType enum added earlier this week.
>
> Let me know what you think.

This feature looks useful. Here are some comments.

+/*
+ * Write MESSAGE to stream
+ */
+void
+logicalrep_write_message(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr lsn,
+ bool transactional, const char *prefix, Size sz,
+ const char *message)
+{
+ uint8 flags = 0;
+
+ pq_sendbyte(out, LOGICAL_REP_MSG_MESSAGE);
+

Similar to the UPDATE/DELETE/INSERT records decoded when streaming is being
used, we need to add transaction id for transactional messages. May be we add
that even in case of non-streaming case and use it to decide whether it's a
transactional message or not. That might save us a byte when we are adding a
transaction id.

+ /* encode and send message flags */
+ if (transactional)
+ flags |= MESSAGE_TRANSACTIONAL;
+
+ pq_sendint8(out, flags);

Is 8 bits enough considering future improvements? What if we need to use more
than 8 bit flags?

@@ -1936,6 +1936,9 @@ apply_dispatch(StringInfo s)
apply_handle_origin(s);
return;

+ case LOGICAL_REP_MSG_MESSAGE:

Should we add the logical message to the WAL downstream so that it flows
further down to a cascaded logical replica. Should that be controlled
by an option?

--
Best Wishes,
Ashutosh Bapat