On Tue, Aug 15, 2023 at 09:16:53PM -0700, Andres Freund wrote: > To be clear: I don't just object to backpatching, I also object to making > existing invocations flush WAL in HEAD. I do not at all object to adding a > parameter that indicates flushing, or a separate function to do so. The latter > might be better, because it allows you to flush a group of messages, rather > than a single one.
For the latter, am I getting it right that you mean a function completely outside of the scope of LogLogicalMessage() and pg_logical_emit_message()? Say, a single pg_wal_flush(lsn)? I am a bit concerned by that, because anybody calling directly LogLogicalMessage() or the existing function would never really think about durability but they may want to ensure a message flush in their code calling it. Adding an argument does not do much about the SQL function if it has a DEFAULT, still it addresses my first concern about the C part. Anyway, attached is a patch to add a 4th argument "flush" that defaults to false. Thoughts about this version are welcome. -- Michael
From c70c001539b91422d4090d4e02d10473536f9cc7 Mon Sep 17 00:00:00 2001
From: Michael Paquier <mich...@paquier.xyz>
Date: Wed, 16 Aug 2023 16:49:52 +0900
Subject: [PATCH v3] Add flush argument to pg_logical_emit_message()
The default is false, to not flush messages. If set to true, the
message is flushed before returning back to the caller.
---
src/include/catalog/pg_proc.dat | 4 ++--
src/include/replication/message.h | 3 ++-
src/backend/catalog/system_functions.sql | 20 +++++++++++++++++++
.../replication/logical/logicalfuncs.c | 3 ++-
src/backend/replication/logical/message.c | 13 ++++++++++--
doc/src/sgml/func.sgml | 6 +++++-
contrib/test_decoding/expected/messages.out | 5 +++--
contrib/test_decoding/sql/messages.sql | 5 +++--
8 files changed, 48 insertions(+), 11 deletions(-)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 6996073989..142e0c3fea 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11139,11 +11139,11 @@
prosrc => 'pg_replication_slot_advance' },
{ oid => '3577', descr => 'emit a textual logical decoding message',
proname => 'pg_logical_emit_message', provolatile => 'v', proparallel => 'u',
- prorettype => 'pg_lsn', proargtypes => 'bool text text',
+ prorettype => 'pg_lsn', proargtypes => 'bool text text bool',
prosrc => 'pg_logical_emit_message_text' },
{ oid => '3578', descr => 'emit a binary logical decoding message',
proname => 'pg_logical_emit_message', provolatile => 'v', proparallel => 'u',
- prorettype => 'pg_lsn', proargtypes => 'bool text bytea',
+ prorettype => 'pg_lsn', proargtypes => 'bool text bytea bool',
prosrc => 'pg_logical_emit_message_bytea' },
# event triggers
diff --git a/src/include/replication/message.h b/src/include/replication/message.h
index 6ce7f2038b..0f168d572c 100644
--- a/src/include/replication/message.h
+++ b/src/include/replication/message.h
@@ -30,7 +30,8 @@ typedef struct xl_logical_message
#define SizeOfLogicalMessage (offsetof(xl_logical_message, message))
extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message,
- size_t size, bool transactional);
+ size_t size, bool transactional,
+ bool flush);
/* RMGR API */
#define XLOG_LOGICAL_MESSAGE 0x00
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index 07c0d89c4f..714f5da941 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -446,6 +446,26 @@ LANGUAGE INTERNAL
VOLATILE ROWS 1000 COST 1000
AS 'pg_logical_slot_peek_binary_changes';
+CREATE OR REPLACE FUNCTION pg_catalog.pg_logical_emit_message(
+ transactional boolean,
+ prefix text,
+ message text,
+ flush boolean DEFAULT false)
+RETURNS pg_lsn
+LANGUAGE INTERNAL
+VOLATILE STRICT
+AS 'pg_logical_emit_message_text';
+
+CREATE OR REPLACE FUNCTION pg_catalog.pg_logical_emit_message(
+ transactional boolean,
+ prefix text,
+ message bytea,
+ flush boolean DEFAULT false)
+RETURNS pg_lsn
+LANGUAGE INTERNAL
+VOLATILE STRICT
+AS 'pg_logical_emit_message_bytea';
+
CREATE OR REPLACE FUNCTION pg_create_physical_replication_slot(
IN slot_name name, IN immediately_reserve boolean DEFAULT false,
IN temporary boolean DEFAULT false,
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 55a24c02c9..a472a0d873 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -362,10 +362,11 @@ pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
bool transactional = PG_GETARG_BOOL(0);
char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
bytea *data = PG_GETARG_BYTEA_PP(2);
+ bool flush = PG_GETARG_BOOL(3);
XLogRecPtr lsn;
lsn = LogLogicalMessage(prefix, VARDATA_ANY(data), VARSIZE_ANY_EXHDR(data),
- transactional);
+ transactional, flush);
PG_RETURN_LSN(lsn);
}
diff --git a/src/backend/replication/logical/message.c b/src/backend/replication/logical/message.c
index c5de14afc6..d49557f46e 100644
--- a/src/backend/replication/logical/message.c
+++ b/src/backend/replication/logical/message.c
@@ -44,9 +44,10 @@
*/
XLogRecPtr
LogLogicalMessage(const char *prefix, const char *message, size_t size,
- bool transactional)
+ bool transactional, bool flush)
{
xl_logical_message xlrec;
+ XLogRecPtr lsn;
/*
* Force xid to be allocated if we're emitting a transactional message.
@@ -71,7 +72,15 @@ LogLogicalMessage(const char *prefix, const char *message, size_t size,
/* allow origin filtering */
XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
- return XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);
+ lsn = XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);
+
+ /*
+ * Make sure that the message hits disk before leaving if not emitting a
+ * transactional message, if flush is requested.
+ */
+ if (!transactional && flush)
+ XLogFlush(lsn);
+ return lsn;
}
/*
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index be2f54c914..97ae707acf 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -27581,7 +27581,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<returnvalue>pg_lsn</returnvalue>
</para>
<para role="func_signature">
- <function>pg_logical_emit_message</function> ( <parameter>transactional</parameter> <type>boolean</type>, <parameter>prefix</parameter> <type>text</type>, <parameter>content</parameter> <type>bytea</type> )
+ <function>pg_logical_emit_message</function> ( <parameter>transactional</parameter> <type>boolean</type>, <parameter>prefix</parameter> <type>text</type>, <parameter>content</parameter> <type>bytea</type> [, <parameter>flush</parameter> <type>boolean</type> <literal>DEFAULT</literal> <literal>false</literal>] )
<returnvalue>pg_lsn</returnvalue>
</para>
<para>
@@ -27595,6 +27595,10 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
recognize messages that are interesting for them.
The <parameter>content</parameter> parameter is the content of the
message, given either in text or binary form.
+ The <parameter>flush</parameter> parameter controls if the message
+ is immediately flushed to WAL or not. <parameter>flush</parameter>
+ has no effect with <parameter>transactional</parameter>, as the
+ message's WAL record is flushed when its transaction is committed.
</para></entry>
</row>
</tbody>
diff --git a/contrib/test_decoding/expected/messages.out b/contrib/test_decoding/expected/messages.out
index 0fd70036bd..84baf8af3e 100644
--- a/contrib/test_decoding/expected/messages.out
+++ b/contrib/test_decoding/expected/messages.out
@@ -6,13 +6,14 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d
init
(1 row)
-SELECT 'msg1' FROM pg_logical_emit_message(true, 'test', 'msg1');
+-- These two cover the path for the flush variant.
+SELECT 'msg1' FROM pg_logical_emit_message(true, 'test', 'msg1', true);
?column?
----------
msg1
(1 row)
-SELECT 'msg2' FROM pg_logical_emit_message(false, 'test', 'msg2');
+SELECT 'msg2' FROM pg_logical_emit_message(false, 'test', 'msg2', true);
?column?
----------
msg2
diff --git a/contrib/test_decoding/sql/messages.sql b/contrib/test_decoding/sql/messages.sql
index 3d8500f99c..1f3dcb63ee 100644
--- a/contrib/test_decoding/sql/messages.sql
+++ b/contrib/test_decoding/sql/messages.sql
@@ -3,8 +3,9 @@ SET synchronous_commit = on;
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
-SELECT 'msg1' FROM pg_logical_emit_message(true, 'test', 'msg1');
-SELECT 'msg2' FROM pg_logical_emit_message(false, 'test', 'msg2');
+-- These two cover the path for the flush variant.
+SELECT 'msg1' FROM pg_logical_emit_message(true, 'test', 'msg1', true);
+SELECT 'msg2' FROM pg_logical_emit_message(false, 'test', 'msg2', true);
BEGIN;
SELECT 'msg3' FROM pg_logical_emit_message(true, 'test', 'msg3');
--
2.40.1
signature.asc
Description: PGP signature
