Re: Proposal: Generic WAL logical messages

On 22/03/16 14:11, Andres Freund wrote:
> On 2016-03-22 14:03:06 +0100, Petr Jelinek wrote:
>> On 22/03/16 12:47, Andres Freund wrote:
>>> On 2016-03-21 18:10:55 +0100, Petr Jelinek wrote:
>>>
>>>> +
>>>> +    <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,
>>>> +    const char *prefix,
>>>> +    Size message_size,
>>>> +    const char *message
>>>> +);
>>>
>>> I see you removed the transactional parameter. I'm doubtful that that's
>>> a good idea: It seems like it'd be rather helpful to pass the
>>> transaction for a nontransaction message that's emitted while an xid was
>>> assigned?
>>>
>>
>> Hmm but won't that give the output plugin even transactions that were later
>> aborted? That seems quite different behavior from how the txn parameter
>> works everywhere else.
>
> Seems harmless to me if called out.
>

All right, after some consideration I agree.

>
>>>
>>>> +/*
>>>> + * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
>>>> + */
>>>> +static void
>>>> +DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
>>>> +{
>>>> +    SnapBuild  *builder = ctx->snapshot_builder;
>>>> +    XLogReaderState *r = buf->record;
>>>> +    uint8        info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
>>>> +    xl_logical_message *message;
>>>> +
>>>> +    if (info != XLOG_LOGICAL_MESSAGE)
>>>> +        elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
>>>> +
>>>> +    message = (xl_logical_message *) XLogRecGetData(r);
>>>> +
>>>> +    if (message->transactional)
>>>> +    {
>>>> +        if (!SnapBuildProcessChange(builder, XLogRecGetXid(r), buf->origptr))
>>>> +            return;
>>>> +
>>>> +        ReorderBufferQueueMessage(ctx->reorder, XLogRecGetXid(r),
>>>> +                                  buf->endptr,
>>>> +                                  message->message, /* first part of message is prefix */
>>>> +                                  message->message_size,
>>>> +                                  message->message + message->prefix_size);
>>>> +    }
>>>> +    else if (SnapBuildCurrentState(builder) == SNAPBUILD_CONSISTENT &&
>>>> +             !SnapBuildXactNeedsSkip(builder, buf->origptr))
>>>> +    {
>>>> +        volatile Snapshot    snapshot_now;
>>>> +        ReorderBuffer       *rb = ctx->reorder;
>>>> +
>>>> +        /* setup snapshot to allow catalog access */
>>>> +        snapshot_now = SnapBuildGetOrBuildSnapshot(builder, XLogRecGetXid(r));
>>>> +        SetupHistoricSnapshot(snapshot_now, NULL);
>>>> +        rb->message(rb, NULL, buf->origptr, message->message,
>>>> +                    message->message_size,
>>>> +                    message->message + message->prefix_size);
>>>> +        TeardownHistoricSnapshot(false);
>>>> +    }
>>>> +}
>>>
>>> A number of things:
>>> 1) The SnapBuildProcessChange needs to be toplevel, not just for
>>>     transactional messages - we can't yet necessarily build a snapshot.
>>
>> Nope, the snapshot state is checked in the else if.
>>
>>> 2) I'm inclined to move even the non-transactional stuff to reorderbuffer.
>>
>> Well, it's not doing anything with reorderbuffer but sure it can be done
>> (didn't do it in the attached though).
>
> I think there'll be some interactions if we ever do some parts in
> parallel and such.  I'd rather keep decode.c to only do the lowest level
> stuff.
>

Did it that way but I do like the resulting code less.

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