From 155aa2c3bce8b2df8f8d7bc9751cd32d4bae2452 Mon Sep 17 00:00:00 2001 From: Shi Yu Date: Wed, 21 Dec 2022 18:46:57 +0800 Subject: [PATCH v3] Allow streaming or serializing each change in logical decoding. Add a new GUC 'logical_decoding_mode'. The default is the empty string, which disables this feature. When set to "immediate", stream each change immediately if "streaming" option is enabled, and if "streaming" option is disabled, serialize each change. --- contrib/test_decoding/expected/stream.out | 13 +++++++ contrib/test_decoding/sql/stream.sql | 8 ++++ doc/src/sgml/config.sgml | 21 +++++++++++ .../replication/logical/reorderbuffer.c | 37 +++++++++++++------ src/backend/utils/misc/guc_tables.c | 17 +++++++++ src/include/replication/reorderbuffer.h | 8 ++++ 6 files changed, 92 insertions(+), 12 deletions(-) diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out index 0f21dcb8e0..b7af8147a3 100644 --- a/contrib/test_decoding/expected/stream.out +++ b/contrib/test_decoding/expected/stream.out @@ -106,6 +106,19 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'incl committing streamed transaction (17 rows) +-- streaming test with logical_decoding_mode=immediate +SET logical_decoding_mode=immediate; +TRUNCATE table stream_test; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1'); + data +------------------------------------------ + opening a streamed block for transaction + streaming truncate for transaction + closing a streamed block for transaction + committing streamed transaction +(4 rows) + +RESET logical_decoding_mode; DROP TABLE stream_test; SELECT pg_drop_replication_slot('regression_slot'); pg_drop_replication_slot diff --git a/contrib/test_decoding/sql/stream.sql b/contrib/test_decoding/sql/stream.sql index 4feec62972..b184c13e6b 100644 --- a/contrib/test_decoding/sql/stream.sql +++ b/contrib/test_decoding/sql/stream.sql @@ -44,5 +44,13 @@ toasted-123456789012345678901234567890123456789012345678901234567890123456789012 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1'); +-- streaming test with logical_decoding_mode=immediate +SET logical_decoding_mode=immediate; + +TRUNCATE table stream_test; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1'); + +RESET logical_decoding_mode; + DROP TABLE stream_test; SELECT pg_drop_replication_slot('regression_slot'); diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 9eedab652d..28d78473d7 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11597,6 +11597,27 @@ LOG: CleanUpLock: deleting: lock(0xb7acd844) id(24688,24696,0,0,0,1) + + logical_decoding_mode (enum) + + logical_decoding_mode configuration parameter + + + + + Allows streaming or serializing changes immediately in logical decoding. + The allowed values of logical_decoding_mode are the + empty string and immediate. When set to + immediate, stream each change if + streaming option is enabled, and if + streaming option is disabled, serialize each change. + The default is the empty string, which disables this feature (stream or + serialize until logical_decoding_work_mem is + exceeded). + + + + diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index b567b8b59e..ae6fe97cf9 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -209,6 +209,9 @@ typedef struct ReorderBufferDiskChange int logical_decoding_work_mem; static const Size max_changes_in_memory = 4096; /* XXX for restore only */ +/* GUC variable */ +int logical_decoding_mode = LOGICAL_DECODING_MODE_DEFAULT; + /* --------------------------------------- * primary reorderbuffer support routines * --------------------------------------- @@ -3540,32 +3543,42 @@ ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb) /* * Check whether the logical_decoding_work_mem limit was reached, and if yes * pick the largest (sub)transaction at-a-time to evict and spill its changes to - * disk until we reach under the memory limit. + * disk or send to the output plugin until we reach under the memory limit. + * + * If logical_decoding_mode is set to "immediate", stream or serialize the changes + * immediately. * * XXX At this point we select the transactions until we reach under the memory - * limit, but we might also adapt a more elaborate eviction strategy - for example - * evicting enough transactions to free certain fraction (e.g. 50%) of the memory - * limit. + * limit, but we might also adapt a more elaborate eviction strategy - for + * example evicting enough transactions to free certain fraction (e.g. 50%) of + * the memory limit. */ static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb) { ReorderBufferTXN *txn; - /* bail out if we haven't exceeded the memory limit */ - if (rb->size < logical_decoding_work_mem * 1024L) + /* + * Bail out if logical_decoding_mode is disabled and we haven't exceeded the + * memory limit. + */ + if (logical_decoding_mode == LOGICAL_DECODING_MODE_DEFAULT + && rb->size < logical_decoding_work_mem * 1024L) return; /* - * Loop until we reach under the memory limit. One might think that just - * by evicting the largest (sub)transaction we will come under the memory - * limit based on assumption that the selected transaction is at least as - * large as the most recent change (which caused us to go over the memory - * limit). However, that is not true because a user can reduce the + * If logical_decoding_mode is immediate, loop until there's no change. + * Otherwise, loop until we reach under the memory limit. One might think + * that just by evicting the largest (sub)transaction we will come under the + * memory limit based on assumption that the selected transaction is at + * least as large as the most recent change (which caused us to go over the + * memory limit). However, that is not true because a user can reduce the * logical_decoding_work_mem to a smaller value before the most recent * change. */ - while (rb->size >= logical_decoding_work_mem * 1024L) + while (rb->size >= logical_decoding_work_mem * 1024L || + (logical_decoding_mode == LOGICAL_DECODING_MODE_IMMEDIATE + && rb->size > 0)) { /* * Pick the largest transaction (or subtransaction) and evict it from diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 436afe1d21..58f46da500 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -395,6 +395,12 @@ static const struct config_enum_entry ssl_protocol_versions_info[] = { {NULL, 0, false} }; +static const struct config_enum_entry logical_decoding_mode_options[] = { + {"", LOGICAL_DECODING_MODE_DEFAULT, false}, + {"immediate", LOGICAL_DECODING_MODE_IMMEDIATE, false}, + {NULL, 0, false} +}; + StaticAssertDecl(lengthof(ssl_protocol_versions_info) == (PG_TLS1_3_VERSION + 2), "array length mismatch"); @@ -4877,6 +4883,17 @@ struct config_enum ConfigureNamesEnum[] = NULL, NULL, NULL }, + { + {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS, + gettext_noop("Allows sending or serializing each change in logical decoding."), + NULL, + GUC_NOT_IN_SAMPLE + }, + &logical_decoding_mode, + LOGICAL_DECODING_MODE_DEFAULT, logical_decoding_mode_options, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index c700b55b1c..046833f1a5 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -18,6 +18,14 @@ #include "utils/timestamp.h" extern PGDLLIMPORT int logical_decoding_work_mem; +extern PGDLLIMPORT int logical_decoding_mode; + +/* possible values for logical_decoding_mode */ +typedef enum +{ + LOGICAL_DECODING_MODE_DEFAULT, + LOGICAL_DECODING_MODE_IMMEDIATE +} LogicalDecodingMode; /* an individual tuple, stored in one chunk of memory */ typedef struct ReorderBufferTupleBuf -- 2.31.1