Re: Proposal: Generic WAL logical messages

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.

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

> 3) This lacks error handling, we surely don't want to error out while
> still having the historic snapshot setup
> 4) Without 3) the volatile is bogus.
> 5) Misses a ReorderBufferProcessXid() call.

Fixed (all 3 above).

>
>
>> @@ -414,6 +414,14 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change)
>> change->data.tp.oldtuple = NULL;
>> }
>> break;
>> + case REORDER_BUFFER_CHANGE_MESSAGE:
>> + if (change->data.msg.prefix != NULL)
>> + pfree(change->data.msg.prefix);
>> + change->data.msg.prefix = NULL;
>> + if (change->data.msg.message != NULL)
>> + pfree(change->data.msg.message);
>> + change->data.msg.message = NULL;
>> + break;
>
> Hm, this will have some overhead, but I guess the messages won't be
> super frequent, and usually not very large.

Yeah but since we don't really know the size of the future messages it's
hard to have some preallocated buffer for this so I dunno how else to do it.

>
>> +/*
>> + * Queue message into a transaction so it can be processed upon commit.
>> + */
>> +void
>> +ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
>> + const char *prefix, Size msg_sz, const char *msg)
>> +{
>> + ReorderBufferChange *change;
>> +
>> + Assert(xid != InvalidTransactionId);
>> +
>> + change = ReorderBufferGetChange(rb);
>> + change->action = REORDER_BUFFER_CHANGE_MESSAGE;
>> + 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);
>> +}
>
> I'm not sure right now if there's any guarantee that the current memory
> context is meaningful here? IIRC other long-lived allocations explicitly
> use a context?
>

I didn't find any explicit guarantee so I added one.

>> + case REORDER_BUFFER_CHANGE_MESSAGE:
>> + {
>> + char *data;
>> + size_t prefix_size = strlen(change->data.msg.prefix) + 1;
>> +
>> + sz += prefix_size + change->data.msg.message_size;
>> + ReorderBufferSerializeReserve(rb, sz);
>> +
>> + data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
>> + memcpy(data, change->data.msg.prefix,
>> + prefix_size);
>> + memcpy(data + prefix_size, change->data.msg.message,
>> + change->data.msg.message_size);
>> + break;
>> + }
>
> Can you please include the sizes of the blocks explicitly, rather than
> relying on 0 termination?
>

Okay, I see I did that in WAL, no idea why I didn't do the same here.

>
>> @@ -45,3 +45,4 @@ PG_RMGR(RM_SPGIST_ID, "SPGist", spg_redo, spg_desc, spg_identify, spg_xlog_start
>> PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL)
>> PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_identify, NULL, NULL)
>> PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL)
>> +PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo,
>> logicalmsg_desc, logicalmsg_identify, NULL, NULL)
>
> Did you consider doing this via the standby rmgr instead?
>

Yes in one of the first versions I did that but Simon didn't like that
in his review as this has nothing to do with standby.

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