database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-11-09 06:21:09 +03:00
Code Activity

Add 'logical_decoding_mode' GUC.

Browse Source 
This enables streaming or serializing changes immediately in logical
decoding. This parameter is intended to be used to test logical decoding
and replication of large transactions for which otherwise we need to
generate the changes till logical_decoding_work_mem is reached.

This helps in reducing the timing of existing tests related to logical
replication of in-progress transactions and will help in writing tests for
for the upcoming feature for parallelly applying large in-progress
transactions.

Author: Shi yu
Reviewed-by: Sawada Masahiko, Shveta Mallik, Amit Kapila, Dilip Kumar, Kuroda Hayato, Kyotaro Horiguchi
Discussion: https://postgr.es/m/OSZPR01MB63104E7449DBE41932DB19F1FD1B9@OSZPR01MB6310.jpnprd01.prod.outlook.com
This commit is contained in:
Amit Kapila
2022-12-26 08:58:16 +05:30
parent d3c0cc4447
commit 5de94a041e
9 changed files with 134 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/backend/replication/logical/reorderbuffer.c
View File

@@ -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_BUFFERED;
/* ---------------------------------------
* primary reorderbuffer support routines
* ---------------------------------------
@@ -3540,7 +3543,10 @@ 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
@@ -3552,20 +3558,27 @@ 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 buffered and we haven't exceeded
* the memory limit.
*/
if (logical_decoding_mode == LOGICAL_DECODING_MODE_BUFFERED &&
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
* logical_decoding_work_mem to a smaller value before the most recent
* 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