thanks for looking Andres,

On 27/02/16 01:05, Andres Freund wrote:

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', 
+ ?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> 
+       </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.
+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).

+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" : 
+                                        xlrec->message_size);

Shouldn't we check
           uint8         info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
           if XLogRecGetInfo(record) == XLOG_LOGICAL_MESSAGE

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

+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

Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:

Reply via email to