Re: Proposal: Generic WAL logical messages

Hi,

thanks for looking Andres,

On 27/02/16 01:05, Andres Freund wrote:
> Hi,
>
> I'm not really convinced by RegisterStandbyMsgPrefix() et al. There's
> not much documentation about what it actually is supposed to
> acomplish. Afaics you're basically forced to use
> shared_preload_libraries with it right now? Also, iterating through a
> linked list everytime something is logged doesn't seem very satisfying?
>

Well, my reasoning there was to stop multiple plugins from using same
prefix and for that you need some kind of registry. Making this a shared
catalog seemed like huge overkill given the potentially transient nature
of output plugins and this was the best I could come up with. And yes it
requires you to load your plugin before you can log a message for it.

I am not married to this solution so if you have better ideas or if you
think the clashes are not read issue, we can change it.

>
> On 2016-02-24 18:35:16 +0100, Petr Jelinek wrote:
>> +SET synchronous_commit = on;
>> +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
>> + ?column?
>> +----------
>> + init
>> +(1 row)
>
>> +SELECT 'msg1' FROM pg_logical_send_message(true, 'test', 'msg1');
>> + ?column?
>> +----------
>> + msg1
>> +(1 row)
>
> Hm. Somehow 'sending' a message seems wrong here. Maybe 'emit'?
>
>> + <row>
>> + <entry id="pg-logical-send-message-text">
>> + <indexterm>
>> + <primary>pg_logical_send_message</primary>
>> + </indexterm>
>> + <literal><function>pg_logical_send_message(<parameter>transactional</parameter> <type>bool</type>, <parameter>prefix</parameter> <type>text</type>, <parameter>content</parameter> <type>text</type>)</function></literal>
>> + </entry>
>> + <entry>
>> + void
>> + </entry>
>> + <entry>
>> + Write text logical decoding message. This can be used to pass generic
>> + messages to logical decoding plugins through WAL. The parameter
>> + <parameter>transactional</parameter> specifies if the message should
>> + be part of current transaction or if it should be written and decoded
>> + immediately. The <parameter>prefix</parameter> has to be prefix which
>> + was registered by a plugin. The <parameter>content</parameter> is
>> + content of the message.
>> + </entry>
>> + </row>
>
> It's not decoded immediately, even if emitted non-transactionally.
>

Okay, immediately is somewhat misleading. How does "should be written
immediately and decoded as soon as logical decoding reads the WAL
record" sound ?

>> + <sect3 id="logicaldecoding-output-plugin-message">
>> + <title>Generic Message Callback</title>
>> +
>> + <para>
>> + The optional <function>message_cb</function> callback is called whenever
>> + a logical decoding message has been decoded.
>> +<programlisting>
>> +typedef void (*LogicalDecodeMessageCB) (
>> + struct LogicalDecodingContext *,
>> + ReorderBufferTXN *txn,
>> + XLogRecPtr message_lsn,
>> + bool transactional,
>> + const char *prefix,
>> + Size message_size,
>> + const char *message
>> +);
>
> We should at least document what txn is set to if not transactional.
>

Will do (it's NULL).

>> +void
>> +logicalmsg_desc(StringInfo buf, XLogReaderState *record)
>> +{
>> + char *rec = XLogRecGetData(record);
>> + xl_logical_message *xlrec = (xl_logical_message *) rec;
>> +
>> + appendStringInfo(buf, "%s message size %zu bytes",
>> + xlrec->transactional ? "transactional" : "nontransactional",
>> + xlrec->message_size);
>> +}
>
> Shouldn't we check
> uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
> if XLogRecGetInfo(record) == XLOG_LOGICAL_MESSAGE
> here?
>

I thought it's kinda pointless, but we seem to be doing it in other
places so will add.

>
>> +void
>> +ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
>> + bool transactional, const char *prefix, Size msg_sz,
>> + const char *msg)
>> +{
>> + ReorderBufferTXN *txn = NULL;
>> +
>> + if (transactional)
>> + {
>> + ReorderBufferChange *change;
>> +
>> + txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
>> +
>> + Assert(xid != InvalidTransactionId);
>> + Assert(txn != NULL);
>> +
>> + change = ReorderBufferGetChange(rb);
>> + change->action = REORDER_BUFFER_CHANGE_MESSAGE;
>> + change->data.msg.transactional = true;
>> + change->data.msg.prefix = pstrdup(prefix);
>> + change->data.msg.message_size = msg_sz;
>> + change->data.msg.message = palloc(msg_sz);
>> + memcpy(change->data.msg.message, msg, msg_sz);
>> +
>> + ReorderBufferQueueChange(rb, xid, lsn, change);
>> + }
>> + else
>> + {
>> + rb->message(rb, txn, lsn, transactional, prefix, msg_sz, msg);
>> + }
>> +}
>
>
> This approach prohibts catalog access when processing a nontransaction
> message as there's no snapshot set up.
>

Hmm I do see usefulness in having snapshot, although I wonder if that
does not kill the point of non-transactional messages. Question is then
though which snapshot should the message see, base_snapshot of
transaction? That would mean we'd have to call SnapBuildProcessChange
for non-transactional messages which we currently avoid. Alternatively
we could probably invent lighter version of that interface that would
just make sure builder->snapshot is valid and if not then build it but
I am honestly sure if that's a win or not.

--
Petr Jelinek http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services