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