public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v11] Prevent jumbling of every element in ArrayExpr
5+ messages / 3 participants
[nested] [flat]
* [PATCH v11] Prevent jumbling of every element in ArrayExpr
@ 2022-07-24 09:43 Dmitrii Dolgov <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Dmitrii Dolgov @ 2022-07-24 09:43 UTC (permalink / raw)
pg_stat_statements produces multiple entries for queries like
SELECT something FROM table WHERE col IN (1, 2, 3, ...)
depending on number of parameters, because every element of ArrayExpr is
jumbled. In certain situations it's undesirable, especially if the list
becomes too large.
Make Const expressions contribute nothing to the jumble hash if they're
a part of an ArrayExpr, which length is larger than specified threshold.
Allow to configure the threshold via the new GUC const_merge_threshold
with the default value zero, which disables this feature.
Reviewed-by: Zhihong Yu, Sergey Dudoladov, Robert Haas, Tom Lane
Tested-by: Chengxi Sun
---
.../expected/pg_stat_statements.out | 412 ++++++++++++++++++
.../pg_stat_statements/pg_stat_statements.c | 33 +-
.../sql/pg_stat_statements.sql | 107 +++++
doc/src/sgml/config.sgml | 26 ++
doc/src/sgml/pgstatstatements.sgml | 28 +-
src/backend/nodes/queryjumblefuncs.c | 105 ++++-
src/backend/utils/misc/guc_tables.c | 13 +
src/backend/utils/misc/postgresql.conf.sample | 2 +-
src/include/nodes/queryjumble.h | 5 +-
9 files changed, 712 insertions(+), 19 deletions(-)
diff --git a/contrib/pg_stat_statements/expected/pg_stat_statements.out b/contrib/pg_stat_statements/expected/pg_stat_statements.out
index 9ac5c87c3a..f18f34ae5b 100644
--- a/contrib/pg_stat_statements/expected/pg_stat_statements.out
+++ b/contrib/pg_stat_statements/expected/pg_stat_statements.out
@@ -1141,4 +1141,416 @@ SELECT COUNT(*) FROM pg_stat_statements WHERE query LIKE '%SELECT GROUPING%';
2
(1 row)
+--
+-- Consts merging
+--
+CREATE TABLE test_merge (id int, data int);
+-- IN queries
+-- No merging
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+--------------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4, $5, $6) | 1
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4, $5, $6, $7) | 1
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4, $5, $6, $7, $8) | 1
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4, $5, $6, $7, $8, $9) | 1
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(7 rows)
+
+-- Normal
+SET const_merge_threshold = 5;
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3) | 1
+ SELECT * FROM test_merge WHERE id IN (...) | 5
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(4 rows)
+
+-- On the merge threshold
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4) | 1
+ SELECT * FROM test_merge WHERE id IN (...) | 6
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(4 rows)
+
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 6
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(3 rows)
+
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+ id | data
+----+------
+(0 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 5
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(3 rows)
+
+-- With gaps on the threshold
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN ($1, $2, $3, $4) | 1
+ SELECT * FROM test_merge WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(4 rows)
+
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 2
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(3 rows)
+
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) | 2
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 1
+(3 rows)
+
+-- test constants after merge
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) and data = 2;
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge WHERE id IN (...) and data = $3 | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+-- On table, numeric type causes every constant being wrapped into functions.
+CREATE TABLE test_merge_numeric (id int, data numeric(5, 2));
+SELECT pg_stat_statements_reset();
+ pg_stat_statements_reset
+--------------------------
+
+(1 row)
+
+SELECT * FROM test_merge_numeric WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ id | data
+----+------
+(0 rows)
+
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+ query | calls
+------------------------------------------------------------------------+-------
+ SELECT * FROM test_merge_numeric WHERE id IN (...) | 1
+ SELECT pg_stat_statements_reset() | 1
+ SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C" | 0
+(3 rows)
+
+-- Test find_const_walker
+WITH cte AS (
+ SELECT 'const' as const FROM test_merge
+)
+SELECT ARRAY['a', 'b', 'c', const::varchar] AS result
+FROM cte;
+ result
+--------
+(0 rows)
+
+RESET const_merge_threshold;
DROP EXTENSION pg_stat_statements;
diff --git a/contrib/pg_stat_statements/pg_stat_statements.c b/contrib/pg_stat_statements/pg_stat_statements.c
index ad1fe44496..b26ae1f234 100644
--- a/contrib/pg_stat_statements/pg_stat_statements.c
+++ b/contrib/pg_stat_statements/pg_stat_statements.c
@@ -2666,6 +2666,9 @@ generate_normalized_query(JumbleState *jstate, const char *query,
n_quer_loc = 0, /* Normalized query byte location */
last_off = 0, /* Offset from start for previous tok */
last_tok_len = 0; /* Length (in bytes) of that tok */
+ bool skip = false; /* Signals that certain constants are
+ merged together and have to be skipped */
+
/*
* Get constants' lengths (core system only gives us locations). Note
@@ -2689,7 +2692,6 @@ generate_normalized_query(JumbleState *jstate, const char *query,
{
int off, /* Offset from start for cur tok */
tok_len; /* Length (in bytes) of that tok */
-
off = jstate->clocations[i].location;
/* Adjust recorded location if we're dealing with partial string */
off -= query_loc;
@@ -2704,12 +2706,31 @@ generate_normalized_query(JumbleState *jstate, const char *query,
len_to_wrt -= last_tok_len;
Assert(len_to_wrt >= 0);
- memcpy(norm_query + n_quer_loc, query + quer_loc, len_to_wrt);
- n_quer_loc += len_to_wrt;
- /* And insert a param symbol in place of the constant token */
- n_quer_loc += sprintf(norm_query + n_quer_loc, "$%d",
- i + 1 + jstate->highest_extern_param_id);
+ /* Normal path, non merged constant */
+ if (!jstate->clocations[i].merged)
+ {
+ memcpy(norm_query + n_quer_loc, query + quer_loc, len_to_wrt);
+ n_quer_loc += len_to_wrt;
+
+ /* And insert a param symbol in place of the constant token */
+ n_quer_loc += sprintf(norm_query + n_quer_loc, "$%d",
+ i + 1 + jstate->highest_extern_param_id);
+
+ /* In case previous constants were merged away, stop doing that */
+ if (skip)
+ skip = false;
+ }
+ /* The firsts merged constant */
+ else if (!skip)
+ {
+ memcpy(norm_query + n_quer_loc, query + quer_loc, len_to_wrt);
+ n_quer_loc += len_to_wrt;
+
+ /* Skip the following until a non merged constant appear */
+ skip = true;
+ n_quer_loc += sprintf(norm_query + n_quer_loc, "...");
+ }
quer_loc = off + tok_len;
last_off = off;
diff --git a/contrib/pg_stat_statements/sql/pg_stat_statements.sql b/contrib/pg_stat_statements/sql/pg_stat_statements.sql
index 8f5c866225..8f9d284ed3 100644
--- a/contrib/pg_stat_statements/sql/pg_stat_statements.sql
+++ b/contrib/pg_stat_statements/sql/pg_stat_statements.sql
@@ -464,4 +464,111 @@ SELECT (
SELECT COUNT(*) FROM pg_stat_statements WHERE query LIKE '%SELECT GROUPING%';
+--
+-- Consts merging
+--
+CREATE TABLE test_merge (id int, data int);
+
+-- IN queries
+
+-- No merging
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- Normal
+SET const_merge_threshold = 5;
+
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- On the merge threshold
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- With gaps on the threshold
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- test constants after merge
+SELECT pg_stat_statements_reset();
+
+SELECT * FROM test_merge WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10) and data = 2;
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- On table, numeric type causes every constant being wrapped into functions.
+CREATE TABLE test_merge_numeric (id int, data numeric(5, 2));
+SELECT pg_stat_statements_reset();
+SELECT * FROM test_merge_numeric WHERE id IN (1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+SELECT query, calls FROM pg_stat_statements ORDER BY query COLLATE "C";
+
+-- Test find_const_walker
+WITH cte AS (
+ SELECT 'const' as const FROM test_merge
+)
+SELECT ARRAY['a', 'b', 'c', const::varchar] AS result
+FROM cte;
+
+RESET const_merge_threshold;
+
DROP EXTENSION pg_stat_statements;
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f985afc009..e4306cdb89 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -8278,6 +8278,32 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv;
</listitem>
</varlistentry>
+ <varlistentry id="guc-const-merge-threshold" xreflabel="const_merge_treshold">
+ <term><varname>const_merge_threshold</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>const_merge_threshold</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Specifies the minimal length of an array to be eligible for constants
+ collapsing. Normally every element of an array contributes to a query
+ identifier, which means the same query containing an array of constants
+ could get multiple different identifiers, depending of size of the
+ array. If this parameter is nonzero, the array contains only constants
+ and it's length is larger than <varname> const_merge_threshold </varname>,
+ then array elements will contribute nothing to the query identifier.
+ Thus the query will get the same identifier no matter how many constants
+ it contains.
+
+ Zero turns off collapsing, and it is the default value.
+
+ The <xref linkend="pgstatstatements"/> extension will represent such
+ collapsed constants via <literal>'(...)'</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</sect2>
diff --git a/doc/src/sgml/pgstatstatements.sgml b/doc/src/sgml/pgstatstatements.sgml
index efc36da602..f7e2e9fe85 100644
--- a/doc/src/sgml/pgstatstatements.sgml
+++ b/doc/src/sgml/pgstatstatements.sgml
@@ -519,10 +519,30 @@
<para>
In some cases, queries with visibly different texts might get merged into a
single <structname>pg_stat_statements</structname> entry. Normally this will happen
- only for semantically equivalent queries, but there is a small chance of
- hash collisions causing unrelated queries to be merged into one entry.
- (This cannot happen for queries belonging to different users or databases,
- however.)
+ only for semantically equivalent queries, for example when queries are
+ different only in values of constants they use. Another valid possibility for
+ merging queries into a single <structname>pg_stat_statements</structname>
+ entry is when <xref linkend="guc-const-merge-threshold"/> is nonzero and the
+ queries contain an array with more than <varname>const_merge_threshold</varname>
+ constants in it:
+
+<screen>
+=# SET const_merge_threshold = 5;
+=# SELECT pg_stat_statements_reset();
+=# SELECT * FROM test WHERE a IN (1, 2, 3, 4, 5, 6, 7, 8, 9);
+=# SELECT * FROM test WHERE a IN (1, 2, 3, 4, 5, 6, 7);
+=# SELECT query, calls FROM pg_stat_statements;
+-[ RECORD 1 ]------------------------------
+query | SELECT * FROM test WHERE a IN (...)
+calls | 2
+-[ RECORD 2 ]------------------------------
+query | SELECT pg_stat_statements_reset()
+calls | 1
+</screen>
+
+ But there is a small chance of hash collisions causing unrelated queries to
+ be merged into one entry. (This cannot happen for queries belonging to
+ different users or databases, however.)
</para>
<para>
diff --git a/src/backend/nodes/queryjumblefuncs.c b/src/backend/nodes/queryjumblefuncs.c
index 16084842a3..1ea1cc66f8 100644
--- a/src/backend/nodes/queryjumblefuncs.c
+++ b/src/backend/nodes/queryjumblefuncs.c
@@ -42,6 +42,9 @@
/* GUC parameters */
int compute_query_id = COMPUTE_QUERY_ID_AUTO;
+/* Minimal numer of constants in an array after which they will be merged */
+int const_merge_threshold = 0;
+
/* True when compute_query_id is ON, or AUTO and a module requests them */
bool query_id_enabled = false;
@@ -53,7 +56,8 @@ static void JumbleQueryInternal(JumbleState *jstate, Query *query);
static void JumbleRangeTable(JumbleState *jstate, List *rtable);
static void JumbleRowMarks(JumbleState *jstate, List *rowMarks);
static void JumbleExpr(JumbleState *jstate, Node *node);
-static void RecordConstLocation(JumbleState *jstate, int location);
+static void JumbleExprList(JumbleState *jstate, List *node);
+static void RecordConstLocation(JumbleState *jstate, int location, bool merged);
/*
* Given a possibly multi-statement source string, confine our attention to the
@@ -120,7 +124,7 @@ JumbleQuery(Query *query, const char *querytext)
jstate->jumble_len = 0;
jstate->clocations_buf_size = 32;
jstate->clocations = (LocationLen *)
- palloc(jstate->clocations_buf_size * sizeof(LocationLen));
+ palloc0(jstate->clocations_buf_size * sizeof(LocationLen));
jstate->clocations_count = 0;
jstate->highest_extern_param_id = 0;
@@ -343,6 +347,90 @@ JumbleRowMarks(JumbleState *jstate, List *rowMarks)
}
}
+/*
+ * Jubmle a list of expressions
+ *
+ * This function enforces const_merge_threshold limitation, i.e. if the
+ * provided list contains only constant expressions and its length is greater
+ * than or equal to const_merge_threshold, such list will not contribute to
+ * jumble. Otherwise it falls back to JumbleExpr.
+ */
+static void
+JumbleExprList(JumbleState *jstate, List *elements)
+{
+ ListCell *temp;
+ Node *firstExpr = NULL;
+ bool allConst = true;
+
+ if (elements == NULL)
+ return;
+
+ if (const_merge_threshold == 0)
+ {
+ /* Merging is disabled, process everything one by one. */
+ JumbleExpr(jstate, (Node *) elements);
+ return;
+ }
+
+ if (elements->length < const_merge_threshold)
+ {
+ /* The list is not large enough to collapse it. */
+ JumbleExpr(jstate, (Node *) elements);
+ return;
+ }
+
+ /* Guard against stack overflow due to overly complex expressions */
+ check_stack_depth();
+
+ firstExpr = linitial(elements);
+
+ /*
+ * We always emit the node's NodeTag, then any additional fields that are
+ * considered significant, and then we recurse to any child nodes.
+ */
+ APP_JUMB(elements->type);
+
+ /*
+ * If the first expression is a constant, verify if the following elements
+ * are constants as well. If yes, the list is eligible for collapsing --
+ * mark it as merged and return from the function.
+ */
+ if (IsA(firstExpr, Const))
+ {
+ foreach(temp, elements)
+ {
+ Node *expr = (Node *) lfirst(temp);
+
+ if (!IsA(expr, Const))
+ {
+ allConst = false;
+ break;
+ }
+ }
+
+ if (allConst)
+ {
+ Const *firstConst = (Const *) firstExpr;
+ Const *lastConst = llast_node(Const, elements);
+
+ /*
+ * First and last constants are needed to identify which part of
+ * the query to skip in generate_normalized_query.
+ */
+ RecordConstLocation(jstate, firstConst->location, true);
+ RecordConstLocation(jstate, lastConst->location, true);
+ return;
+ }
+ }
+
+ /*
+ * If we end up here, it means no constants merging is possible, process
+ * the list as usual.
+ */
+ JumbleExpr(jstate, (Node *) elements);
+ return;
+}
+
/*
* Jumble an expression tree
*
@@ -392,7 +480,7 @@ JumbleExpr(JumbleState *jstate, Node *node)
/* We jumble only the constant's type, not its value */
APP_JUMB(c->consttype);
/* Also, record its parse location for query normalization */
- RecordConstLocation(jstate, c->location);
+ RecordConstLocation(jstate, c->location, false);
}
break;
case T_Param:
@@ -581,7 +669,7 @@ JumbleExpr(JumbleState *jstate, Node *node)
}
break;
case T_ArrayExpr:
- JumbleExpr(jstate, (Node *) ((ArrayExpr *) node)->elements);
+ JumbleExprList(jstate, (List *) ((ArrayExpr *) node)->elements);
break;
case T_RowExpr:
JumbleExpr(jstate, (Node *) ((RowExpr *) node)->args);
@@ -835,11 +923,13 @@ JumbleExpr(JumbleState *jstate, Node *node)
}
/*
- * Record location of constant within query string of query tree
- * that is currently being walked.
+ * Record location of constant within query string of query tree that is
+ * currently being walked. Merged argument signals that the constant do not
+ * contribute to the jumble hash, and any reader of constants array may want to
+ * use this information to represent such constants differently.
*/
static void
-RecordConstLocation(JumbleState *jstate, int location)
+RecordConstLocation(JumbleState *jstate, int location, bool merged)
{
/* -1 indicates unknown or undefined location */
if (location >= 0)
@@ -854,6 +944,7 @@ RecordConstLocation(JumbleState *jstate, int location)
sizeof(LocationLen));
}
jstate->clocations[jstate->clocations_count].location = location;
+ jstate->clocations[jstate->clocations_count].merged = merged;
/* initialize lengths to -1 to simplify third-party module usage */
jstate->clocations[jstate->clocations_count].length = -1;
jstate->clocations_count++;
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 4ac808ed22..663aded290 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -3467,6 +3467,19 @@ struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+ {
+ {"const_merge_threshold", PGC_SUSET, STATS_MONITORING,
+ gettext_noop("Sets the minimal numer of constants in an array"
+ " after which they will be merged"),
+ gettext_noop("Computing query id for an array of constants"
+ " will produce the same id for all arrays with length"
+ " larger than this value. Zero turns off merging."),
+ },
+ &const_merge_threshold,
+ 0, 0, INT_MAX,
+ NULL, NULL, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index d06074b86f..0594eb17b2 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -627,7 +627,7 @@
#log_parser_stats = off
#log_planner_stats = off
#log_executor_stats = off
-
+#const_merge_threshold = 0
#------------------------------------------------------------------------------
# AUTOVACUUM
diff --git a/src/include/nodes/queryjumble.h b/src/include/nodes/queryjumble.h
index 204b8f74fd..4410e2cf61 100644
--- a/src/include/nodes/queryjumble.h
+++ b/src/include/nodes/queryjumble.h
@@ -15,6 +15,7 @@
#define QUERYJUBLE_H
#include "nodes/parsenodes.h"
+#include "nodes/nodeFuncs.h"
/*
* Struct for tracking locations/lengths of constants during normalization
@@ -23,6 +24,8 @@ typedef struct LocationLen
{
int location; /* start offset in query text */
int length; /* length in bytes, or -1 to ignore */
+ bool merged; /* whether or not the location was marked as
+ not contributing to jumble */
} LocationLen;
/*
@@ -61,7 +64,7 @@ enum ComputeQueryIdType
/* GUC parameters */
extern PGDLLIMPORT int compute_query_id;
-
+extern PGDLLIMPORT int const_merge_threshold;
extern const char *CleanQuerytext(const char *query, int *location, int *len);
extern JumbleState *JumbleQuery(Query *query, const char *querytext);
--
2.32.0
--sx5skzdffeabe3wq--
^ permalink raw reply [nested|flat] 5+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-05-17 03:06 Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 2 replies; 5+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-05-17 03:06 UTC (permalink / raw)
To: 'Peter Smith' <[email protected]>; +Cc: [email protected] <[email protected]>; Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>
Dear Peter,
Thanks for reviewing! Here are new patches.
> ======
> doc/src/sgml/ref/alter_subscription.sgml
>
> 2.1.
> <command>ALTER SUBSCRIPTION ... SET (failover = on|off)</command>
> and
> - <command>ALTER SUBSCRIPTION ... SET (two_phase =
> on|off)</command>
> + <command>ALTER SUBSCRIPTION ... SET (two_phase = off)</command>
>
> My other thread patch has already been pushed [1], so now you can
> modify this to say "true|false" as previously suggested.
Fixed accordingly.
> //////////
> v11-0003
> //////////
>
> ======
> src/backend/commands/subscriptioncmds.c
>
> 3.1. AlterSubscription
>
> + subtwophase = LOGICALREP_TWOPHASE_STATE_DISABLED;
> + }
> + else
> + subtwophase = LOGICALREP_TWOPHASE_STATE_PENDING;
> +
> +
> /* Change system catalog acoordingly */
> values[Anum_pg_subscription_subtwophasestate - 1] =
> - CharGetDatum(opts.twophase ?
> - LOGICALREP_TWOPHASE_STATE_PENDING :
> - LOGICALREP_TWOPHASE_STATE_DISABLED);
> + CharGetDatum(subtwophase);
> replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
>
> Sorry, I don't think that 'subtwophase' change is an improvement --
> IMO the existing ternary code was fine as-is.
>
> I only reported the excessive flag checking in the previous v10-0003
> review because of some extra "if (!opts.twophase)" code but that was
> caused by what you called "wrong git operations." and is already fixed
> now.
Ok, the part was reverted.
> //////////
> v11-0004
> //////////
>
> ======
> src/sgml/catalogs.sgml
>
> 4.1.
> + <row>
> + <entry role="catalog_table_entry"><para role="column_definition">
> + <structfield>subforcealter</structfield> <type>bool</type>
> + </para>
> + <para>
> + If true, then the <link
> linkend="sql-altersubscription"><command>ALTER
> SUBSCRIPTION</command></link>
> + can disable <literal>two_phase</literal> option, even if there are
> + uncommitted prepared transactions from when
> <literal>two_phase</literal>
> + was enabled
> + </para></entry>
> + </row>
> +
>
> I think this description should be changed to say what it *really*
> does. IMO, the stuff about 'two_phase' is more like a side-effect.
>
> SUGGESTION (this is just pseudo-code. You can add the CREATE
> SUBSCRIPTION 'force_alter' link appropriately)
>
> If true, then the <command>ALTER SUBSCRIPTION</command> command can
> sometimes be forced to proceed instead of giving an error. See
> <link>force_alter</link> parameter for details about when this might
> be useful.
>
Fixed. But One point, the word "command" was removed. I checked other parts and
it seemed not to be needed.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v12-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (25.5K, ../../OSBPR01MB255262AFDADA5126451BD23CF5EE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v12-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch)
download | inline diff:
From 5351d720e91921738cbc37a0dc01c1b75d183071 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v12 1/4] Allow altering of two_phase option of a SUBSCRIPTION
This patch allows the user to alter the 'two_phase' option of a subscriber provided no
uncommitted prepared transactions are pending on that subscription.
Author: Cherian Ajin, Hayato Kuroda
---
doc/src/sgml/ref/alter_subscription.sgml | 12 ++--
src/backend/access/transam/twophase.c | 62 ++++++++++++++++
src/backend/commands/subscriptioncmds.c | 67 +++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 9 +--
src/backend/replication/logical/launcher.c | 22 ++++++
src/backend/replication/logical/worker.c | 21 +-----
src/backend/replication/slot.c | 18 ++++-
src/backend/replication/walsender.c | 18 ++++-
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 5 ++
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 11 +--
src/include/replication/worker_internal.h | 1 +
src/test/regress/expected/subscription.out | 5 +-
src/test/regress/sql/subscription.sql | 5 +-
src/test/subscription/t/021_twophase.pl | 71 ++++++++++++++++++-
16 files changed, 269 insertions(+), 63 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 476f195622..0b23df1b77 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -68,8 +68,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>,
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
- with <literal>refresh</literal> option as <literal>true</literal> and
- <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command>
+ with <literal>refresh</literal> option as <literal>true</literal>,
+ <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
@@ -228,9 +229,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
- <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled.
</para>
<para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 8090ac9fc1..66fa591eb5 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,65 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
LWLockRelease(TwoPhaseStateLock);
return found;
}
+
+/*
+ * TwoPhaseTransactionGid
+ * Form the prepared transaction GID for two_phase transactions.
+ *
+ * Return the GID in the supplied buffer.
+ */
+void
+TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
+{
+ Assert(subid != InvalidRepOriginId);
+
+ if (!TransactionIdIsValid(xid))
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("invalid two-phase transaction ID")));
+
+ snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
+}
+
+/*
+ * IsTwoPhaseTransactionGidForSubid
+ * Check whether the given GID (as formed by TwoPhaseTransactionGid) is
+ * for the specified 'subid'.
+ */
+static bool
+IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
+{
+ int ret;
+ Oid subid_written;
+ TransactionId xid;
+
+ ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
+
+ return (ret == 2 && subid == subid_written);
+}
+
+/*
+ * LookupGXactBySubid
+ * Check if the prepared transaction done by apply worker exists.
+ */
+bool
+LookupGXactBySubid(Oid subid)
+{
+ bool found = false;
+
+ LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+ for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
+ {
+ GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+ /* Ignore not-yet-valid GIDs. */
+ if (gxact->valid &&
+ IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
+ {
+ found = true;
+ break;
+ }
+ }
+ LWLockRelease(TwoPhaseStateLock);
+ return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e407428dbc..90d967eb7c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -16,6 +16,7 @@
#include "access/htup_details.h"
#include "access/table.h"
+#include "access/twophase.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
@@ -1143,7 +1144,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+ SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+ SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_ORIGIN);
@@ -1151,6 +1153,52 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * Do not allow changing the two_phase option if the
+ * subscription is enabled. This is because the two_phase
+ * option of the slot on the publisher cannot be modified
+ * if the slot is currently acquired by the apply worker.
+ */
+ if (form->subenabled)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for enabled subscription",
+ "two_phase")));
+
+ /*
+ * Stop all the subscription workers, just in case. Workers
+ * may still survive even if the subscription is disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /*
+ * two_phase cannot be disabled if there are any
+ * uncommitted prepared transactions present.
+ */
+ if (!opts.twophase &&
+ form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ LookupGXactBySubid(subid))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
+ errhint("Resolve these transactions and try again")));
+
+ /*
+ * The changed two_phase option of the slot can't be rolled
+ * back.
+ */
+ PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+
+ /* Change system catalog acoordingly */
+ values[Anum_pg_subscription_subtwophasestate - 1] =
+ CharGetDatum(opts.twophase ?
+ LOGICALREP_TWOPHASE_STATE_PENDING :
+ LOGICALREP_TWOPHASE_STATE_DISABLED);
+ replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
{
/*
@@ -1505,7 +1553,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1])
+ if (replaces[Anum_pg_subscription_subfailover - 1] ||
+ replaces[Anum_pg_subscription_subtwophasestate - 1])
{
bool must_use_password;
char *err;
@@ -1525,7 +1574,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
}
PG_FINALLY();
{
@@ -1562,7 +1611,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1720,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
* New workers won't be started because we hold an exclusive lock on the
* subscription till the end of the transaction.
*/
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- subworkers = logicalrep_workers_find(subid, false);
- LWLockRelease(LogicalRepWorkerLock);
- foreach(lc, subworkers)
- {
- LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
- logicalrep_worker_stop(w->subid, w->relid);
- }
- list_free(subworkers);
+ logicalrep_workers_stop(subid);
/*
* Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 3c2b1bb496..998bbd517a 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover);
+ bool failover, bool two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,15 +1121,16 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool failover, bool two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
quote_identifier(slotname),
- failover ? "true" : "false");
+ failover ? "true" : "false",
+ two_phase ? "true" : "false");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 66070e9131..548f6e0edb 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,28 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockRelease(LogicalRepWorkerLock);
}
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+ List *subworkers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ /* XXX clarify the reason why not only running workers are listed. */
+ subworkers = logicalrep_workers_find(subid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+ foreach(lc, subworkers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ logicalrep_worker_stop(w->subid, w->relid);
+ }
+ list_free(subworkers);
+}
+
/*
* Stop the given logical replication parallel apply worker.
*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..dcf656fd45 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -402,7 +402,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata,
CmdType operation);
/* Compute GID for two_phase transactions */
-static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid);
/* Functions for skipping changes */
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn);
@@ -3911,7 +3910,7 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
+ /* two-phase cannot be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
/*
@@ -4396,24 +4395,6 @@ cleanup_subxact_info()
subxact_data.nsubxacts_max = 0;
}
-/*
- * Form the prepared transaction GID for two_phase transactions.
- *
- * Return the GID in the supplied buffer.
- */
-static void
-TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
-{
- Assert(subid != InvalidRepOriginId);
-
- if (!TransactionIdIsValid(xid))
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("invalid two-phase transaction ID")));
-
- snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
-}
-
/*
* Common function to run the apply loop with error handling. Disable the
* subscription, if necessary.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index aa4ea387da..d0c8d5a4df 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -804,8 +804,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
{
+ bool update_slot = false;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover)
MyReplicationSlot->data.failover = failover;
SpinLockRelease(&MyReplicationSlot->mutex);
+ update_slot = true;
+ }
+
+ if (MyReplicationSlot->data.two_phase != two_phase)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.two_phase = two_phase;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ update_slot = true;
+ }
+
+ if (update_slot)
+ {
ReplicationSlotMarkDirty();
ReplicationSlotSave();
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c623b07cf0..2e6ca35049 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,9 +1411,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+ bool *failover, bool *two_phase)
{
bool failover_given = false;
+ bool two_phase_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
@@ -1427,6 +1429,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "two_phase") == 0)
+ {
+ if (two_phase_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ two_phase_given = true;
+ *two_phase = defGetBoolean(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1439,9 +1450,10 @@ static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
bool failover = false;
+ bool two_phase = false;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &failover, &two_phase);
+ ReplicationSlotAlter(cmd->slotname, failover, two_phase);
}
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index d453e224d9..891face1b6 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
"password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ "streaming", "synchronous_commit", "two_phase");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0006..d37e06fdee 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,9 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
+
+extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
+ int szgid);
+extern bool LookupGXactBySubid(Oid subid);
+
#endif /* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 1bc80960ef..014e216cf5 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool failover,
+ bool two_phase);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..31fa1257ec 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,12 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
/*
* walrcv_alter_slot_fn
*
- * Change the definition of a replication slot. Currently, it only supports
- * changing the failover property of the slot.
+ * Change the definition of a replication slot. Currently, it supports
+ * changing the failover and the two_phase property of the slot.
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover);
+ bool failover,
+ bool two_phase);
/*
* walrcv_get_backend_pid_fn
@@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
- WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_alter_slot(conn, slotname, failover, two_phase) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 515aefd519..d5428263c1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid userid, Oid relid,
dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_workers_stop(Oid subid);
extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 0f2a25cdc1..51fa4b9690 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -377,10 +377,7 @@ HINT: To initiate replication, you must manually create the replication slot, e
regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-ERROR: unrecognized subscription parameter: "two_phase"
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
List of subscriptions
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 3e5ba4cb8c..a3886d79ca 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,10 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
\dRs+
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 9437cd4c3b..72df258000 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,75 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
is($result, qq(2), 'replicated data in subscriber table');
+# Clean up
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+
+###############################
+# Disable the subscription and alter it to two_phase = false,
+# then verify that the altered subscription reflects the two_phase option.
+###############################
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is disabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(d), 'two-phase should be disabled');
+
+# Now do a prepare on the publisher and make sure that it is not replicated.
+$node_publisher->safe_psql(
+ 'postgres', qq{
+ BEGIN;
+ INSERT INTO tab_copy VALUES (100);
+ PREPARE TRANSACTION 'newgid';
+ });
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure there are no prepared transactions on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'should be no prepared transactions on subscriber');
+
+# Now commit the insert and verify that it IS replicated
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure that the committed transaction is replicated.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
+is($result, qq(3), 'replicated data in subscriber table');
+
+# Alter subscription two_phase to true
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is enabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(e), 'two-phase should be enabled');
+
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
@@ -374,8 +443,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
# check all the cleanup
###############################
-$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
-
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription");
is($result, qq(0), 'check subscription was dropped on subscriber');
--
2.43.0
[application/octet-stream] v12-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (12.3K, ../../OSBPR01MB255262AFDADA5126451BD23CF5EE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v12-0002-Alter-slot-option-two_phase-only-when-altering-t.patch)
download | inline diff:
From b9eeaee48f2026a31e50a42c0ee869c065c3639d Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v12 2/4] Alter slot option two_phase only when altering "true"
to "false"
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Since the two_phase option is controlled by both the publisher (as a slot option)
and the subscriber (as a subscription option), the slot option must also be
modified.
Regarding the false->true case, the backend process alters the subtwophase to
LOGICALREP_TWOPHASE_STATE_PENDING once. After the subscription is enabled, a new
logical replication worker requests to change the two_phase option of its slot
from pending to true after the initial data synchronization is done. The code
path is the same as the case in which two_phase is initially set to true, so
there is no need to do something remarkable. However, for the true->false case,
the backend must connect to the publisher and expressly change the parameter
because the apply worker does not alter the option to false. Because this
operation cannot be rolled back, altering the two_phase parameter from "true"
to "false" within a transaction is prohibited.
---
doc/src/sgml/ref/alter_subscription.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 42 ++++++--
.../libpqwalreceiver/libpqwalreceiver.c | 23 +++--
src/include/replication/walreceiver.h | 5 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 96 +++++++++++++++++++
6 files changed, 153 insertions(+), 16 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0b23df1b77..475a42a2e3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -70,7 +70,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
with <literal>refresh</literal> option as <literal>true</literal>,
<command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
- <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = off)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 90d967eb7c..ff733e3810 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1097,6 +1097,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
Form_pg_subscription form;
bits32 supported_opts;
SubOpts opts = {0};
+ bool update_failover;
+ bool update_two_phase;
rel = table_open(SubscriptionRelationId, RowExclusiveLock);
@@ -1186,10 +1188,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errhint("Resolve these transactions and try again")));
/*
- * The changed two_phase option of the slot can't be rolled
- * back.
+ * Altering the parameter from "true" to "false" within a
+ * transaction is prohibited. Since the apply worker does
+ * not alter the slot option to false, the backend must
+ * connect to the publisher and expressly change the
+ * parameter.
+ *
+ * There is no need to do something remarkable regarding
+ * the "false" to "true" case; the backend process alters
+ * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
+ * After the subscription is enabled, a new logical
+ * replication worker requests to change the two_phase
+ * option of its slot from pending to true when the initial
+ * data synchronization is done. The code path is the same
+ * as the case in which two_phase is initially set to true.
*/
- PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+ if (!opts.twophase)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = false)");
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -1547,14 +1563,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
}
/*
- * Try to acquire the connection necessary for altering slot.
+ * Check the need to alter the replication slot. Failover and two_phase
+ * options are controlled by both the publisher (as a slot option) and the
+ * subscriber (as a subscription option). The slot option must be altered
+ * only when changing "true" to "false". The reason has already been
+ * described in the ALTER_SUBSCRIPTION_OPTIONS section of this function.
+ */
+ update_failover = replaces[Anum_pg_subscription_subfailover - 1];
+ update_two_phase = (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+
+ /*
+ * Try to acquire the connection necessary for altering slot, if needed.
*
* This has to be at the end because otherwise if there is an error while
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1] ||
- replaces[Anum_pg_subscription_subtwophasestate - 1])
+ if (update_failover || update_two_phase)
{
bool must_use_password;
char *err;
@@ -1574,7 +1600,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
+ walrcv_alter_slot(wrconn, sub->slotname,
+ update_failover ? &opts.failover : NULL,
+ update_two_phase ? &opts.twophase : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 998bbd517a..ff013aa987 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase);
+ const bool *failover, const bool *two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,16 +1121,27 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase)
+ const bool *failover, const bool *two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
- quote_identifier(slotname),
- failover ? "true" : "false",
- two_phase ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+ quote_identifier(slotname));
+
+ if (failover)
+ appendStringInfo(&cmd, "FAILOVER %s",
+ *failover ? "true" : "false");
+
+ if (failover && two_phase)
+ appendStringInfo(&cmd, ", ");
+
+ if (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s",
+ *two_phase ? "true" : "false");
+
+ appendStringInfoString(&cmd, " );");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 31fa1257ec..7ffa5a58b3 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,8 +377,9 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover,
- bool two_phase);
+ const bool *failover,
+ const bool *two_phase);
+
/*
* walrcv_get_backend_pid_fn
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..b4bd522c3d 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
't/031_column_list.pl',
't/032_subscribe_use_index.pl',
't/033_run_as_table_owner.pl',
+ 't/099_twophase_added.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
new file mode 100644
index 0000000000..ac08969b32
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,96 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Additional tests for altering two_phase option
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10
+ log_min_messages = debug1));
+$node_subscriber->start;
+
+# Define tables on both nodes
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication, with two_phase = "false"
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION pub FOR ALL TABLES");
+
+my $log_offset = -s $node_subscriber->logfile;
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION regress_sub
+ CONNECTION '$publisher_connstr' PUBLICATION pub
+ WITH (two_phase = false, copy_data = false, failover = false)");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED', $log_offset);
+
+#####################
+# Check the case that prepared transactions exist on the publisher node.
+#
+# Since the two_phase is "false", then normally, this PREPARE will do nothing
+# until the COMMIT PREPARED, but in this test, we toggle the two_phase to
+# "true" again before the COMMIT PREPARED happens.
+
+# Prepare a transaction to insert some tuples into the table
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(1, 5));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the prepared transaction is not yet replicated to the subscriber
+# because two_phase is set to "false".
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Toggle the two_phase to "true" *before* the COMMIT PREPARED. Since we are the
+# special path for the case where both two_phase and failover are altered, it
+# is also set to "true".
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = true, failover = true);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was enabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is ENABLED', $log_offset);
+
+# And do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM tab_full;");
+is($result, q(5),
+ "prepared transactions done before altering can be replicated");
+
+done_testing();
--
2.43.0
[application/octet-stream] v12-0003-Abort-prepared-transactions-while-altering-two_p.patch (9.6K, ../../OSBPR01MB255262AFDADA5126451BD23CF5EE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v12-0003-Abort-prepared-transactions-while-altering-two_p.patch)
download | inline diff:
From 9cbcac569f7eb28b15d9afc5645cfb97f9cd6164 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v12 3/4] Abort prepared transactions while altering two_phase
to off
If we alter the two_phase parameter from "on" to "off" and there are prepared
transactions on the subscriber, they won't be resolved. To avoid this issue, we
allow the backend to abort all prepared transactions while altering the
subscription.
---
doc/src/sgml/ref/alter_subscription.sgml | 11 +++-
src/backend/access/transam/twophase.c | 17 ++---
src/backend/commands/subscriptioncmds.c | 65 +++++++++++--------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 45 +++++++++++++
5 files changed, 102 insertions(+), 39 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 475a42a2e3..8801f37f0e 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -233,8 +233,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
- The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled.
</para>
<para>
@@ -256,6 +254,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
option is enabled.
</para>
+
+ <para>
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled. When altering the parameter from <literal>true</literal>
+ to <literal>false</literal>, the backend process checks for any incomplete
+ prepared transactions done by the logical replication worker (from when
+ <literal>two_phase</literal> parameter was still <literal>true</literal>)
+ and, if any are found, those are aborted.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 66fa591eb5..f384bd8c0a 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2720,13 +2720,13 @@ IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
}
/*
- * LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ * Get a list of GIDs which is PREPARE'd by the given subscription.
*/
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
{
- bool found = false;
+ List *list = NIL;
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2736,11 +2736,8 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid &&
IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
- {
- found = true;
- break;
- }
+ list = lappend(list, pstrdup(gxact->gid));
}
LWLockRelease(TwoPhaseStateLock);
- return found;
+ return list;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index ff733e3810..1c16d52673 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1176,37 +1176,50 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
logicalrep_workers_stop(subid);
/*
- * two_phase cannot be disabled if there are any
- * uncommitted prepared transactions present.
- */
- if (!opts.twophase &&
- form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
- errhint("Resolve these transactions and try again")));
-
- /*
- * Altering the parameter from "true" to "false" within a
- * transaction is prohibited. Since the apply worker does
- * not alter the slot option to false, the backend must
- * connect to the publisher and expressly change the
- * parameter.
- *
- * There is no need to do something remarkable regarding
- * the "false" to "true" case; the backend process alters
- * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
- * After the subscription is enabled, a new logical
- * replication worker requests to change the two_phase
- * option of its slot from pending to true when the initial
- * data synchronization is done. The code path is the same
- * as the case in which two_phase is initially set to true.
+ * If two_phase was previously enabled, there is a
+ * possibility that transactions have already been
+ * PREPARE'd. They must be checked and rolled back.
*/
if (!opts.twophase)
+ {
+ List *prepared_xacts;
+
+ /*
+ * Altering the parameter from "true" to "false" within
+ * a transaction is prohibited. Since the apply worker
+ * does not alter the slot option to false, the backend
+ * must connect to the publisher and expressly change
+ * the parameter.
+ *
+ * There is no need to do something remarkable
+ * regarding the "false" to "true" case; the backend
+ * process alters subtwophase to
+ * LOGICALREP_TWOPHASE_STATE_PENDING once. After the
+ * subscription is enabled, a new logical replication
+ * worker requests to change the two_phase option of
+ * its slot from pending to true when the initial data
+ * synchronization is done. The code path is the same
+ * as the case in which two_phase is initially set to
+ * true.
+ */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = false)");
+ /*
+ * To prevent prepared transactions from being
+ * isolated, they must manually be aborted.
+ */
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ /* Abort all listed transactions */
+ foreach_ptr(char, gid, prepared_xacts)
+ FinishPreparedTransaction(gid, false);
+
+ list_free_deep(prepared_xacts);
+ }
+ }
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d37e06fdee..f7a5cf0c12 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "nodes/pg_list.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
@@ -65,6 +66,6 @@ extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
int szgid);
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
#endif /* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index ac08969b32..5a9a6c6476 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -93,4 +93,49 @@ $result = $node_subscriber->safe_psql('postgres',
is($result, q(5),
"prepared transactions done before altering can be replicated");
+#####################
+# Check the case that prepared transactions exist on the subscriber node
+#
+# If the two_phase is altering from "true" to "false" and there are prepared
+# transactions on the subscriber, they must be aborted. This test checks it.
+
+# Prepare a transaction to insert some tuples into the table
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(6, 10));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the prepared transaction has been replicated to the subscriber because
+# two_phase is set to "true".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+# Toggle the two_phase to "false" before the COMMIT PREPARED
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify any prepared transactions are aborted because two_phase is changed to
+# "false".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+# Do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql( 'postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(10) FROM tab_full;");
+is($result, q(10),
+ "prepared transactions on publisher can be replicated");
+
done_testing();
--
2.43.0
[application/octet-stream] v12-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch (56.9K, ../../OSBPR01MB255262AFDADA5126451BD23CF5EE2@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v12-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch)
download | inline diff:
From f2f90480f0c72a83c421e847cce9524576b71a3c Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v12 4/4] Add force_alter option for ALTER SUBSCRIPTION ... SET
command
Previously, all prepared transactions on the standby were rolled back when
toggling two_phase from "true" to "false". However, this operation may not be
expected by users. To ensure users understand what happens, we added the
"force_alter" parameter. When two_phase is toggling to "false", and there are
prepared transactions, they will be aborted only when "force_alter" is set to
true. Otherwise, an ERROR occurs.
---
doc/src/sgml/catalogs.sgml | 12 ++
doc/src/sgml/ref/alter_subscription.sgml | 16 +-
doc/src/sgml/ref/create_subscription.sgml | 24 +++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 2 +-
src/backend/commands/subscriptioncmds.c | 36 ++++-
src/bin/pg_dump/pg_dump.c | 14 ++
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 7 +-
src/include/catalog/pg_subscription.h | 13 ++
src/test/regress/expected/subscription.out | 152 +++++++++---------
src/test/subscription/t/099_twophase_added.pl | 39 ++++-
12 files changed, 222 insertions(+), 95 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 15f6255d86..bab3dcbce4 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8038,6 +8038,18 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subforcealter</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, then the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can sometimes be forced to proceed instead of giving an error. See
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ parameter for details about when this might be useful.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 8801f37f0e..ab2cdaeaa3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -230,8 +230,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
<link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
- <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
Only a superuser can set <literal>password_required = false</literal>.
</para>
@@ -257,11 +258,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled. When altering the parameter from <literal>true</literal>
- to <literal>false</literal>, the backend process checks for any incomplete
- prepared transactions done by the logical replication worker (from when
- <literal>two_phase</literal> parameter was still <literal>true</literal>)
- and, if any are found, those are aborted.
+ subscription is disabled. Altering the parameter from <literal>true</literal>
+ to <literal>false</literal> will give an error when there are prepared
+ transactions done by the logical replication worker. If you want to alter
+ the parameter forcibly in this case,
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ option must be set to <literal>true</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 740b7d9421..83ac52f865 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -428,6 +428,30 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-force-alter">
+ <term><literal>force_alter</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies if the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can be forced to proceed instead of giving an error.
+ </para>
+ <para>
+ There is currently only one scenario where this parameter has any
+ effect: When altering <literal>two_phase</literal> option from
+ <literal>true</literal> to <literal>false</literal> it is possible
+ for there to be incomplete prepared transactions done by the logical
+ replication worker (from when <literal>two_phase</literal> parameter
+ was still <literal>true</literal>). If <literal>force_alter</literal>
+ is <literal>false</literal>, then this will give an error; if
+ <literal>force_alter</literal> is <literal>true</literal>, then the
+ incomplete prepared transactions are aborted and the alter will proceed.
+ </para>
+ <para>
+ The default is <literal>false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..b568fe3470 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -72,6 +72,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
+ sub->forcealter = subform->subforcealter;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 53047cab5f..de3d3d8f3e 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1358,7 +1358,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subpasswordrequired, subrunasowner, subfailover,
+ subpasswordrequired, subrunasowner, subfailover, subforcealter,
subslotname, subsynccommit, subpublications, suborigin)
ON pg_subscription TO public;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 1c16d52673..82177c8a49 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,6 +73,7 @@
#define SUBOPT_FAILOVER 0x00002000
#define SUBOPT_LSN 0x00004000
#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_FORCE_ALTER 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -100,6 +101,7 @@ typedef struct SubOpts
bool failover;
char *origin;
XLogRecPtr lsn;
+ bool force_alter;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -162,6 +164,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->failover = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_FORCE_ALTER))
+ opts->force_alter = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -367,6 +371,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_FORCE_ALTER) &&
+ strcmp(defel->defname, "force_alter") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_FORCE_ALTER))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_FORCE_ALTER;
+ opts->force_alter = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -604,7 +617,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN |
+ SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -711,6 +725,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+ values[Anum_pg_subscription_subforcealter - 1] = BoolGetDatum(opts.force_alter);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1150,7 +1165,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1212,6 +1227,23 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
(prepared_xacts = GetGidListBySubid(subid)) != NIL)
{
+ bool raise_error =
+ IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) ?
+ !opts.force_alter : !sub->forcealter;
+
+ /*
+ * Abort prepared transactions only if
+ * 'force_alter' option is true. Otherwise raise
+ * an ERROR.
+ */
+ if (raise_error)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s when there are prepared transactions",
+ "two_phase = false"),
+ errhint("Resolve these transactions or set %s, and then try again.",
+ "force_alter = true")));
+
/* Abort all listed transactions */
foreach_ptr(char, gid, prepared_xacts)
FinishPreparedTransaction(gid, false);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index cb14fcafea..16a6a225ae 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4739,6 +4739,7 @@ getSubscriptions(Archive *fout)
int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
+ int i_subforcealter;
int i,
ntups;
@@ -4816,6 +4817,13 @@ getSubscriptions(Archive *fout)
appendPQExpBuffer(query,
" false AS subfailover\n");
+ if (fout->remoteVersion >= 170000)
+ appendPQExpBufferStr(query,
+ " s.subforcealter\n");
+ else
+ appendPQExpBuffer(query,
+ " false AS subforcealter\n");
+
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4854,6 +4862,7 @@ getSubscriptions(Archive *fout)
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
i_subenabled = PQfnumber(res, "subenabled");
i_subfailover = PQfnumber(res, "subfailover");
+ i_subforcealter = PQfnumber(res, "subforcealter");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4900,6 +4909,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subenabled));
subinfo[i].subfailover =
pg_strdup(PQgetvalue(res, i, i_subfailover));
+ subinfo[i].subforcealter =
+ pg_strdup(PQgetvalue(res, i, i_subforcealter));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5140,6 +5151,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subfailover, "t") == 0)
appendPQExpBufferStr(query, ", failover = true");
+ if (strcmp(subinfo->subforcealter, "t") == 0)
+ appendPQExpBufferStr(query, ", force_alter = true");
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 865823868f..bb0fda3b29 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -671,6 +671,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *suboriginremotelsn;
char *subfailover;
+ char *subforcealter;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index f67bf0b892..dba229a25d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6529,7 +6529,7 @@ describeSubscriptions(const char *pattern, bool verbose)
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false, false, false, false, false, false,
- false};
+ false, false};
if (pset.sversion < 100000)
{
@@ -6598,6 +6598,11 @@ describeSubscriptions(const char *pattern, bool verbose)
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subforcealter AS \"%s\"\n",
+ gettext_noop("Force alter"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..c23de43d79 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -98,6 +98,13 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool subforcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true" to
+ * "false". */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -151,6 +158,12 @@ typedef struct Subscription
* (i.e. the main slot and the table sync
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool forcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true" to
+ * "false". */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9690..b36fc6b8f7 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,10 +393,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -409,18 +409,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index 5a9a6c6476..662ecf308b 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -114,15 +114,38 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-# Toggle the two_phase to "false" before the COMMIT PREPARED
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION regress_sub DISABLE;
- ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
- ALTER SUBSCRIPTION regress_sub ENABLE;");
+# Disable the subscription to alter the two_phase option
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub DISABLE;");
+
+# Try altering the two_phase option to "false". The command will fail since
+# there is a prepared transaction and the 'force_alter' option is not specified
+# as true.
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql(
+ 'postgres', "ALTER SUBSCRIPTION regress_sub SET (two_phase = false);");
+ok($stderr =~ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+# Verify the prepared transaction still exists
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exists");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Alter the two_phase true to false with the force_alter option enabled. This
+# command will succeed after aborting the prepared transaction.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub SET (two_phase = false, force_alter = true);");
+$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED', $log_offset);
-# Verify any prepared transactions are aborted because two_phase is changed to
-# "false".
+# # Verify the prepared transaction was aborted
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
--
2.43.0
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-05-17 04:08 Peter Smith <[email protected]>
parent: Hayato Kuroda (Fujitsu) <[email protected]>
1 sibling, 0 replies; 5+ messages in thread
From: Peter Smith @ 2024-05-17 04:08 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: [email protected] <[email protected]>; Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>
Hi Kuroda-san,
I did not apply these v12* patches, but I have diff'ed all of them
with the previous v11* patches and confirmed that all of my previous
review comments now seem to be addressed.
I don't have any more comments to make at this time. Thanks!
======
Kind Regards,
Peter Smith.
Fujitsu Australia
^ permalink raw reply [nested|flat] 5+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-06-25 08:06 Hayato Kuroda (Fujitsu) <[email protected]>
parent: Hayato Kuroda (Fujitsu) <[email protected]>
1 sibling, 1 reply; 5+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-06-25 08:06 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; 'Peter Smith' <[email protected]>; +Cc: [email protected] <[email protected]>; Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>
Dear hackers,
I found that v12 patch set could not be accepted by the cfbot. PSA new version.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
Attachments:
[application/octet-stream] v13-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch (57.1K, ../../OSBPR01MB25528ABFBE8B7DB1C4718BC6F5D52@OSBPR01MB2552.jpnprd01.prod.outlook.com/2-v13-0004-Add-force_alter-option-for-ALTER-SUBSCRIPTION-.-.patch)
download | inline diff:
From 40deb2d49d46b9682e6805fca5c27d119acf6c3e Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 19 Apr 2024 11:03:19 +0000
Subject: [PATCH v13 4/4] Add force_alter option for ALTER SUBSCRIPTION ... SET
command
Previously, all prepared transactions on the standby were rolled back when
toggling two_phase from "true" to "false". However, this operation may not be
expected by users. To ensure users understand what happens, we added the
"force_alter" parameter. When two_phase is toggling to "false", and there are
prepared transactions, they will be aborted only when "force_alter" is set to
true. Otherwise, an ERROR occurs.
---
doc/src/sgml/catalogs.sgml | 12 ++
doc/src/sgml/ref/alter_subscription.sgml | 16 +-
doc/src/sgml/ref/create_subscription.sgml | 24 +++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 2 +-
src/backend/commands/subscriptioncmds.c | 36 ++++-
src/bin/pg_dump/pg_dump.c | 18 ++-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 7 +-
src/include/catalog/pg_subscription.h | 13 ++
src/test/regress/expected/subscription.out | 152 +++++++++---------
src/test/subscription/t/099_twophase_added.pl | 44 ++++-
12 files changed, 229 insertions(+), 97 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index a63cc71efa..89b4fd07f1 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8034,6 +8034,18 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subforcealter</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, then the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can sometimes be forced to proceed instead of giving an error. See
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ parameter for details about when this might be useful.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 8801f37f0e..ab2cdaeaa3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -230,8 +230,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
<link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
- <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
Only a superuser can set <literal>password_required = false</literal>.
</para>
@@ -257,11 +258,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled. When altering the parameter from <literal>true</literal>
- to <literal>false</literal>, the backend process checks for any incomplete
- prepared transactions done by the logical replication worker (from when
- <literal>two_phase</literal> parameter was still <literal>true</literal>)
- and, if any are found, those are aborted.
+ subscription is disabled. Altering the parameter from <literal>true</literal>
+ to <literal>false</literal> will give an error when there are prepared
+ transactions done by the logical replication worker. If you want to alter
+ the parameter forcibly in this case,
+ <link linkend="sql-createsubscription-params-with-force-alter"><literal>force_alter</literal></link>
+ option must be set to <literal>true</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 740b7d9421..83ac52f865 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -428,6 +428,30 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-force-alter">
+ <term><literal>force_alter</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies if the <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION</command></link>
+ can be forced to proceed instead of giving an error.
+ </para>
+ <para>
+ There is currently only one scenario where this parameter has any
+ effect: When altering <literal>two_phase</literal> option from
+ <literal>true</literal> to <literal>false</literal> it is possible
+ for there to be incomplete prepared transactions done by the logical
+ replication worker (from when <literal>two_phase</literal> parameter
+ was still <literal>true</literal>). If <literal>force_alter</literal>
+ is <literal>false</literal>, then this will give an error; if
+ <literal>force_alter</literal> is <literal>true</literal>, then the
+ incomplete prepared transactions are aborted and the alter will proceed.
+ </para>
+ <para>
+ The default is <literal>false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..b568fe3470 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -72,6 +72,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
+ sub->forcealter = subform->subforcealter;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index efb29adeb3..3366da57fd 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1359,7 +1359,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subpasswordrequired, subrunasowner, subfailover,
+ subpasswordrequired, subrunasowner, subfailover, subforcealter,
subslotname, subsynccommit, subpublications, suborigin)
ON pg_subscription TO public;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 2532b40365..5ab5172a38 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,6 +73,7 @@
#define SUBOPT_FAILOVER 0x00002000
#define SUBOPT_LSN 0x00004000
#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_FORCE_ALTER 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -100,6 +101,7 @@ typedef struct SubOpts
bool failover;
char *origin;
XLogRecPtr lsn;
+ bool force_alter;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -162,6 +164,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->failover = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_FORCE_ALTER))
+ opts->force_alter = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -367,6 +371,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_FORCE_ALTER) &&
+ strcmp(defel->defname, "force_alter") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_FORCE_ALTER))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_FORCE_ALTER;
+ opts->force_alter = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -604,7 +617,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN |
+ SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -711,6 +725,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+ values[Anum_pg_subscription_subforcealter - 1] = BoolGetDatum(opts.force_alter);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1150,7 +1165,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_FORCE_ALTER);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1213,6 +1228,23 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
(prepared_xacts = GetGidListBySubid(subid)) != NIL)
{
+ bool raise_error =
+ IsSet(opts.specified_opts, SUBOPT_FORCE_ALTER) ?
+ !opts.force_alter : !sub->forcealter;
+
+ /*
+ * Abort prepared transactions only if
+ * 'force_alter' option is true. Otherwise raise
+ * an ERROR.
+ */
+ if (raise_error)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot alter %s when there are prepared transactions",
+ "two_phase = false"),
+ errhint("Resolve these transactions or set %s, and then try again.",
+ "force_alter = true")));
+
/* Abort all listed transactions */
foreach_ptr(char, gid, prepared_xacts)
FinishPreparedTransaction(gid, false);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index e324070828..80139ea2cd 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4739,6 +4739,7 @@ getSubscriptions(Archive *fout)
int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
+ int i_subforcealter;
int i,
ntups;
@@ -4811,10 +4812,17 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 170000)
appendPQExpBufferStr(query,
- " s.subfailover\n");
+ " s.subfailover,\n");
else
appendPQExpBuffer(query,
- " false AS subfailover\n");
+ " false AS subfailover,\n");
+
+ if (fout->remoteVersion >= 170000)
+ appendPQExpBufferStr(query,
+ " s.subforcealter\n");
+ else
+ appendPQExpBuffer(query,
+ " false AS subforcealter\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4854,6 +4862,7 @@ getSubscriptions(Archive *fout)
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
i_subenabled = PQfnumber(res, "subenabled");
i_subfailover = PQfnumber(res, "subfailover");
+ i_subforcealter = PQfnumber(res, "subforcealter");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4900,6 +4909,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subenabled));
subinfo[i].subfailover =
pg_strdup(PQgetvalue(res, i, i_subfailover));
+ subinfo[i].subforcealter =
+ pg_strdup(PQgetvalue(res, i, i_subforcealter));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5140,6 +5151,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subfailover, "t") == 0)
appendPQExpBufferStr(query, ", failover = true");
+ if (strcmp(subinfo->subforcealter, "t") == 0)
+ appendPQExpBufferStr(query, ", force_alter = true");
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 865823868f..bb0fda3b29 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -671,6 +671,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *suboriginremotelsn;
char *subfailover;
+ char *subforcealter;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index f67bf0b892..dba229a25d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6529,7 +6529,7 @@ describeSubscriptions(const char *pattern, bool verbose)
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false, false, false, false, false, false,
- false};
+ false, false};
if (pset.sversion < 100000)
{
@@ -6598,6 +6598,11 @@ describeSubscriptions(const char *pattern, bool verbose)
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subforcealter AS \"%s\"\n",
+ gettext_noop("Force alter"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..deac7aa943 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -98,6 +98,13 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool subforcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true"
+ * to "false". */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -151,6 +158,12 @@ typedef struct Subscription
* (i.e. the main slot and the table sync
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool forcealter; /* True allows the ALTER SUBSCRIPTION command
+ * to proceed under conditions that would
+ * otherwise result in an error. Currently,
+ * 'force_alter' only has an effect when
+ * altering the two_phase option from "true"
+ * to "false". */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9690..b36fc6b8f7 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,10 +393,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -409,18 +409,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Force alter | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index 635daf7a78..4da8392962 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -117,15 +117,43 @@ $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(1), "transaction has been prepared on subscriber");
-# Toggle the two_phase to "false" before the COMMIT PREPARED
-$node_subscriber->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION regress_sub DISABLE;
- ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
- ALTER SUBSCRIPTION regress_sub ENABLE;");
+# Disable the subscription to alter the two_phase option
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub DISABLE;");
+
+# Try altering the two_phase option to "false". The command will fail since
+# there is a prepared transaction and the 'force_alter' option is not specified
+# as true.
+my $stdout;
+my $stderr;
+
+($result, $stdout, $stderr) = $node_subscriber->psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub SET (two_phase = false);");
+ok( $stderr =~
+ /cannot alter two_phase = false when there are prepared transactions/,
+ 'ALTER SUBSCRIPTION failed');
+
+# Verify the prepared transaction still exists
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "prepared transaction still exists");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Alter the two_phase true to false with the force_alter option enabled. This
+# command will succeed after aborting the prepared transaction.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub SET (two_phase = false, force_alter = true);"
+);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED',
+ $log_offset);
-# Verify any prepared transactions are aborted because two_phase is changed to
-# "false".
+# # Verify the prepared transaction was aborted
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_prepared_xacts;");
is($result, q(0), "prepared transaction done by worker is aborted");
--
2.43.0
[application/octet-stream] v13-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch (25.5K, ../../OSBPR01MB25528ABFBE8B7DB1C4718BC6F5D52@OSBPR01MB2552.jpnprd01.prod.outlook.com/3-v13-0001-Allow-altering-of-two_phase-option-of-a-SUBSCRIP.patch)
download | inline diff:
From 3eeec94aa618c92b3ffedafe2499532cff75a8ee Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Fri, 5 Apr 2024 06:47:18 -0400
Subject: [PATCH v13 1/4] Allow altering of two_phase option of a SUBSCRIPTION
This patch allows the user to alter the 'two_phase' option of a subscriber provided no
uncommitted prepared transactions are pending on that subscription.
Author: Cherian Ajin, Hayato Kuroda
---
doc/src/sgml/ref/alter_subscription.sgml | 12 ++--
src/backend/access/transam/twophase.c | 62 ++++++++++++++++
src/backend/commands/subscriptioncmds.c | 68 ++++++++++++++----
.../libpqwalreceiver/libpqwalreceiver.c | 9 +--
src/backend/replication/logical/launcher.c | 22 ++++++
src/backend/replication/logical/worker.c | 21 +-----
src/backend/replication/slot.c | 18 ++++-
src/backend/replication/walsender.c | 18 ++++-
src/bin/psql/tab-complete.c | 2 +-
src/include/access/twophase.h | 5 ++
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 11 +--
src/include/replication/worker_internal.h | 1 +
src/test/regress/expected/subscription.out | 5 +-
src/test/regress/sql/subscription.sql | 5 +-
src/test/subscription/t/021_twophase.pl | 71 ++++++++++++++++++-
16 files changed, 270 insertions(+), 63 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 476f195622..0b23df1b77 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -68,8 +68,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command>,
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
- with <literal>refresh</literal> option as <literal>true</literal> and
- <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command>
+ with <literal>refresh</literal> option as <literal>true</literal>,
+ <command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
@@ -228,9 +229,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
- <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled.
</para>
<para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index bf451d42ff..0d51c31eb6 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2682,3 +2682,65 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
LWLockRelease(TwoPhaseStateLock);
return found;
}
+
+/*
+ * TwoPhaseTransactionGid
+ * Form the prepared transaction GID for two_phase transactions.
+ *
+ * Return the GID in the supplied buffer.
+ */
+void
+TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
+{
+ Assert(subid != InvalidRepOriginId);
+
+ if (!TransactionIdIsValid(xid))
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("invalid two-phase transaction ID")));
+
+ snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
+}
+
+/*
+ * IsTwoPhaseTransactionGidForSubid
+ * Check whether the given GID (as formed by TwoPhaseTransactionGid) is
+ * for the specified 'subid'.
+ */
+static bool
+IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
+{
+ int ret;
+ Oid subid_written;
+ TransactionId xid;
+
+ ret = sscanf(gid, "pg_gid_%u_%u", &subid_written, &xid);
+
+ return (ret == 2 && subid == subid_written);
+}
+
+/*
+ * LookupGXactBySubid
+ * Check if the prepared transaction done by apply worker exists.
+ */
+bool
+LookupGXactBySubid(Oid subid)
+{
+ bool found = false;
+
+ LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+ for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
+ {
+ GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+ /* Ignore not-yet-valid GIDs. */
+ if (gxact->valid &&
+ IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
+ {
+ found = true;
+ break;
+ }
+ }
+ LWLockRelease(TwoPhaseStateLock);
+ return found;
+}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e407428dbc..e925158a4d 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -16,6 +16,7 @@
#include "access/htup_details.h"
#include "access/table.h"
+#include "access/twophase.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
@@ -1143,7 +1144,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+ SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+ SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_ORIGIN);
@@ -1151,6 +1153,53 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
+ if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
+ {
+ /*
+ * Do not allow changing the two_phase option if the
+ * subscription is enabled. This is because the two_phase
+ * option of the slot on the publisher cannot be modified
+ * if the slot is currently acquired by the apply worker.
+ */
+ if (form->subenabled)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for enabled subscription",
+ "two_phase")));
+
+ /*
+ * Stop all the subscription workers, just in case.
+ * Workers may still survive even if the subscription is
+ * disabled.
+ */
+ logicalrep_workers_stop(subid);
+
+ /*
+ * two_phase cannot be disabled if there are any
+ * uncommitted prepared transactions present.
+ */
+ if (!opts.twophase &&
+ form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ LookupGXactBySubid(subid))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
+ errhint("Resolve these transactions and try again")));
+
+ /*
+ * The changed two_phase option of the slot can't be
+ * rolled back.
+ */
+ PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+
+ /* Change system catalog acoordingly */
+ values[Anum_pg_subscription_subtwophasestate - 1] =
+ CharGetDatum(opts.twophase ?
+ LOGICALREP_TWOPHASE_STATE_PENDING :
+ LOGICALREP_TWOPHASE_STATE_DISABLED);
+ replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
{
/*
@@ -1505,7 +1554,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1])
+ if (replaces[Anum_pg_subscription_subfailover - 1] ||
+ replaces[Anum_pg_subscription_subtwophasestate - 1])
{
bool must_use_password;
char *err;
@@ -1525,7 +1575,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
}
PG_FINALLY();
{
@@ -1562,7 +1612,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
- List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
@@ -1672,16 +1721,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
* New workers won't be started because we hold an exclusive lock on the
* subscription till the end of the transaction.
*/
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- subworkers = logicalrep_workers_find(subid, false);
- LWLockRelease(LogicalRepWorkerLock);
- foreach(lc, subworkers)
- {
- LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
-
- logicalrep_worker_stop(w->subid, w->relid);
- }
- list_free(subworkers);
+ logicalrep_workers_stop(subid);
/*
* Remove the no-longer-useful entry in the launcher's table of apply
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 02f12f2921..2f035a0c3c 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover);
+ bool failover, bool two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,15 +1121,16 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool failover, bool two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
quote_identifier(slotname),
- failover ? "true" : "false");
+ failover ? "true" : "false",
+ two_phase ? "true" : "false");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 27c3a91fb7..bef65b839b 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -623,6 +623,28 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockRelease(LogicalRepWorkerLock);
}
+/*
+ * Stop all the subscription workers.
+ */
+void
+logicalrep_workers_stop(Oid subid)
+{
+ List *subworkers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ /* XXX clarify the reason why not only running workers are listed. */
+ subworkers = logicalrep_workers_find(subid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+ foreach(lc, subworkers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ logicalrep_worker_stop(w->subid, w->relid);
+ }
+ list_free(subworkers);
+}
+
/*
* Stop the given logical replication parallel apply worker.
*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..dcf656fd45 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -402,7 +402,6 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata,
CmdType operation);
/* Compute GID for two_phase transactions */
-static void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid);
/* Functions for skipping changes */
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn);
@@ -3911,7 +3910,7 @@ maybe_reread_subscription(void)
/* !slotname should never happen when enabled is true. */
Assert(newsub->slotname);
- /* two-phase should not be altered */
+ /* two-phase cannot be altered while the worker exists */
Assert(newsub->twophasestate == MySubscription->twophasestate);
/*
@@ -4396,24 +4395,6 @@ cleanup_subxact_info()
subxact_data.nsubxacts_max = 0;
}
-/*
- * Form the prepared transaction GID for two_phase transactions.
- *
- * Return the GID in the supplied buffer.
- */
-static void
-TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
-{
- Assert(subid != InvalidRepOriginId);
-
- if (!TransactionIdIsValid(xid))
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("invalid two-phase transaction ID")));
-
- snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
-}
-
/*
* Common function to run the apply loop with error handling. Disable the
* subscription, if necessary.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 564cfee127..02e58d2a68 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -804,8 +804,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
{
+ bool update_slot = false;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -854,6 +856,20 @@ ReplicationSlotAlter(const char *name, bool failover)
MyReplicationSlot->data.failover = failover;
SpinLockRelease(&MyReplicationSlot->mutex);
+ update_slot = true;
+ }
+
+ if (MyReplicationSlot->data.two_phase != two_phase)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.two_phase = two_phase;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ update_slot = true;
+ }
+
+ if (update_slot)
+ {
ReplicationSlotMarkDirty();
ReplicationSlotSave();
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c623b07cf0..2e6ca35049 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,9 +1411,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
+ bool *failover, bool *two_phase)
{
bool failover_given = false;
+ bool two_phase_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
@@ -1427,6 +1429,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "two_phase") == 0)
+ {
+ if (two_phase_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ two_phase_given = true;
+ *two_phase = defGetBoolean(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1439,9 +1450,10 @@ static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
bool failover = false;
+ bool two_phase = false;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &failover, &two_phase);
+ ReplicationSlotAlter(cmd->slotname, failover, two_phase);
}
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index d453e224d9..891face1b6 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
"password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ "streaming", "synchronous_commit", "two_phase");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 56248c0006..d37e06fdee 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -62,4 +62,9 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
TimestampTz origin_prepare_timestamp);
+
+extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
+ int szgid);
+extern bool LookupGXactBySubid(Oid subid);
+
#endif /* TWOPHASE_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 1bc80960ef..014e216cf5 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,7 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool failover,
+ bool two_phase);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..31fa1257ec 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -372,12 +372,13 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
/*
* walrcv_alter_slot_fn
*
- * Change the definition of a replication slot. Currently, it only supports
- * changing the failover property of the slot.
+ * Change the definition of a replication slot. Currently, it supports
+ * changing the failover and the two_phase property of the slot.
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover);
+ bool failover,
+ bool two_phase);
/*
* walrcv_get_backend_pid_fn
@@ -455,8 +456,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
- WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_alter_slot(conn, slotname, failover, two_phase) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 515aefd519..d5428263c1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -246,6 +246,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
Oid userid, Oid relid,
dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_workers_stop(Oid subid);
extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 0f2a25cdc1..51fa4b9690 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -377,10 +377,7 @@ HINT: To initiate replication, you must manually create the replication slot, e
regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-ERROR: unrecognized subscription parameter: "two_phase"
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
List of subscriptions
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 3e5ba4cb8c..a3886d79ca 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,10 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
\dRs+
---fail - alter of two_phase option not supported.
-ALTER SUBSCRIPTION regress_testsub SET (two_phase = false);
-
--- but can alter streaming when two_phase enabled
+-- We can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 9437cd4c3b..0436cafdb8 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -367,6 +367,75 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
is($result, qq(2), 'replicated data in subscriber table');
+# Clean up
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+
+###############################
+# Disable the subscription and alter it to two_phase = false,
+# then verify that the altered subscription reflects the two_phase option.
+###############################
+
+# Alter subscription two_phase to false
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = false)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is disabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(d), 'two-phase should be disabled');
+
+# Now do a prepare on the publisher and make sure that it is not replicated.
+$node_publisher->safe_psql(
+ 'postgres', qq{
+ BEGIN;
+ INSERT INTO tab_copy VALUES (100);
+ PREPARE TRANSACTION 'newgid';
+ });
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure there are no prepared transactions on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'should be no prepared transactions on subscriber');
+
+# Now commit the insert and verify that it IS replicated
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
+
+# Wait for the subscriber to catchup
+$node_publisher->wait_for_catchup($appname_copy);
+
+# Make sure that the committed transaction is replicated.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
+is($result, qq(3), 'replicated data in subscriber table');
+
+# Alter subscription two_phase to true
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy DISABLE");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy SET (two_phase = true)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_copy ENABLE");
+
+# Wait for subscription startup
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_copy);
+
+# Make sure that the two-phase is enabled on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subtwophasestate FROM pg_subscription WHERE subname = 'tap_sub_copy';"
+);
+is($result, qq(e), 'two-phase should be enabled');
+
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_copy;");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
@@ -374,8 +443,6 @@ $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_copy;");
# check all the cleanup
###############################
-$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
-
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_subscription");
is($result, qq(0), 'check subscription was dropped on subscriber');
--
2.43.0
[application/octet-stream] v13-0002-Alter-slot-option-two_phase-only-when-altering-t.patch (12.3K, ../../OSBPR01MB25528ABFBE8B7DB1C4718BC6F5D52@OSBPR01MB2552.jpnprd01.prod.outlook.com/4-v13-0002-Alter-slot-option-two_phase-only-when-altering-t.patch)
download | inline diff:
From e66634d369106b908fec321a15f6514c22ef847e Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Wed, 17 Apr 2024 06:18:23 +0000
Subject: [PATCH v13 2/4] Alter slot option two_phase only when altering "true"
to "false"
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Since the two_phase option is controlled by both the publisher (as a slot option)
and the subscriber (as a subscription option), the slot option must also be
modified.
Regarding the false->true case, the backend process alters the subtwophase to
LOGICALREP_TWOPHASE_STATE_PENDING once. After the subscription is enabled, a new
logical replication worker requests to change the two_phase option of its slot
from pending to true after the initial data synchronization is done. The code
path is the same as the case in which two_phase is initially set to true, so
there is no need to do something remarkable. However, for the true->false case,
the backend must connect to the publisher and expressly change the parameter
because the apply worker does not alter the option to false. Because this
operation cannot be rolled back, altering the two_phase parameter from "true"
to "false" within a transaction is prohibited.
---
doc/src/sgml/ref/alter_subscription.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 43 ++++++--
.../libpqwalreceiver/libpqwalreceiver.c | 23 +++--
src/include/replication/walreceiver.h | 5 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/099_twophase_added.pl | 99 +++++++++++++++++++
6 files changed, 157 insertions(+), 16 deletions(-)
create mode 100644 src/test/subscription/t/099_twophase_added.pl
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0b23df1b77..475a42a2e3 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -70,7 +70,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command>
with <literal>refresh</literal> option as <literal>true</literal>,
<command>ALTER SUBSCRIPTION ... SET (failover = true|false)</command> and
- <command>ALTER SUBSCRIPTION ... SET (two_phase = true|false)</command>
+ <command>ALTER SUBSCRIPTION ... SET (two_phase = off)</command>
cannot be executed inside a transaction block.
These commands also cannot be executed when the subscription has
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e925158a4d..996ea6b6de 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1097,6 +1097,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
Form_pg_subscription form;
bits32 supported_opts;
SubOpts opts = {0};
+ bool update_failover;
+ bool update_two_phase;
rel = table_open(SubscriptionRelationId, RowExclusiveLock);
@@ -1187,10 +1189,25 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errhint("Resolve these transactions and try again")));
/*
- * The changed two_phase option of the slot can't be
- * rolled back.
+ * Altering the parameter from "true" to "false" within a
+ * transaction is prohibited. Since the apply worker does
+ * not alter the slot option to false, the backend must
+ * connect to the publisher and expressly change the
+ * parameter.
+ *
+ * There is no need to do something remarkable regarding
+ * the "false" to "true" case; the backend process alters
+ * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
+ * After the subscription is enabled, a new logical
+ * replication worker requests to change the two_phase
+ * option of its slot from pending to true when the
+ * initial data synchronization is done. The code path is
+ * the same as the case in which two_phase is initially
+ * set to true.
*/
- PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... SET (two_phase)");
+ if (!opts.twophase)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER SUBSCRIPTION ... SET (two_phase = false)");
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
@@ -1548,14 +1565,24 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
}
/*
- * Try to acquire the connection necessary for altering slot.
+ * Check the need to alter the replication slot. Failover and two_phase
+ * options are controlled by both the publisher (as a slot option) and the
+ * subscriber (as a subscription option). The slot option must be altered
+ * only when changing "true" to "false". The reason has already been
+ * described in the ALTER_SUBSCRIPTION_OPTIONS section of this function.
+ */
+ update_failover = replaces[Anum_pg_subscription_subfailover - 1];
+ update_two_phase = (replaces[Anum_pg_subscription_subtwophasestate - 1] &&
+ !opts.twophase);
+
+ /*
+ * Try to acquire the connection necessary for altering slot, if needed.
*
* This has to be at the end because otherwise if there is an error while
* doing the database operations we won't be able to rollback altered
* slot.
*/
- if (replaces[Anum_pg_subscription_subfailover - 1] ||
- replaces[Anum_pg_subscription_subtwophasestate - 1])
+ if (update_failover || update_two_phase)
{
bool must_use_password;
char *err;
@@ -1575,7 +1602,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover, opts.twophase);
+ walrcv_alter_slot(wrconn, sub->slotname,
+ update_failover ? &opts.failover : NULL,
+ update_two_phase ? &opts.twophase : NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 2f035a0c3c..07dfec947d 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,7 +80,7 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase);
+ const bool *failover, const bool *two_phase);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1121,16 +1121,27 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover, bool two_phase)
+ const bool *failover, const bool *two_phase)
{
StringInfoData cmd;
PGresult *res;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s, TWO_PHASE %s )",
- quote_identifier(slotname),
- failover ? "true" : "false",
- two_phase ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
+ quote_identifier(slotname));
+
+ if (failover)
+ appendStringInfo(&cmd, "FAILOVER %s",
+ *failover ? "true" : "false");
+
+ if (failover && two_phase)
+ appendStringInfo(&cmd, ", ");
+
+ if (two_phase)
+ appendStringInfo(&cmd, "TWO_PHASE %s",
+ *two_phase ? "true" : "false");
+
+ appendStringInfoString(&cmd, " );");
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 31fa1257ec..7ffa5a58b3 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -377,8 +377,9 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover,
- bool two_phase);
+ const bool *failover,
+ const bool *two_phase);
+
/*
* walrcv_get_backend_pid_fn
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..b4bd522c3d 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
't/031_column_list.pl',
't/032_subscribe_use_index.pl',
't/033_run_as_table_owner.pl',
+ 't/099_twophase_added.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
new file mode 100644
index 0000000000..1124f7fa00
--- /dev/null
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -0,0 +1,99 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Additional tests for altering two_phase option
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf(
+ 'postgresql.conf',
+ qq(max_prepared_transactions = 10
+ log_min_messages = debug1));
+$node_subscriber->start;
+
+# Define tables on both nodes
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication, with two_phase = "false"
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION pub FOR ALL TABLES");
+
+my $log_offset = -s $node_subscriber->logfile;
+
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION regress_sub
+ CONNECTION '$publisher_connstr' PUBLICATION pub
+ WITH (two_phase = false, copy_data = false, failover = false)");
+
+# Verify the started worker recognized two_phase was disabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is DISABLED',
+ $log_offset);
+
+#####################
+# Check the case that prepared transactions exist on the publisher node.
+#
+# Since the two_phase is "false", then normally, this PREPARE will do nothing
+# until the COMMIT PREPARED, but in this test, we toggle the two_phase to
+# "true" again before the COMMIT PREPARED happens.
+
+# Prepare a transaction to insert some tuples into the table
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(1, 5));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the prepared transaction is not yet replicated to the subscriber
+# because two_phase is set to "false".
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "transaction is not prepared on subscriber");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Toggle the two_phase to "true" *before* the COMMIT PREPARED. Since we are the
+# special path for the case where both two_phase and failover are altered, it
+# is also set to "true".
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = true, failover = true);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify the started worker recognized two_phase was enabled
+$node_subscriber->wait_for_log(
+ 'logical replication apply worker for subscription "regress_sub" two_phase is ENABLED',
+ $log_offset);
+
+# And do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_full;");
+is($result, q(5),
+ "prepared transactions done before altering can be replicated");
+
+done_testing();
--
2.43.0
[application/octet-stream] v13-0003-Abort-prepared-transactions-while-altering-two_p.patch (9.5K, ../../OSBPR01MB25528ABFBE8B7DB1C4718BC6F5D52@OSBPR01MB2552.jpnprd01.prod.outlook.com/5-v13-0003-Abort-prepared-transactions-while-altering-two_p.patch)
download | inline diff:
From bce3f77fb1009f866a0fe3fc616c8218ec900f13 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 8 Apr 2024 12:39:12 +0000
Subject: [PATCH v13 3/4] Abort prepared transactions while altering two_phase
to off
If we alter the two_phase parameter from "on" to "off" and there are prepared
transactions on the subscriber, they won't be resolved. To avoid this issue, we
allow the backend to abort all prepared transactions while altering the
subscription.
---
doc/src/sgml/ref/alter_subscription.sgml | 11 +++-
src/backend/access/transam/twophase.c | 17 ++---
src/backend/commands/subscriptioncmds.c | 66 +++++++++++--------
src/include/access/twophase.h | 3 +-
src/test/subscription/t/099_twophase_added.pl | 44 +++++++++++++
5 files changed, 101 insertions(+), 40 deletions(-)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 475a42a2e3..8801f37f0e 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -233,8 +233,6 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
- The <literal>two_phase</literal> parameter can only be altered when the
- subscription is disabled.
</para>
<para>
@@ -256,6 +254,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
option is enabled.
</para>
+
+ <para>
+ The <literal>two_phase</literal> parameter can only be altered when the
+ subscription is disabled. When altering the parameter from <literal>true</literal>
+ to <literal>false</literal>, the backend process checks for any incomplete
+ prepared transactions done by the logical replication worker (from when
+ <literal>two_phase</literal> parameter was still <literal>true</literal>)
+ and, if any are found, those are aborted.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 0d51c31eb6..ae1a97177c 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2720,13 +2720,13 @@ IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
}
/*
- * LookupGXactBySubid
- * Check if the prepared transaction done by apply worker exists.
+ * GetGidListBySubid
+ * Get a list of GIDs which is PREPARE'd by the given subscription.
*/
-bool
-LookupGXactBySubid(Oid subid)
+List *
+GetGidListBySubid(Oid subid)
{
- bool found = false;
+ List *list = NIL;
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
@@ -2736,11 +2736,8 @@ LookupGXactBySubid(Oid subid)
/* Ignore not-yet-valid GIDs. */
if (gxact->valid &&
IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
- {
- found = true;
- break;
- }
+ list = lappend(list, pstrdup(gxact->gid));
}
LWLockRelease(TwoPhaseStateLock);
- return found;
+ return list;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 996ea6b6de..2532b40365 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1177,38 +1177,50 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
logicalrep_workers_stop(subid);
/*
- * two_phase cannot be disabled if there are any
- * uncommitted prepared transactions present.
- */
- if (!opts.twophase &&
- form->subtwophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
- LookupGXactBySubid(subid))
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
- errhint("Resolve these transactions and try again")));
-
- /*
- * Altering the parameter from "true" to "false" within a
- * transaction is prohibited. Since the apply worker does
- * not alter the slot option to false, the backend must
- * connect to the publisher and expressly change the
- * parameter.
- *
- * There is no need to do something remarkable regarding
- * the "false" to "true" case; the backend process alters
- * subtwophase to LOGICALREP_TWOPHASE_STATE_PENDING once.
- * After the subscription is enabled, a new logical
- * replication worker requests to change the two_phase
- * option of its slot from pending to true when the
- * initial data synchronization is done. The code path is
- * the same as the case in which two_phase is initially
- * set to true.
+ * If two_phase was previously enabled, there is a
+ * possibility that transactions have already been
+ * PREPARE'd. They must be checked and rolled back.
*/
if (!opts.twophase)
+ {
+ List *prepared_xacts;
+
+ /*
+ * Altering the parameter from "true" to "false"
+ * within a transaction is prohibited. Since the apply
+ * worker does not alter the slot option to false, the
+ * backend must connect to the publisher and expressly
+ * change the parameter.
+ *
+ * There is no need to do something remarkable
+ * regarding the "false" to "true" case; the backend
+ * process alters subtwophase to
+ * LOGICALREP_TWOPHASE_STATE_PENDING once. After the
+ * subscription is enabled, a new logical replication
+ * worker requests to change the two_phase option of
+ * its slot from pending to true when the initial data
+ * synchronization is done. The code path is the same
+ * as the case in which two_phase is initially
+ * set to true.
+ */
PreventInTransactionBlock(isTopLevel,
"ALTER SUBSCRIPTION ... SET (two_phase = false)");
+ /*
+ * To prevent prepared transactions from being
+ * isolated, they must manually be aborted.
+ */
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ (prepared_xacts = GetGidListBySubid(subid)) != NIL)
+ {
+ /* Abort all listed transactions */
+ foreach_ptr(char, gid, prepared_xacts)
+ FinishPreparedTransaction(gid, false);
+
+ list_free_deep(prepared_xacts);
+ }
+ }
+
/* Change system catalog acoordingly */
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index d37e06fdee..f7a5cf0c12 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "nodes/pg_list.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
@@ -65,6 +66,6 @@ extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid,
int szgid);
-extern bool LookupGXactBySubid(Oid subid);
+extern List *GetGidListBySubid(Oid subid);
#endif /* TWOPHASE_H */
diff --git a/src/test/subscription/t/099_twophase_added.pl b/src/test/subscription/t/099_twophase_added.pl
index 1124f7fa00..635daf7a78 100644
--- a/src/test/subscription/t/099_twophase_added.pl
+++ b/src/test/subscription/t/099_twophase_added.pl
@@ -96,4 +96,48 @@ $result =
is($result, q(5),
"prepared transactions done before altering can be replicated");
+#####################
+# Check the case that prepared transactions exist on the subscriber node
+#
+# If the two_phase is altering from "true" to "false" and there are prepared
+# transactions on the subscriber, they must be aborted. This test checks it.
+
+# Prepare a transaction to insert some tuples into the table
+$node_publisher->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO tab_full VALUES (generate_series(6, 10));
+ PREPARE TRANSACTION 'test_prepared_tab_full';");
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the prepared transaction has been replicated to the subscriber because
+# two_phase is set to "true".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(1), "transaction has been prepared on subscriber");
+
+# Toggle the two_phase to "false" before the COMMIT PREPARED
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub DISABLE;
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = false);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Verify any prepared transactions are aborted because two_phase is changed to
+# "false".
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, q(0), "prepared transaction done by worker is aborted");
+
+# Do COMMIT PREPARED the prepared transaction
+$node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab_full';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify inserted tuples are replicated
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(10) FROM tab_full;");
+is($result, q(10), "prepared transactions on publisher can be replicated");
+
done_testing();
--
2.43.0
^ permalink raw reply [nested|flat] 5+ messages in thread
* RE: Slow catchup of 2PC (twophase) transactions on replica in LR
@ 2024-06-25 09:11 Hayato Kuroda (Fujitsu) <[email protected]>
parent: Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2024-06-25 09:11 UTC (permalink / raw)
To: [email protected] <[email protected]>; +Cc: Amit Kapila <[email protected]>; Ajin Cherian <[email protected]>; Давыдов Виталий <[email protected]>; 'Peter Smith' <[email protected]>
> Dear hackers,
>
> I found that v12 patch set could not be accepted by the cfbot. PSA new version.
To make others more trackable, I shared changes just in case. All failures were occurred
on the pg_dump code. I added an attribute in pg_subscription and modified pg_dump code,
but it was wrong. A constructed SQL became incomplete. I.e., in [1]:
```
pg_dump: error: query failed: ERROR: syntax error at or near "."
LINE 15: s.subforcealter
^
pg_dump: detail: Query was: SELECT s.tableoid, s.oid, s.subname,
s.subowner,
s.subconninfo, s.subslotname, s.subsynccommit,
s.subpublications,
s.subbinary,
s.substream,
s.subtwophasestate,
s.subdisableonerr,
s.subpasswordrequired,
s.subrunasowner,
s.suborigin,
NULL AS suboriginremotelsn,
false AS subenabled,
s.subfailover
s.subforcealter
FROM pg_subscription s
WHERE s.subdbid = (SELECT oid FROM pg_database
WHERE datname = current_database())
```
Based on that I just added a comma in 0004 patch.
[1]: https://cirrus-ci.com/task/6710166165389312
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/
^ permalink raw reply [nested|flat] 5+ messages in thread
end of thread, other threads:[~2024-06-25 09:11 UTC | newest]
Thread overview: 5+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2022-07-24 09:43 [PATCH v11] Prevent jumbling of every element in ArrayExpr Dmitrii Dolgov <[email protected]>
2024-05-17 03:06 RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-05-17 04:08 ` Re: Slow catchup of 2PC (twophase) transactions on replica in LR Peter Smith <[email protected]>
2024-06-25 08:06 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
2024-06-25 09:11 ` RE: Slow catchup of 2PC (twophase) transactions on replica in LR Hayato Kuroda (Fujitsu) <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox