Re: Proposal: Generic WAL logical messages

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.

> >
> >>+/*
> >>+ * 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.

Greetings,

Andres Freund