173b56f1ef
Since its introduction, LogLogicalMessage() (via the SQL interface pg_logical_emit_message()) has never included a call to XLogFlush(), causing it to potentially lose messages on a crash when used in non-transactional mode. This has come up to me as a problem while playing with ideas to design a test suite for what has become 039_end_of_wal.pl introduced in bae868caf222 by Thomas Munro, because there are no direct ways to force a WAL flush via SQL. The default is false, to not flush messages and influence existing use-cases where this function could be used. If set to true, the message emitted is flushed before returning back to the caller, making the message durable on crash. This new option has no effect when using pg_logical_emit_message() in transactional mode, as the record's flush is guaranteed by the WAL record generated by the transaction committed. Two queries of test_decoding are tweaked to cover the new code path for the flush. Bump catalog version. Author: Michael Paquier Reviewed-by: Andres Freund, Amit Kapila, Fujii Masao, Tung Nguyen, Tomas Vondra Discussion: https://postgr.es/m/ZNsdThSe2qgsfs7R@paquier.xyz
99 lines
2.9 KiB
C
99 lines
2.9 KiB
C
/*-------------------------------------------------------------------------
|
|
*
|
|
* message.c
|
|
* Generic logical messages.
|
|
*
|
|
* Copyright (c) 2013-2023, PostgreSQL Global Development Group
|
|
*
|
|
* IDENTIFICATION
|
|
* src/backend/replication/logical/message.c
|
|
*
|
|
* NOTES
|
|
*
|
|
* Generic logical messages allow XLOG logging of arbitrary binary blobs that
|
|
* get passed to the logical decoding plugin. In normal XLOG processing they
|
|
* are same as NOOP.
|
|
*
|
|
* These messages can be either transactional or non-transactional.
|
|
* Transactional messages are part of current transaction and will be sent to
|
|
* decoding plugin using in a same way as DML operations.
|
|
* Non-transactional messages are sent to the plugin at the time when the
|
|
* logical decoding reads them from XLOG. This also means that transactional
|
|
* messages won't be delivered if the transaction was rolled back but the
|
|
* non-transactional one will always be delivered.
|
|
*
|
|
* Every message carries prefix to avoid conflicts between different decoding
|
|
* plugins. The plugin authors must take extra care to use unique prefix,
|
|
* good options seems to be for example to use the name of the extension.
|
|
*
|
|
* ---------------------------------------------------------------------------
|
|
*/
|
|
|
|
#include "postgres.h"
|
|
|
|
#include "access/xact.h"
|
|
#include "access/xloginsert.h"
|
|
#include "miscadmin.h"
|
|
#include "nodes/execnodes.h"
|
|
#include "replication/logical.h"
|
|
#include "replication/message.h"
|
|
#include "utils/memutils.h"
|
|
|
|
/*
|
|
* Write logical decoding message into XLog.
|
|
*/
|
|
XLogRecPtr
|
|
LogLogicalMessage(const char *prefix, const char *message, size_t size,
|
|
bool transactional, bool flush)
|
|
{
|
|
xl_logical_message xlrec;
|
|
XLogRecPtr lsn;
|
|
|
|
/*
|
|
* Force xid to be allocated if we're emitting a transactional message.
|
|
*/
|
|
if (transactional)
|
|
{
|
|
Assert(IsTransactionState());
|
|
GetCurrentTransactionId();
|
|
}
|
|
|
|
xlrec.dbId = MyDatabaseId;
|
|
xlrec.transactional = transactional;
|
|
/* trailing zero is critical; see logicalmsg_desc */
|
|
xlrec.prefix_size = strlen(prefix) + 1;
|
|
xlrec.message_size = size;
|
|
|
|
XLogBeginInsert();
|
|
XLogRegisterData((char *) &xlrec, SizeOfLogicalMessage);
|
|
XLogRegisterData(unconstify(char *, prefix), xlrec.prefix_size);
|
|
XLogRegisterData(unconstify(char *, message), size);
|
|
|
|
/* allow origin filtering */
|
|
XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
|
|
|
|
lsn = XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);
|
|
|
|
/*
|
|
* Make sure that the message hits disk before leaving if emitting a
|
|
* non-transactional message when flush is requested.
|
|
*/
|
|
if (!transactional && flush)
|
|
XLogFlush(lsn);
|
|
return lsn;
|
|
}
|
|
|
|
/*
|
|
* Redo is basically just noop for logical decoding messages.
|
|
*/
|
|
void
|
|
logicalmsg_redo(XLogReaderState *record)
|
|
{
|
|
uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
|
|
|
|
if (info != XLOG_LOGICAL_MESSAGE)
|
|
elog(PANIC, "logicalmsg_redo: unknown op code %u", info);
|
|
|
|
/* This is only interesting for logical decoding, see decode.c. */
|
|
}
|