public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 08/10] Default to zstd..
8+ messages / 6 participants
[nested] [flat]
* [PATCH 08/10] Default to zstd..
@ 2021-03-12 21:35 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 8+ messages in thread
From: Justin Pryzby @ 2021-03-12 21:35 UTC (permalink / raw)
for CI, not for merge
---
configure | 6 ++++--
configure.ac | 2 +-
src/backend/access/transam/xlog.c | 2 +-
src/backend/utils/misc/guc.c | 2 +-
4 files changed, 7 insertions(+), 5 deletions(-)
diff --git a/configure b/configure
index 81e23418b2..253f028fc4 100755
--- a/configure
+++ b/configure
@@ -1582,7 +1582,7 @@ Optional Packages:
use system time zone data in DIR
--without-zlib do not use Zlib
--without-lz4 build without LZ4 support
- --with-zstd build with Zstd compression library
+ --without-zstd build without Zstd compression library
--with-gnu-ld assume the C compiler uses GNU ld [default=no]
--with-ssl=LIB use LIB for SSL/TLS support (openssl)
--with-openssl obsolete spelling of --with-ssl=openssl
@@ -8740,7 +8740,9 @@ $as_echo "#define USE_ZSTD 1" >>confdefs.h
esac
else
- with_zstd=no
+ with_zstd=yes
+
+$as_echo "#define USE_ZSTD 1" >>confdefs.h
fi
diff --git a/configure.ac b/configure.ac
index d6f6349067..8d72710fa7 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1005,7 +1005,7 @@ fi
# ZSTD
#
AC_MSG_CHECKING([whether to build with zstd support])
-PGAC_ARG_BOOL(with, zstd, no, [build with Zstd compression library],
+PGAC_ARG_BOOL(with, zstd, yes, [build without Zstd compression library],
[AC_DEFINE([USE_ZSTD], 1, [Define to 1 to build with zstd support. (--with-zstd)])])
AC_MSG_RESULT([$with_zstd])
AC_SUBST(with_zstd)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 307eee6626..92023de9f5 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -99,7 +99,7 @@ bool EnableHotStandby = false;
bool fullPageWrites = true;
bool wal_log_hints = false;
bool wal_compression = false;
-int wal_compression_method = WAL_COMPRESSION_LZ4;
+int wal_compression_method = WAL_COMPRESSION_ZSTD;
char *wal_consistency_checking_string = NULL;
bool *wal_consistency_checking = NULL;
bool wal_init_zero = true;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 52f9cd0242..8031e027aa 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -4728,7 +4728,7 @@ static struct config_enum ConfigureNamesEnum[] =
NULL
},
&wal_compression_method,
- WAL_COMPRESSION_LZ4, wal_compression_options,
+ WAL_COMPRESSION_ZSTD, wal_compression_options,
NULL, NULL, NULL
},
--
2.17.0
--jozmn01XJZjDjM3N
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0009-Add-zstd-compression-levels.patch"
^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v5 1/2] fix MAINTAIN privs for partitions
@ 2023-01-09 22:29 Nathan Bossart <[email protected]>
0 siblings, 0 replies; 8+ messages in thread
From: Nathan Bossart @ 2023-01-09 22:29 UTC (permalink / raw)
---
doc/src/sgml/ref/analyze.sgml | 5 ++-
doc/src/sgml/ref/cluster.sgml | 17 ++++++---
doc/src/sgml/ref/lock.sgml | 5 ++-
doc/src/sgml/ref/reindex.sgml | 6 +++-
doc/src/sgml/ref/vacuum.sgml | 5 ++-
src/backend/commands/cluster.c | 35 +++++++++++++------
src/backend/commands/indexcmds.c | 22 ++++++------
src/backend/commands/lockcmds.c | 8 +++++
src/backend/commands/tablecmds.c | 30 ++++++++++++++--
src/backend/commands/vacuum.c | 9 +++--
src/include/commands/tablecmds.h | 1 +
.../expected/cluster-conflict-partition.out | 6 ++--
.../specs/cluster-conflict-partition.spec | 5 +--
src/test/regress/expected/cluster.out | 4 ++-
src/test/regress/expected/vacuum.out | 18 ----------
src/test/regress/sql/cluster.sql | 2 ++
16 files changed, 119 insertions(+), 59 deletions(-)
diff --git a/doc/src/sgml/ref/analyze.sgml b/doc/src/sgml/ref/analyze.sgml
index a26834da4f..2f94e89cb0 100644
--- a/doc/src/sgml/ref/analyze.sgml
+++ b/doc/src/sgml/ref/analyze.sgml
@@ -156,7 +156,10 @@ ANALYZE [ VERBOSE ] [ <replaceable class="parameter">table_and_columns</replacea
analyze all tables in their databases, except shared catalogs.
(The restriction for shared catalogs means that a true database-wide
<command>ANALYZE</command> can only be performed by superusers and roles
- with privileges of <literal>pg_maintain</literal>.)
+ with privileges of <literal>pg_maintain</literal>.) If a role has
+ permission to <command>ANALYZE</command> a partitioned table, it is also
+ permitted to <command>ANALYZE</command> each of its partitions, regardless
+ of whether the role has the aforementioned privileges on the partition.
<command>ANALYZE</command> will skip over any tables that the calling user
does not have permission to analyze.
</para>
diff --git a/doc/src/sgml/ref/cluster.sgml b/doc/src/sgml/ref/cluster.sgml
index 145101e6a5..b9f2acb1de 100644
--- a/doc/src/sgml/ref/cluster.sgml
+++ b/doc/src/sgml/ref/cluster.sgml
@@ -69,10 +69,7 @@ CLUSTER [VERBOSE]
<para>
<command>CLUSTER</command> without any parameter reclusters all the
previously-clustered tables in the current database that the calling user
- owns or has the <literal>MAINTAIN</literal> privilege for, or all such tables
- if called by a superuser or a role with privileges of the
- <link linkend="predefined-roles-table"><literal>pg_maintain</literal></link>
- role. This form of <command>CLUSTER</command> cannot be
+ has privileges for. This form of <command>CLUSTER</command> cannot be
executed inside a transaction block.
</para>
@@ -134,6 +131,18 @@ CLUSTER [VERBOSE]
<refsect1>
<title>Notes</title>
+ <para>
+ To cluster a table, one must have the <literal>MAINTAIN</literal> privilege
+ on the table or be the table's owner, a superuser, or a role with
+ privileges of the
+ <link linkend="predefined-roles-table"><literal>pg_maintain</literal></link>
+ role. If a role has permission to <command>CLUSTER</command> a partitioned
+ table, it is also permitted to <command>CLUSTER</command> each of its
+ partitions, regardless of whether the role has the aforementioned
+ privileges on the partition. <command>CLUSTER</command> will skip over any
+ tables that the calling user does not have permission to cluster.
+ </para>
+
<para>
In cases where you are accessing single rows randomly
within a table, the actual order of the data in the
diff --git a/doc/src/sgml/ref/lock.sgml b/doc/src/sgml/ref/lock.sgml
index d9c5bf9a1d..21a9f88c70 100644
--- a/doc/src/sgml/ref/lock.sgml
+++ b/doc/src/sgml/ref/lock.sgml
@@ -176,7 +176,10 @@ LOCK [ TABLE ] [ ONLY ] <replaceable class="parameter">name</replaceable> [ * ]
or <literal>TRUNCATE</literal> privileges on the target table. All other
forms of <command>LOCK</command> are allowed with
table-level <literal>UPDATE</literal>, <literal>DELETE</literal>,
- or <literal>TRUNCATE</literal> privileges.
+ or <literal>TRUNCATE</literal> privileges. If a role has permission to
+ lock a partitioned table, it is also permitted to lock each of its
+ partitions, regardless of whether the role has the aforementioned
+ privileges on the partition.
</para>
<para>
diff --git a/doc/src/sgml/ref/reindex.sgml b/doc/src/sgml/ref/reindex.sgml
index 192513f34e..c6ad2546f9 100644
--- a/doc/src/sgml/ref/reindex.sgml
+++ b/doc/src/sgml/ref/reindex.sgml
@@ -306,7 +306,11 @@ REINDEX [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] { DA
indexes on shared catalogs will be skipped unless the user owns the
catalog (which typically won't be the case), has privileges of the
<literal>pg_maintain</literal> role, or has the <literal>MAINTAIN</literal>
- privilege on the catalog. Of course, superusers can always reindex anything.
+ privilege on the catalog. If a role has permission to
+ <command>REINDEX</command> a partitioned table, it is also permitted to
+ <command>REINDEX</command> each of its partitions, regardless of whether the
+ role has the aforementioned privileges on the partition. Of course,
+ superusers can always reindex anything.
</para>
<para>
diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml
index 8fa8421847..545b23b54f 100644
--- a/doc/src/sgml/ref/vacuum.sgml
+++ b/doc/src/sgml/ref/vacuum.sgml
@@ -401,7 +401,10 @@ VACUUM [ FULL ] [ FREEZE ] [ VERBOSE ] [ ANALYZE ] [ <replaceable class="paramet
vacuum all tables in their databases, except shared catalogs.
(The restriction for shared catalogs means that a true database-wide
<command>VACUUM</command> can only be performed by superusers and roles
- with privileges of <literal>pg_maintain</literal>.)
+ with privileges of <literal>pg_maintain</literal>.) If a role has
+ permission to <command>VACUUM</command> a partitioned table, it is also
+ permitted to <command>VACUUM</command> each of its partitions, regardless
+ of whether the role has the aforementioned privileges on the partition.
<command>VACUUM</command> will skip over any tables that the calling user
does not have permission to vacuum.
</para>
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index f11691aff7..369fea7c04 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -80,6 +80,7 @@ static void copy_table_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex,
static List *get_tables_to_cluster(MemoryContext cluster_context);
static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context,
Oid indexOid);
+static bool cluster_is_permitted_for_relation(Oid relid, Oid userid);
/*---------------------------------------------------------------------------
@@ -366,8 +367,7 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params)
if (recheck)
{
/* Check that the user still has privileges for the relation */
- if (!object_ownercheck(RelationRelationId, tableOid, save_userid) &&
- pg_class_aclcheck(tableOid, save_userid, ACL_MAINTAIN) != ACLCHECK_OK)
+ if (!cluster_is_permitted_for_relation(tableOid, save_userid))
{
relation_close(OldHeap, AccessExclusiveLock);
goto out;
@@ -1645,8 +1645,7 @@ get_tables_to_cluster(MemoryContext cluster_context)
index = (Form_pg_index) GETSTRUCT(indexTuple);
- if (!object_ownercheck(RelationRelationId, index->indrelid, GetUserId()) &&
- pg_class_aclcheck(index->indrelid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (!cluster_is_permitted_for_relation(index->indrelid, GetUserId()))
continue;
/* Use a permanent memory context for the result list */
@@ -1694,12 +1693,11 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
if (get_rel_relkind(indexrelid) != RELKIND_INDEX)
continue;
- /* Silently skip partitions which the user has no access to. */
- if (!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
- (!object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) ||
- IsSharedRelation(relid)))
- continue;
+ /*
+ * We already checked that the user has privileges to CLUSTER the
+ * partitioned table when we locked it earlier, so there's no need to
+ * check the privileges again here.
+ */
/* Use a permanent memory context for the result list */
old_context = MemoryContextSwitchTo(cluster_context);
@@ -1714,3 +1712,20 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
return rtcs;
}
+
+/*
+ * Return whether userid has privileges to CLUSTER relid. If not, this
+ * function emits a WARNING.
+ */
+static bool
+cluster_is_permitted_for_relation(Oid relid, Oid userid)
+{
+ if (pg_class_aclcheck(relid, userid, ACL_MAINTAIN) == ACLCHECK_OK ||
+ has_partition_ancestor_privs(relid, userid, ACL_MAINTAIN))
+ return true;
+
+ ereport(WARNING,
+ (errmsg("permission denied to cluster \"%s\", skipping it",
+ get_rel_name(relid))));
+ return false;
+}
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 8afc006f89..16ec0b114e 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -2796,9 +2796,9 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation,
/* Check permissions */
table_oid = IndexGetRelation(relId, true);
- if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
- OidIsValid(table_oid) &&
- pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (OidIsValid(table_oid) &&
+ pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(table_oid, GetUserId(), ACL_MAINTAIN))
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
relation->relname);
@@ -3008,16 +3008,16 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
/*
* The table can be reindexed if the user has been granted MAINTAIN on
- * the table or the user is a superuser, the table owner, or the
- * database/schema owner (but in the latter case, only if it's not a
- * shared relation). object_ownercheck includes the superuser case,
- * and depending on objectKind we already know that the user has
- * permission to run REINDEX on this database or schema per the
- * permission checks at the beginning of this routine.
+ * the table or one of its partition ancestors or the user is a
+ * superuser, the table owner, or the database/schema owner (but in the
+ * latter case, only if it's not a shared relation). pg_class_aclcheck
+ * includes the superuser case, and depending on objectKind we already
+ * know that the user has permission to run REINDEX on this database or
+ * schema per the permission checks at the beginning of this routine.
*/
if (classtuple->relisshared &&
- !object_ownercheck(RelationRelationId, relid, GetUserId()) &&
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(relid, GetUserId(), ACL_MAINTAIN))
continue;
/*
diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index 99e68bff85..de2a975882 100644
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -19,6 +19,7 @@
#include "catalog/namespace.h"
#include "catalog/pg_inherits.h"
#include "commands/lockcmds.h"
+#include "commands/tablecmds.h"
#include "miscadmin.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
@@ -305,5 +306,12 @@ LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)
aclresult = pg_class_aclcheck(reloid, userid, aclmask);
+ /*
+ * If this is a partition, check permissions of its ancestors if needed.
+ */
+ if (aclresult != ACLCHECK_OK &&
+ has_partition_ancestor_privs(reloid, userid, ACL_MAINTAIN))
+ aclresult = ACLCHECK_OK;
+
return aclresult;
}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1fbdad4b64..7c697a285b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16886,12 +16886,38 @@ RangeVarCallbackMaintainsTable(const RangeVar *relation,
errmsg("\"%s\" is not a table or materialized view", relation->relname)));
/* Check permissions */
- if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
- pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(relId, GetUserId(), ACL_MAINTAIN))
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
relation->relname);
}
+/*
+ * If relid is a partition, returns whether userid has any of the privileges
+ * specified in acl on any of its ancestors. Otherwise, returns false.
+ */
+bool
+has_partition_ancestor_privs(Oid relid, Oid userid, AclMode acl)
+{
+ List *ancestors;
+ ListCell *lc;
+
+ if (!get_rel_relispartition(relid))
+ return false;
+
+ ancestors = get_partition_ancestors(relid);
+ foreach(lc, ancestors)
+ {
+ Oid ancestor = lfirst_oid(lc);
+
+ if (OidIsValid(ancestor) &&
+ pg_class_aclcheck(ancestor, userid, acl) == ACLCHECK_OK)
+ return true;
+ }
+
+ return false;
+}
+
/*
* Callback to RangeVarGetRelidExtended() for TRUNCATE processing.
*/
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index c4ed7efce3..5908733347 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -41,6 +41,7 @@
#include "catalog/pg_namespace.h"
#include "commands/cluster.h"
#include "commands/defrem.h"
+#include "commands/tablecmds.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
@@ -598,11 +599,13 @@ vacuum_is_permitted_for_relation(Oid relid, Form_pg_class reltuple,
* - the role owns the relation
* - the role owns the current database and the relation is not shared
* - the role has been granted the MAINTAIN privilege on the relation
+ * - the role has privileges to vacuum/analyze any of the relation's
+ * partition ancestors
*----------
*/
- if (object_ownercheck(RelationRelationId, relid, GetUserId()) ||
- (object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK)
+ if ((object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
+ pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK ||
+ has_partition_ancestor_privs(relid, GetUserId(), ACL_MAINTAIN))
return true;
relname = NameStr(reltuple->relname);
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 2e717fa815..e7c2b91a58 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -98,6 +98,7 @@ extern void AtEOSubXact_on_commit_actions(bool isCommit,
extern void RangeVarCallbackMaintainsTable(const RangeVar *relation,
Oid relId, Oid oldRelId,
void *arg);
+extern bool has_partition_ancestor_privs(Oid relid, Oid userid, AclMode acl);
extern void RangeVarCallbackOwnsRelation(const RangeVar *relation,
Oid relId, Oid oldRelId, void *arg);
diff --git a/src/test/isolation/expected/cluster-conflict-partition.out b/src/test/isolation/expected/cluster-conflict-partition.out
index 7acb675c97..8d21276996 100644
--- a/src/test/isolation/expected/cluster-conflict-partition.out
+++ b/src/test/isolation/expected/cluster-conflict-partition.out
@@ -22,14 +22,16 @@ starting permutation: s1_begin s1_lock_child s2_auth s2_cluster s1_commit s2_res
step s1_begin: BEGIN;
step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
step s2_auth: SET ROLE regress_cluster_part;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
step s1_commit: COMMIT;
+step s2_cluster: <... completed>
step s2_reset: RESET ROLE;
starting permutation: s1_begin s2_auth s1_lock_child s2_cluster s1_commit s2_reset
step s1_begin: BEGIN;
step s2_auth: SET ROLE regress_cluster_part;
step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
step s1_commit: COMMIT;
+step s2_cluster: <... completed>
step s2_reset: RESET ROLE;
diff --git a/src/test/isolation/specs/cluster-conflict-partition.spec b/src/test/isolation/specs/cluster-conflict-partition.spec
index 5091f684a9..ae38cb4ee3 100644
--- a/src/test/isolation/specs/cluster-conflict-partition.spec
+++ b/src/test/isolation/specs/cluster-conflict-partition.spec
@@ -27,11 +27,8 @@ step s2_auth { SET ROLE regress_cluster_part; }
step s2_cluster { CLUSTER cluster_part_tab USING cluster_part_ind; }
step s2_reset { RESET ROLE; }
-# CLUSTER on the parent waits if locked, passes for all cases.
+# CLUSTER waits if locked, passes for all cases.
permutation s1_begin s1_lock_parent s2_auth s2_cluster s1_commit s2_reset
permutation s1_begin s2_auth s1_lock_parent s2_cluster s1_commit s2_reset
-
-# When taking a lock on a partition leaf, CLUSTER on the parent skips
-# the leaf, passes for all cases.
permutation s1_begin s1_lock_child s2_auth s2_cluster s1_commit s2_reset
permutation s1_begin s2_auth s1_lock_child s2_cluster s1_commit s2_reset
diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expected/cluster.out
index 542c2e098c..2eec483eaa 100644
--- a/src/test/regress/expected/cluster.out
+++ b/src/test/regress/expected/cluster.out
@@ -352,7 +352,9 @@ INSERT INTO clstr_3 VALUES (1);
-- this user can only cluster clstr_1 and clstr_3, but the latter
-- has not been clustered
SET SESSION AUTHORIZATION regress_clstr_user;
+SET client_min_messages = ERROR; -- order of "skipping" warnings may vary
CLUSTER;
+RESET client_min_messages;
SELECT * FROM clstr_1 UNION ALL
SELECT * FROM clstr_2 UNION ALL
SELECT * FROM clstr_3;
@@ -513,7 +515,7 @@ SELECT a.relname, a.relfilenode=b.relfilenode FROM pg_class a
-----------+----------
ptnowner | t
ptnowner1 | f
- ptnowner2 | t
+ ptnowner2 | f
(3 rows)
DROP TABLE ptnowner;
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index d860be0e20..458adee7f8 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -353,20 +353,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
ALTER TABLE vacowned_part1 OWNER TO regress_vacuum;
SET ROLE regress_vacuum;
VACUUM vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM vacowned_part1;
VACUUM vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
ANALYZE vacowned_parted;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
ANALYZE vacowned_part1;
ANALYZE vacowned_part2;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_part1;
VACUUM (ANALYZE) vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
RESET ROLE;
-- Only one partition owned by other user.
ALTER TABLE vacowned_parted OWNER TO CURRENT_USER;
@@ -395,26 +389,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
ALTER TABLE vacowned_part1 OWNER TO CURRENT_USER;
SET ROLE regress_vacuum;
VACUUM vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM vacowned_part1;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
VACUUM vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
ANALYZE vacowned_parted;
-WARNING: permission denied to analyze "vacowned_part1", skipping it
-WARNING: permission denied to analyze "vacowned_part2", skipping it
ANALYZE vacowned_part1;
-WARNING: permission denied to analyze "vacowned_part1", skipping it
ANALYZE vacowned_part2;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_part1;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
VACUUM (ANALYZE) vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
RESET ROLE;
DROP TABLE vacowned;
DROP TABLE vacowned_parted;
diff --git a/src/test/regress/sql/cluster.sql b/src/test/regress/sql/cluster.sql
index 6cb9c926c0..a4cfaae807 100644
--- a/src/test/regress/sql/cluster.sql
+++ b/src/test/regress/sql/cluster.sql
@@ -145,7 +145,9 @@ INSERT INTO clstr_3 VALUES (1);
-- this user can only cluster clstr_1 and clstr_3, but the latter
-- has not been clustered
SET SESSION AUTHORIZATION regress_clstr_user;
+SET client_min_messages = ERROR; -- order of "skipping" warnings may vary
CLUSTER;
+RESET client_min_messages;
SELECT * FROM clstr_1 UNION ALL
SELECT * FROM clstr_2 UNION ALL
SELECT * FROM clstr_3;
--
2.25.1
--AqsLC8rIMeq19msA
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v5-0002-fix-MAINTAIN-privs-for-TOAST-tables.patch"
^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v6 1/2] fix MAINTAIN privs for partitions
@ 2023-01-09 22:29 Nathan Bossart <[email protected]>
0 siblings, 0 replies; 8+ messages in thread
From: Nathan Bossart @ 2023-01-09 22:29 UTC (permalink / raw)
---
doc/src/sgml/ref/analyze.sgml | 5 ++-
doc/src/sgml/ref/cluster.sgml | 17 ++++++---
doc/src/sgml/ref/lock.sgml | 5 ++-
doc/src/sgml/ref/reindex.sgml | 6 +++-
doc/src/sgml/ref/vacuum.sgml | 5 ++-
src/backend/commands/cluster.c | 35 +++++++++++++------
src/backend/commands/indexcmds.c | 22 ++++++------
src/backend/commands/lockcmds.c | 8 +++++
src/backend/commands/tablecmds.c | 30 ++++++++++++++--
src/backend/commands/vacuum.c | 9 +++--
src/include/commands/tablecmds.h | 1 +
.../expected/cluster-conflict-partition.out | 6 ++--
.../specs/cluster-conflict-partition.spec | 5 +--
src/test/regress/expected/cluster.out | 4 ++-
src/test/regress/expected/vacuum.out | 18 ----------
src/test/regress/sql/cluster.sql | 2 ++
16 files changed, 119 insertions(+), 59 deletions(-)
diff --git a/doc/src/sgml/ref/analyze.sgml b/doc/src/sgml/ref/analyze.sgml
index a26834da4f..2f94e89cb0 100644
--- a/doc/src/sgml/ref/analyze.sgml
+++ b/doc/src/sgml/ref/analyze.sgml
@@ -156,7 +156,10 @@ ANALYZE [ VERBOSE ] [ <replaceable class="parameter">table_and_columns</replacea
analyze all tables in their databases, except shared catalogs.
(The restriction for shared catalogs means that a true database-wide
<command>ANALYZE</command> can only be performed by superusers and roles
- with privileges of <literal>pg_maintain</literal>.)
+ with privileges of <literal>pg_maintain</literal>.) If a role has
+ permission to <command>ANALYZE</command> a partitioned table, it is also
+ permitted to <command>ANALYZE</command> each of its partitions, regardless
+ of whether the role has the aforementioned privileges on the partition.
<command>ANALYZE</command> will skip over any tables that the calling user
does not have permission to analyze.
</para>
diff --git a/doc/src/sgml/ref/cluster.sgml b/doc/src/sgml/ref/cluster.sgml
index 145101e6a5..b9f2acb1de 100644
--- a/doc/src/sgml/ref/cluster.sgml
+++ b/doc/src/sgml/ref/cluster.sgml
@@ -69,10 +69,7 @@ CLUSTER [VERBOSE]
<para>
<command>CLUSTER</command> without any parameter reclusters all the
previously-clustered tables in the current database that the calling user
- owns or has the <literal>MAINTAIN</literal> privilege for, or all such tables
- if called by a superuser or a role with privileges of the
- <link linkend="predefined-roles-table"><literal>pg_maintain</literal></link>
- role. This form of <command>CLUSTER</command> cannot be
+ has privileges for. This form of <command>CLUSTER</command> cannot be
executed inside a transaction block.
</para>
@@ -134,6 +131,18 @@ CLUSTER [VERBOSE]
<refsect1>
<title>Notes</title>
+ <para>
+ To cluster a table, one must have the <literal>MAINTAIN</literal> privilege
+ on the table or be the table's owner, a superuser, or a role with
+ privileges of the
+ <link linkend="predefined-roles-table"><literal>pg_maintain</literal></link>
+ role. If a role has permission to <command>CLUSTER</command> a partitioned
+ table, it is also permitted to <command>CLUSTER</command> each of its
+ partitions, regardless of whether the role has the aforementioned
+ privileges on the partition. <command>CLUSTER</command> will skip over any
+ tables that the calling user does not have permission to cluster.
+ </para>
+
<para>
In cases where you are accessing single rows randomly
within a table, the actual order of the data in the
diff --git a/doc/src/sgml/ref/lock.sgml b/doc/src/sgml/ref/lock.sgml
index 8524182211..5b3b2b793a 100644
--- a/doc/src/sgml/ref/lock.sgml
+++ b/doc/src/sgml/ref/lock.sgml
@@ -177,7 +177,10 @@ LOCK [ TABLE ] [ ONLY ] <replaceable class="parameter">name</replaceable> [ * ]
MODE</literal> (or a less-conflicting mode as described in <xref
linkend="explicit-locking"/>) is permitted. If a user has
<literal>SELECT</literal> privileges on the table, <literal>ACCESS SHARE
- MODE</literal> is permitted.
+ MODE</literal> is permitted. If a role has permission to lock a
+ partitioned table, it is also permitted to lock each of its partitions,
+ regardless of whether the role has the aforementioned privileges on the
+ partition.
</para>
<para>
diff --git a/doc/src/sgml/ref/reindex.sgml b/doc/src/sgml/ref/reindex.sgml
index 192513f34e..c6ad2546f9 100644
--- a/doc/src/sgml/ref/reindex.sgml
+++ b/doc/src/sgml/ref/reindex.sgml
@@ -306,7 +306,11 @@ REINDEX [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] { DA
indexes on shared catalogs will be skipped unless the user owns the
catalog (which typically won't be the case), has privileges of the
<literal>pg_maintain</literal> role, or has the <literal>MAINTAIN</literal>
- privilege on the catalog. Of course, superusers can always reindex anything.
+ privilege on the catalog. If a role has permission to
+ <command>REINDEX</command> a partitioned table, it is also permitted to
+ <command>REINDEX</command> each of its partitions, regardless of whether the
+ role has the aforementioned privileges on the partition. Of course,
+ superusers can always reindex anything.
</para>
<para>
diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml
index 8fa8421847..545b23b54f 100644
--- a/doc/src/sgml/ref/vacuum.sgml
+++ b/doc/src/sgml/ref/vacuum.sgml
@@ -401,7 +401,10 @@ VACUUM [ FULL ] [ FREEZE ] [ VERBOSE ] [ ANALYZE ] [ <replaceable class="paramet
vacuum all tables in their databases, except shared catalogs.
(The restriction for shared catalogs means that a true database-wide
<command>VACUUM</command> can only be performed by superusers and roles
- with privileges of <literal>pg_maintain</literal>.)
+ with privileges of <literal>pg_maintain</literal>.) If a role has
+ permission to <command>VACUUM</command> a partitioned table, it is also
+ permitted to <command>VACUUM</command> each of its partitions, regardless
+ of whether the role has the aforementioned privileges on the partition.
<command>VACUUM</command> will skip over any tables that the calling user
does not have permission to vacuum.
</para>
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index f11691aff7..369fea7c04 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -80,6 +80,7 @@ static void copy_table_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex,
static List *get_tables_to_cluster(MemoryContext cluster_context);
static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context,
Oid indexOid);
+static bool cluster_is_permitted_for_relation(Oid relid, Oid userid);
/*---------------------------------------------------------------------------
@@ -366,8 +367,7 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params)
if (recheck)
{
/* Check that the user still has privileges for the relation */
- if (!object_ownercheck(RelationRelationId, tableOid, save_userid) &&
- pg_class_aclcheck(tableOid, save_userid, ACL_MAINTAIN) != ACLCHECK_OK)
+ if (!cluster_is_permitted_for_relation(tableOid, save_userid))
{
relation_close(OldHeap, AccessExclusiveLock);
goto out;
@@ -1645,8 +1645,7 @@ get_tables_to_cluster(MemoryContext cluster_context)
index = (Form_pg_index) GETSTRUCT(indexTuple);
- if (!object_ownercheck(RelationRelationId, index->indrelid, GetUserId()) &&
- pg_class_aclcheck(index->indrelid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (!cluster_is_permitted_for_relation(index->indrelid, GetUserId()))
continue;
/* Use a permanent memory context for the result list */
@@ -1694,12 +1693,11 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
if (get_rel_relkind(indexrelid) != RELKIND_INDEX)
continue;
- /* Silently skip partitions which the user has no access to. */
- if (!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
- (!object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) ||
- IsSharedRelation(relid)))
- continue;
+ /*
+ * We already checked that the user has privileges to CLUSTER the
+ * partitioned table when we locked it earlier, so there's no need to
+ * check the privileges again here.
+ */
/* Use a permanent memory context for the result list */
old_context = MemoryContextSwitchTo(cluster_context);
@@ -1714,3 +1712,20 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
return rtcs;
}
+
+/*
+ * Return whether userid has privileges to CLUSTER relid. If not, this
+ * function emits a WARNING.
+ */
+static bool
+cluster_is_permitted_for_relation(Oid relid, Oid userid)
+{
+ if (pg_class_aclcheck(relid, userid, ACL_MAINTAIN) == ACLCHECK_OK ||
+ has_partition_ancestor_privs(relid, userid, ACL_MAINTAIN))
+ return true;
+
+ ereport(WARNING,
+ (errmsg("permission denied to cluster \"%s\", skipping it",
+ get_rel_name(relid))));
+ return false;
+}
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 8afc006f89..16ec0b114e 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -2796,9 +2796,9 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation,
/* Check permissions */
table_oid = IndexGetRelation(relId, true);
- if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
- OidIsValid(table_oid) &&
- pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (OidIsValid(table_oid) &&
+ pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(table_oid, GetUserId(), ACL_MAINTAIN))
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
relation->relname);
@@ -3008,16 +3008,16 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
/*
* The table can be reindexed if the user has been granted MAINTAIN on
- * the table or the user is a superuser, the table owner, or the
- * database/schema owner (but in the latter case, only if it's not a
- * shared relation). object_ownercheck includes the superuser case,
- * and depending on objectKind we already know that the user has
- * permission to run REINDEX on this database or schema per the
- * permission checks at the beginning of this routine.
+ * the table or one of its partition ancestors or the user is a
+ * superuser, the table owner, or the database/schema owner (but in the
+ * latter case, only if it's not a shared relation). pg_class_aclcheck
+ * includes the superuser case, and depending on objectKind we already
+ * know that the user has permission to run REINDEX on this database or
+ * schema per the permission checks at the beginning of this routine.
*/
if (classtuple->relisshared &&
- !object_ownercheck(RelationRelationId, relid, GetUserId()) &&
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(relid, GetUserId(), ACL_MAINTAIN))
continue;
/*
diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index 410d78b040..6bf1b815f0 100644
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -19,6 +19,7 @@
#include "catalog/namespace.h"
#include "catalog/pg_inherits.h"
#include "commands/lockcmds.h"
+#include "commands/tablecmds.h"
#include "miscadmin.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
@@ -305,5 +306,12 @@ LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)
aclresult = pg_class_aclcheck(reloid, userid, aclmask);
+ /*
+ * If this is a partition, check permissions of its ancestors if needed.
+ */
+ if (aclresult != ACLCHECK_OK &&
+ has_partition_ancestor_privs(reloid, userid, ACL_MAINTAIN))
+ aclresult = ACLCHECK_OK;
+
return aclresult;
}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1fbdad4b64..7c697a285b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16886,12 +16886,38 @@ RangeVarCallbackMaintainsTable(const RangeVar *relation,
errmsg("\"%s\" is not a table or materialized view", relation->relname)));
/* Check permissions */
- if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
- pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+ if (pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+ !has_partition_ancestor_privs(relId, GetUserId(), ACL_MAINTAIN))
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
relation->relname);
}
+/*
+ * If relid is a partition, returns whether userid has any of the privileges
+ * specified in acl on any of its ancestors. Otherwise, returns false.
+ */
+bool
+has_partition_ancestor_privs(Oid relid, Oid userid, AclMode acl)
+{
+ List *ancestors;
+ ListCell *lc;
+
+ if (!get_rel_relispartition(relid))
+ return false;
+
+ ancestors = get_partition_ancestors(relid);
+ foreach(lc, ancestors)
+ {
+ Oid ancestor = lfirst_oid(lc);
+
+ if (OidIsValid(ancestor) &&
+ pg_class_aclcheck(ancestor, userid, acl) == ACLCHECK_OK)
+ return true;
+ }
+
+ return false;
+}
+
/*
* Callback to RangeVarGetRelidExtended() for TRUNCATE processing.
*/
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ea1428dc8c..076b1e54eb 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -41,6 +41,7 @@
#include "catalog/pg_namespace.h"
#include "commands/cluster.h"
#include "commands/defrem.h"
+#include "commands/tablecmds.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
@@ -598,11 +599,13 @@ vacuum_is_permitted_for_relation(Oid relid, Form_pg_class reltuple,
* - the role owns the relation
* - the role owns the current database and the relation is not shared
* - the role has been granted the MAINTAIN privilege on the relation
+ * - the role has privileges to vacuum/analyze any of the relation's
+ * partition ancestors
*----------
*/
- if (object_ownercheck(RelationRelationId, relid, GetUserId()) ||
- (object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
- pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK)
+ if ((object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
+ pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK ||
+ has_partition_ancestor_privs(relid, GetUserId(), ACL_MAINTAIN))
return true;
relname = NameStr(reltuple->relname);
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 2e717fa815..e7c2b91a58 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -98,6 +98,7 @@ extern void AtEOSubXact_on_commit_actions(bool isCommit,
extern void RangeVarCallbackMaintainsTable(const RangeVar *relation,
Oid relId, Oid oldRelId,
void *arg);
+extern bool has_partition_ancestor_privs(Oid relid, Oid userid, AclMode acl);
extern void RangeVarCallbackOwnsRelation(const RangeVar *relation,
Oid relId, Oid oldRelId, void *arg);
diff --git a/src/test/isolation/expected/cluster-conflict-partition.out b/src/test/isolation/expected/cluster-conflict-partition.out
index 7acb675c97..8d21276996 100644
--- a/src/test/isolation/expected/cluster-conflict-partition.out
+++ b/src/test/isolation/expected/cluster-conflict-partition.out
@@ -22,14 +22,16 @@ starting permutation: s1_begin s1_lock_child s2_auth s2_cluster s1_commit s2_res
step s1_begin: BEGIN;
step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
step s2_auth: SET ROLE regress_cluster_part;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
step s1_commit: COMMIT;
+step s2_cluster: <... completed>
step s2_reset: RESET ROLE;
starting permutation: s1_begin s2_auth s1_lock_child s2_cluster s1_commit s2_reset
step s1_begin: BEGIN;
step s2_auth: SET ROLE regress_cluster_part;
step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
step s1_commit: COMMIT;
+step s2_cluster: <... completed>
step s2_reset: RESET ROLE;
diff --git a/src/test/isolation/specs/cluster-conflict-partition.spec b/src/test/isolation/specs/cluster-conflict-partition.spec
index 5091f684a9..ae38cb4ee3 100644
--- a/src/test/isolation/specs/cluster-conflict-partition.spec
+++ b/src/test/isolation/specs/cluster-conflict-partition.spec
@@ -27,11 +27,8 @@ step s2_auth { SET ROLE regress_cluster_part; }
step s2_cluster { CLUSTER cluster_part_tab USING cluster_part_ind; }
step s2_reset { RESET ROLE; }
-# CLUSTER on the parent waits if locked, passes for all cases.
+# CLUSTER waits if locked, passes for all cases.
permutation s1_begin s1_lock_parent s2_auth s2_cluster s1_commit s2_reset
permutation s1_begin s2_auth s1_lock_parent s2_cluster s1_commit s2_reset
-
-# When taking a lock on a partition leaf, CLUSTER on the parent skips
-# the leaf, passes for all cases.
permutation s1_begin s1_lock_child s2_auth s2_cluster s1_commit s2_reset
permutation s1_begin s2_auth s1_lock_child s2_cluster s1_commit s2_reset
diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expected/cluster.out
index 542c2e098c..2eec483eaa 100644
--- a/src/test/regress/expected/cluster.out
+++ b/src/test/regress/expected/cluster.out
@@ -352,7 +352,9 @@ INSERT INTO clstr_3 VALUES (1);
-- this user can only cluster clstr_1 and clstr_3, but the latter
-- has not been clustered
SET SESSION AUTHORIZATION regress_clstr_user;
+SET client_min_messages = ERROR; -- order of "skipping" warnings may vary
CLUSTER;
+RESET client_min_messages;
SELECT * FROM clstr_1 UNION ALL
SELECT * FROM clstr_2 UNION ALL
SELECT * FROM clstr_3;
@@ -513,7 +515,7 @@ SELECT a.relname, a.relfilenode=b.relfilenode FROM pg_class a
-----------+----------
ptnowner | t
ptnowner1 | f
- ptnowner2 | t
+ ptnowner2 | f
(3 rows)
DROP TABLE ptnowner;
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index d860be0e20..458adee7f8 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -353,20 +353,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
ALTER TABLE vacowned_part1 OWNER TO regress_vacuum;
SET ROLE regress_vacuum;
VACUUM vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM vacowned_part1;
VACUUM vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
ANALYZE vacowned_parted;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
ANALYZE vacowned_part1;
ANALYZE vacowned_part2;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_part1;
VACUUM (ANALYZE) vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
RESET ROLE;
-- Only one partition owned by other user.
ALTER TABLE vacowned_parted OWNER TO CURRENT_USER;
@@ -395,26 +389,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
ALTER TABLE vacowned_part1 OWNER TO CURRENT_USER;
SET ROLE regress_vacuum;
VACUUM vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM vacowned_part1;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
VACUUM vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
ANALYZE vacowned_parted;
-WARNING: permission denied to analyze "vacowned_part1", skipping it
-WARNING: permission denied to analyze "vacowned_part2", skipping it
ANALYZE vacowned_part1;
-WARNING: permission denied to analyze "vacowned_part1", skipping it
ANALYZE vacowned_part2;
-WARNING: permission denied to analyze "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_parted;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
VACUUM (ANALYZE) vacowned_part1;
-WARNING: permission denied to vacuum "vacowned_part1", skipping it
VACUUM (ANALYZE) vacowned_part2;
-WARNING: permission denied to vacuum "vacowned_part2", skipping it
RESET ROLE;
DROP TABLE vacowned;
DROP TABLE vacowned_parted;
diff --git a/src/test/regress/sql/cluster.sql b/src/test/regress/sql/cluster.sql
index 6cb9c926c0..a4cfaae807 100644
--- a/src/test/regress/sql/cluster.sql
+++ b/src/test/regress/sql/cluster.sql
@@ -145,7 +145,9 @@ INSERT INTO clstr_3 VALUES (1);
-- this user can only cluster clstr_1 and clstr_3, but the latter
-- has not been clustered
SET SESSION AUTHORIZATION regress_clstr_user;
+SET client_min_messages = ERROR; -- order of "skipping" warnings may vary
CLUSTER;
+RESET client_min_messages;
SELECT * FROM clstr_1 UNION ALL
SELECT * FROM clstr_2 UNION ALL
SELECT * FROM clstr_3;
--
2.25.1
--sdtB3X0nJg68CQEu
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v6-0002-fix-MAINTAIN-privs-for-TOAST-tables.patch"
^ permalink raw reply [nested|flat] 8+ messages in thread
* Reduce useless changes before reassembly during logical replication
@ 2024-01-17 06:15 li jie <[email protected]>
2024-01-18 06:42 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
0 siblings, 1 reply; 8+ messages in thread
From: li jie @ 2024-01-17 06:15 UTC (permalink / raw)
To: PostgreSQL Hackers <[email protected]>; Amit Kapila <[email protected]>
Hi hackers,
During logical replication, if there is a large write transaction, some
spill files will be written to disk, depending on the setting of
logical_decoding_work_mem.
This behavior can effectively avoid OOM, but if the transaction
generates a lot of change before commit, a large number of files may
fill the disk. For example, you can update a TB-level table.
However, I found an inelegant phenomenon. If the modified large table is not
published, its changes will also be written with a large number of spill files.
Look at an example below:
publisher:
```
create table tbl_pub(id int, val1 text, val2 text,val3 text);
create table tbl_t1(id int, val1 text, val2 text,val3 text);
CREATE PUBLICATION mypub FOR TABLE public.tbl_pub;
```
subscriber:
```
create table tbl_pub(id int, val1 text, val2 text,val3 text);
create table tbl_t1(id int, val1 text, val2 text,val3 text);
CREATE SUBSCRIPTION mysub CONNECTION 'host=127.0.0.1 port=5432
user=postgres dbname=postgres' PUBLICATION mypub;
```
publisher:
```
begin;
insert into tbl_t1 select i,repeat('xyzzy', i),repeat('abcba',
i),repeat('dfds', i) from generate_series(0,999999) i;
```
Later you will see a large number of spill files in the
"/$PGDATA/pg_replslot/mysub/" directory.
```
$ll -sh
total 4.5G
4.0K -rw------- 1 postgres postgres 200 Nov 30 09:24 state
17M -rw------- 1 postgres postgres 17M Nov 30 08:22 xid-750-lsn-0-10000000.spill
12M -rw------- 1 postgres postgres 12M Nov 30 08:20 xid-750-lsn-0-1000000.spill
17M -rw------- 1 postgres postgres 17M Nov 30 08:23 xid-750-lsn-0-11000000.spill
......
```
We can see that table tbl_t1 is not published in mypub. It also won't be sent
downstream because it's not subscribed.
After the transaction is reorganized, the pgoutput decoding plugin filters out
changes to these unpublished relationships when sending logical changes.
See function pgoutput_change.
Most importantly, if we filter out unpublished relationship-related
changes after constructing the changes but before queuing the changes
into a transaction, will it reduce the workload of logical decoding
and avoid disk
or memory growth as much as possible?
The patch in the attachment is a prototype, which can effectively reduce the
memory and disk space usage during logical replication.
Design:
1. Added a callback LogicalDecodeFilterByRelCB for the output plugin.
2. Added this callback function pgoutput_table_filter for the pgoutput plugin.
Its main implementation is based on the table filter in the
pgoutput_change function.
Its main function is to determine whether the change needs to be published based
on the parameters of the publication, and if not, filter it.
3. After constructing a change and before Queue a change into a transaction,
use RelidByRelfilenumber to obtain the relation associated with the change,
just like obtaining the relation in the ReorderBufferProcessTXN function.
4. Relation may be a toast, and there is no good way to get its real
table relation based on toast relation. Here, I get the real table oid
through toast relname, and then get the real table relation.
5. This filtering takes into account INSERT/UPDATE/INSERT. Other
changes have not been considered yet and can be expanded in the future.
Test:
1. Added a test case 034_table_filter.pl
2. Like the case above, create two tables, the published table tbl_pub and
the non-published table tbl_t1
3. Insert 10,000 rows of toast data into tbl_t1 on the publisher, and use
pg_ls_replslotdir to record the total size of the slot directory
every second.
4. Compare the size of the slot directory at the beginning of the
transaction(size1),
the size at the end of the transaction (size2), and the average
size of the entire process(size3).
5. Assert(size1==size2==size3)
Sincerely look forward to your feedback.
Regards, lijie
Attachments:
[application/octet-stream] v1-Reduce-useless-changes-before-reassembly-during-logi.patch (18.6K, ../../CAGfChW62f5NTNbLsqO-6_CrmKPqBEQtWPcPDafu8pCwZznk=xw@mail.gmail.com/2-v1-Reduce-useless-changes-before-reassembly-during-logi.patch)
download | inline diff:
From 60ecc56984f0c2105a9eee24f2f73d05ee6ff1f3 Mon Sep 17 00:00:00 2001
From: "adger.lj" <[email protected]>
Date: Thu, 7 Dec 2023 15:51:12 +0800
Subject: [PATCH] Reduce useless changes before reassembly during logical
replication
In order to reduce unnecessary logical replication, irrelevant relationship
changes can be filtered out before reorganizing transaction fragments.
This can effectively reduce useless changes and prevent storage space from
being filled up with irrelevant data.
By design, Added a callback LogicalDecodeFilterByRelCB for the output plugin.
We implemented a function pgoutput_table_filter for pgoutput. And RelationSyncCache
is reused to determine whether a relationship-related change should be filtered out.
referring to the implementation of the function pgoutput_change, currently only
insert/update/delete is can filtered, and other types of changes are not considered.
Perhaps more detailed analysis can be done and more filters can be filtered.
Most of the code in the FilterByTable function is transplanted from the ReorderBufferProcessTXN
function, which can be called before the ReorderBufferQueueChange function.It is
also the encapsulation of the callback function filter_by_table_cb in logical.c.
In general, this patch concentrates the filtering of changes in the ReorderBufferProcessTXN
function and the pgoutput_change function into the FilterByTable function, and calls
it before the ReorderBufferQueueChange function.
---
src/backend/replication/logical/decode.c | 153 +++++++++++++++++++-
src/backend/replication/logical/logical.c | 31 ++++
src/backend/replication/pgoutput/pgoutput.c | 89 +++++++-----
src/include/replication/logical.h | 2 +
src/include/replication/output_plugin.h | 7 +
src/test/subscription/t/034_table_filter.pl | 91 ++++++++++++
6 files changed, 335 insertions(+), 38 deletions(-)
create mode 100644 src/test/subscription/t/034_table_filter.pl
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index b3f8f908d1..47cda524bd 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -35,6 +35,7 @@
#include "access/xlogrecord.h"
#include "access/xlogutils.h"
#include "catalog/pg_control.h"
+#include "common/string.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
@@ -42,6 +43,7 @@
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/standby.h"
+#include "utils/relfilenumbermap.h"
/* individual record(group)'s handlers */
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
@@ -155,7 +157,7 @@ xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
case XLOG_PARAMETER_CHANGE:
{
xl_parameter_change *xlrec =
- (xl_parameter_change *) XLogRecGetData(buf->record);
+ (xl_parameter_change *) XLogRecGetData(buf->record);
/*
* If wal_level on the primary is reduced to less than
@@ -581,6 +583,134 @@ FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
return filter_by_origin_cb_wrapper(ctx, origin_id);
}
+static bool
+FilterByTable(LogicalDecodingContext *ctx, ReorderBufferChange *change)
+{
+ ReorderBuffer *rb = ctx->reorder;
+ Relation relation = NULL;
+ Oid reloid;
+ bool result = false;
+ bool using_subtxn;
+
+ if (ctx->callbacks.filter_by_table_cb == NULL)
+ return false;
+
+ switch (change->action)
+ {
+ /* intentionally fall through */
+ case REORDER_BUFFER_CHANGE_INSERT:
+ case REORDER_BUFFER_CHANGE_UPDATE:
+ case REORDER_BUFFER_CHANGE_DELETE:
+ break;
+ default:
+ return false;
+ }
+
+ /*
+ * Decoding needs access to syscaches et al., which in turn use
+ * heavyweight locks and such. Thus we need to have enough state around to
+ * keep track of those. The easiest way is to simply use a transaction
+ * internally. That also allows us to easily enforce that nothing writes
+ * to the database by checking for xid assignments.
+ *
+ * When we're called via the SQL SRF there's already a transaction
+ * started, so start an explicit subtransaction there.
+ */
+ using_subtxn = IsTransactionOrTransactionBlock();
+
+ if (using_subtxn)
+ BeginInternalSubTransaction("filter change by table");
+ else
+ StartTransactionCommand();
+
+ reloid = RelidByRelfilenumber(change->data.tp.rlocator.spcOid,
+ change->data.tp.rlocator.relNumber);
+ if (reloid == InvalidOid)
+ {
+ result = true;
+ goto filter_done;
+ }
+
+ relation = RelationIdGetRelation(reloid);
+
+ if (!RelationIsValid(relation))
+ elog(ERROR, "could not open relation with OID %u (for filenumber \"%s\")",
+ reloid,
+ relpathperm(change->data.tp.rlocator,
+ MAIN_FORKNUM));
+
+ if (!RelationIsLogicallyLogged(relation))
+ {
+ result = true;
+ goto filter_done;
+ }
+
+ /*
+ * Ignore temporary heaps created during DDL unless the plugin has asked
+ * for them.
+ */
+ if (relation->rd_rel->relrewrite && !rb->output_rewrites)
+ {
+ result = true;
+ goto filter_done;
+ }
+
+ /*
+ * For now ignore sequence changes entirely. Most of the time they don't
+ * log changes using records we understand, so it doesn't make sense to
+ * handle the few cases we do.
+ */
+ if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
+ {
+ result = true;
+ goto filter_done;
+ }
+
+ if (IsToastRelation(relation))
+ {
+ Oid real_reloid = InvalidOid;
+
+ /* pg_toast_ len is 9 */
+ char *toast_name = RelationGetRelationName(relation);
+ char *start_ch = &toast_name[9];
+
+ real_reloid = strtoint(start_ch, NULL, 10);
+
+ if (real_reloid == InvalidOid)
+ elog(ERROR, "cannot get the real table oid for toast table %s, error: %m", toast_name);
+
+ RelationClose(relation);
+
+ relation = RelationIdGetRelation(real_reloid);
+
+ if (!RelationIsValid(relation))
+ elog(ERROR, "could not open real relation with OID %u (for toast table filenumber \"%s\")",
+ reloid,
+ relpathperm(change->data.tp.rlocator,
+ MAIN_FORKNUM));
+ }
+
+ result = filter_by_table_cb_wrapper(ctx, relation, change);
+
+filter_done:
+
+ if (result && RelationIsValid(relation))
+ elog(DEBUG1, "logical filter change by table %s", RelationGetRelationName(relation));
+
+ if (RelationIsValid(relation))
+ RelationClose(relation);
+
+ AbortCurrentTransaction();
+
+ if (using_subtxn)
+ RollbackAndReleaseCurrentSubTransaction();
+
+ if (result)
+ ReorderBufferReturnChange(rb, change, false);
+
+ return result;
+}
+
/*
* Handle rmgr LOGICALMSG_ID records for LogicalDecodingProcessRecord().
*/
@@ -940,6 +1070,9 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change->data.tp.clear_toast_afterwards = true;
+ if (FilterByTable(ctx, change))
+ return;
+
ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
change,
xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
@@ -1009,6 +1142,9 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change->data.tp.clear_toast_afterwards = true;
+ if (FilterByTable(ctx, change))
+ return;
+
ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
change, false);
}
@@ -1065,6 +1201,9 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change->data.tp.clear_toast_afterwards = true;
+ if (FilterByTable(ctx, change))
+ return;
+
ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
change, false);
}
@@ -1201,11 +1340,14 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
else
change->data.tp.clear_toast_afterwards = false;
- ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
- buf->origptr, change, false);
-
/* move to the next xl_multi_insert_tuple entry */
data += datalen;
+
+ if (FilterByTable(ctx, change))
+ continue;;
+
+ ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
+ buf->origptr, change, false);
}
Assert(data == tupledata + tuplelen);
}
@@ -1240,6 +1382,9 @@ DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change->data.tp.clear_toast_afterwards = true;
+ if (FilterByTable(ctx, change))
+ return;
+
ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
change, false);
}
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index ca09c683f1..8bd0cefb22 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -1225,6 +1225,37 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
return ret;
}
+bool
+filter_by_table_cb_wrapper(LogicalDecodingContext *ctx, Relation relation, ReorderBufferChange *change)
+{
+ LogicalErrorCallbackState state;
+ ErrorContextCallback errcallback;
+ bool ret;
+
+ Assert(!ctx->fast_forward);
+
+ /* Push callback + info on the error context stack */
+ state.ctx = ctx;
+ state.callback_name = "filter_by_table";
+ state.report_location = InvalidXLogRecPtr;
+ errcallback.callback = output_plugin_error_callback;
+ errcallback.arg = (void *) &state;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /* set output state */
+ ctx->accept_writes = false;
+ ctx->end_xact = false;
+
+ /* do the actual work: call callback */
+ ret = ctx->callbacks.filter_by_table_cb(ctx, relation, change);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+
+ return ret;
+}
+
static void
message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
XLogRecPtr message_lsn, bool transactional,
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 425238187f..2966258ea3 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -57,6 +57,9 @@ static void pgoutput_message(LogicalDecodingContext *ctx,
Size sz, const char *message);
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
RepOriginId origin_id);
+static bool pgoutput_table_filter(struct LogicalDecodingContext *ctx,
+ Relation relation,
+ ReorderBufferChange *change);
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
static void pgoutput_prepare_txn(LogicalDecodingContext *ctx,
@@ -259,6 +262,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn;
cb->filter_by_origin_cb = pgoutput_origin_filter;
+ cb->filter_by_table_cb = pgoutput_table_filter;
cb->shutdown_cb = pgoutput_shutdown;
/* transaction streaming */
@@ -1415,9 +1419,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
- if (!is_publishable_relation(relation))
- return;
-
/*
* Remember the xid for the change in streaming mode. We need to send xid
* with each change in the streaming mode so that subscriber can make
@@ -1428,37 +1429,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
xid = change->txn->xid;
relentry = get_rel_sync_entry(data, relation);
-
- /* First check the table filter */
- switch (action)
- {
- case REORDER_BUFFER_CHANGE_INSERT:
- if (!relentry->pubactions.pubinsert)
- return;
- break;
- case REORDER_BUFFER_CHANGE_UPDATE:
- if (!relentry->pubactions.pubupdate)
- return;
- break;
- case REORDER_BUFFER_CHANGE_DELETE:
- if (!relentry->pubactions.pubdelete)
- return;
-
- /*
- * This is only possible if deletes are allowed even when replica
- * identity is not defined for a table. Since the DELETE action
- * can't be published, we simply return.
- */
- if (!change->data.tp.oldtuple)
- {
- elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
- return;
- }
- break;
- default:
- Assert(false);
- }
-
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
@@ -1684,6 +1654,57 @@ pgoutput_origin_filter(LogicalDecodingContext *ctx,
return false;
}
+/*
+ * Return true if the relation has not been published, false otherwise.
+ */
+static bool
+pgoutput_table_filter(struct LogicalDecodingContext *ctx,
+ Relation relation,
+ ReorderBufferChange *change)
+{
+ PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+ RelationSyncEntry *relentry;
+ ReorderBufferChangeType action = change->action;
+
+ if (!is_publishable_relation(relation))
+ return true;
+
+ relentry = get_rel_sync_entry(data, relation);
+
+ /* First check the table filter */
+ switch (action)
+ {
+ case REORDER_BUFFER_CHANGE_INSERT:
+ if (!relentry->pubactions.pubinsert)
+ return true;
+ break;
+ case REORDER_BUFFER_CHANGE_UPDATE:
+ if (!relentry->pubactions.pubupdate)
+ return true;
+ break;
+ case REORDER_BUFFER_CHANGE_DELETE:
+ if (!relentry->pubactions.pubdelete)
+ return true;
+
+ /*
+ * This is only possible if deletes are allowed even when replica
+ * identity is not defined for a table. Since the DELETE action
+ * can't be published, we simply return.
+ */
+ if (!change->data.tp.oldtuple)
+ {
+ elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
+ return true;
+ }
+ break;
+ default:
+ Assert(false);
+ }
+
+ return false;
+}
+
+
/*
* Shutdown the output plugin.
*
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index dc2df4ce92..30a16cd784 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -145,6 +145,8 @@ extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
extern bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx,
TransactionId xid, const char *gid);
extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id);
+extern bool filter_by_table_cb_wrapper(LogicalDecodingContext *ctx, Relation relation,
+ ReorderBufferChange *change);
extern void ResetLogicalStreamingState(void);
extern void UpdateDecodingStats(LogicalDecodingContext *ctx);
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index 44988ebdd8..030eb5afb7 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -96,6 +96,12 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx,
typedef bool (*LogicalDecodeFilterByOriginCB) (struct LogicalDecodingContext *ctx,
RepOriginId origin_id);
+/*
+ * Filter changes by table.
+ */
+typedef bool (*LogicalDecodeFilterByRelCB) (struct LogicalDecodingContext *ctx,
+ Relation relation, ReorderBufferChange *change);
+
/*
* Called to shutdown an output plugin.
*/
@@ -222,6 +228,7 @@ typedef struct OutputPluginCallbacks
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
LogicalDecodeFilterByOriginCB filter_by_origin_cb;
+ LogicalDecodeFilterByRelCB filter_by_table_cb;
LogicalDecodeShutdownCB shutdown_cb;
/* streaming of changes at prepare time */
diff --git a/src/test/subscription/t/034_table_filter.pl b/src/test/subscription/t/034_table_filter.pl
new file mode 100644
index 0000000000..8d4f962adc
--- /dev/null
+++ b/src/test/subscription/t/034_table_filter.pl
@@ -0,0 +1,91 @@
+# Copyright (c) 2021-2023, PostgreSQL Global Development Group
+
+# Basic logical replication test
+use strict;
+use warnings;
+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', 'logical_decoding_work_mem = 64kB');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+
+# Create some preexisting content on publisher
+$node_publisher->safe_psql('postgres',
+ "create table tbl_pub(id int, val1 text, val2 text,size int);");
+$node_publisher->safe_psql('postgres',
+ "create table tbl_t1(id int, val1 text, val2 text,size int);");
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION mypub FOR TABLE public.tbl_pub;");
+$node_publisher->safe_psql('postgres',
+qq(
+CREATE OR REPLACE FUNCTION check_replication_status() RETURNS VOID AS \$\$
+DECLARE
+ replication_record pg_stat_replication;
+BEGIN
+ LOOP
+ SELECT *
+ INTO replication_record
+ FROM pg_stat_replication
+ WHERE application_name = 'mysub';
+
+ IF replication_record.replay_lsn = replication_record.write_lsn THEN
+ EXIT;
+ END IF;
+
+ PERFORM pg_sleep(1);
+ END LOOP;
+END;
+\$\$ LANGUAGE plpgsql;));
+
+# Create some preexisting content on subscriber
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_subscriber->safe_psql('postgres',
+ "create table tbl_pub(id int, val1 text, val2 text,size text);");
+$node_subscriber->safe_psql('postgres',
+ "create table tbl_t1(id int, val1 text, val2 text,size text);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION mysub CONNECTION '$publisher_connstr' PUBLICATION mypub"
+);
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'mysub');
+
+# test filter
+$node_publisher->safe_psql('postgres',
+qq(BEGIN;
+insert into tbl_t1 select 1, 'xyzzy', 'abcba', sum(size) from pg_ls_replslotdir('mysub');
+insert into tbl_t1 select i,repeat('xyzzy', i),repeat('abcba',i),(select sum(size) from pg_ls_replslotdir('mysub')) from generate_series(2,9999) i;
+update tbl_t1 set val2 = repeat('xyzzy',id) where id > 1 and id < 10001;
+select check_replication_status();
+insert into tbl_t1 select 10001, 'xyzzy', 'abcba', sum(size) from pg_ls_replslotdir('mysub');
+COMMIT;)
+);
+
+my $minsize = $node_publisher->safe_psql('postgres',
+ "select size from tbl_t1 order by size asc limit 1;");
+my $maxsize = $node_publisher->safe_psql('postgres',
+ "select size from tbl_t1 order by size desc limit 1;");
+is($minsize, $maxsize, 'check decode filter table between maxsize and minsize');
+
+
+my $fristrow = $node_publisher->safe_psql('postgres',
+ "select size from tbl_t1 where id = 1;");
+my $lastrow = $node_publisher->safe_psql('postgres',
+ "select size from tbl_t1 where id = 10001;");
+is($minsize, $maxsize, 'check decode filter table between fristrow and lastrow');
+
+is($minsize, $lastrow, 'check decode filter table between minsize and lastrow');
+
+print "minsize: " . $minsize . "maxsize: " . $maxsize ."fristrow: " . $fristrow ."lastrow: " . $lastrow . "\n";
+
+done_testing();
\ No newline at end of file
--
2.39.3
^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Reduce useless changes before reassembly during logical replication
2024-01-17 06:15 Reduce useless changes before reassembly during logical replication li jie <[email protected]>
@ 2024-01-18 06:42 ` Bharath Rupireddy <[email protected]>
2024-01-18 09:17 ` Re: Reduce useless changes before reassembly during logical replication Amit Kapila <[email protected]>
0 siblings, 1 reply; 8+ messages in thread
From: Bharath Rupireddy @ 2024-01-18 06:42 UTC (permalink / raw)
To: li jie <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>; Amit Kapila <[email protected]>
On Wed, Jan 17, 2024 at 11:45 AM li jie <[email protected]> wrote:
>
> Hi hackers,
>
> During logical replication, if there is a large write transaction, some
> spill files will be written to disk, depending on the setting of
> logical_decoding_work_mem.
>
> This behavior can effectively avoid OOM, but if the transaction
> generates a lot of change before commit, a large number of files may
> fill the disk. For example, you can update a TB-level table.
>
> However, I found an inelegant phenomenon. If the modified large table is not
> published, its changes will also be written with a large number of spill files.
> Look at an example below:
Thanks. I agree that decoding and queuing the changes of unpublished
tables' data into reorder buffer is an unnecessary task for walsender.
It takes processing efforts (CPU overhead), consumes disk space and
uses memory configured via logical_decoding_work_mem for a replication
connection inefficiently.
> Later you will see a large number of spill files in the
>
> We can see that table tbl_t1 is not published in mypub. It also won't be sent
> downstream because it's not subscribed.
> After the transaction is reorganized, the pgoutput decoding plugin filters out
> changes to these unpublished relationships when sending logical changes.
> See function pgoutput_change.
Right. Here's my testing [1].
> Most importantly, if we filter out unpublished relationship-related
> changes after constructing the changes but before queuing the changes
> into a transaction, will it reduce the workload of logical decoding
> and avoid disk
> or memory growth as much as possible?
Right. It can.
> The patch in the attachment is a prototype, which can effectively reduce the
> memory and disk space usage during logical replication.
>
> Design:
> 1. Added a callback LogicalDecodeFilterByRelCB for the output plugin.
>
> 2. Added this callback function pgoutput_table_filter for the pgoutput plugin.
> Its main implementation is based on the table filter in the
> pgoutput_change function.
> Its main function is to determine whether the change needs to be published based
> on the parameters of the publication, and if not, filter it.
>
> 3. After constructing a change and before Queue a change into a transaction,
> use RelidByRelfilenumber to obtain the relation associated with the change,
> just like obtaining the relation in the ReorderBufferProcessTXN function.
>
> 4. Relation may be a toast, and there is no good way to get its real
> table relation based on toast relation. Here, I get the real table oid
> through toast relname, and then get the real table relation.
>
> 5. This filtering takes into account INSERT/UPDATE/INSERT. Other
> changes have not been considered yet and can be expanded in the future.
Design of this patch is based on the principle of logical decoding
filtering things out early on and looks very similar to
filter_prepare_cb_wrapper/pg_decode_filter_prepare and
filter_by_origin_cb/pgoutput_origin_filter. Per my understanding this
design looks okay unless I'm missing anything.
> Test:
> 1. Added a test case 034_table_filter.pl
> 2. Like the case above, create two tables, the published table tbl_pub and
> the non-published table tbl_t1
> 3. Insert 10,000 rows of toast data into tbl_t1 on the publisher, and use
> pg_ls_replslotdir to record the total size of the slot directory
> every second.
> 4. Compare the size of the slot directory at the beginning of the
> transaction(size1),
> the size at the end of the transaction (size2), and the average
> size of the entire process(size3).
> 5. Assert(size1==size2==size3)
I bet that the above test with 10K rows is going to take a noticeable
time on some buildfarm members (it took 6 seconds on my dev system
which is an AWS EC2 instance). And, the above test can get flaky.
Therefore, IMO, the concrete way of testing this feature is by looking
at the server logs for the following message using
PostgreSQL::Test::Cluster log_contains().
+filter_done:
+
+ if (result && RelationIsValid(relation))
+ elog(DEBUG1, "logical filter change by table %s",
RelationGetRelationName(relation));
+
Here are some comments on the v1 patch:
1.
@@ -1415,9 +1419,6 @@ pgoutput_change(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
- if (!is_publishable_relation(relation))
- return;
-
Instead of removing is_publishable_relation from pgoutput_change, I
think it can just be turned into an assertion
Assert(is_publishable_relation(relation));, no?
2.
+ switch (change->action)
+ {
+ /* intentionally fall through */
Perhaps, it must use /* FALLTHROUGH */ just like elsewhere in the
code, otherwise a warning is thrown.
3. From commit message:
Most of the code in the FilterByTable function is transplanted from
the ReorderBufferProcessTXN
function, which can be called before the ReorderBufferQueueChange function.It is
I think the above note can just be above the FilterByTable function
for better understanding.
+static bool
+FilterByTable(LogicalDecodingContext *ctx, ReorderBufferChange *change)
+{
4. Why is FilterByTable(ctx, change) call placed after DecodeXLogTuple
in DecodeInsert, DecodeUpdate and DecodeDelete? Is there a use for
decoded tuples done by DecodeXLogTuple in the new callback
filter_by_table_cb? If not, can we move FilterByTable call before
DecodeXLogTuple to avoid some more extra processing?
5. Why is ReorderBufferChange needed as a parameter to FilterByTable
and filter_by_table_cb? Can't just the LogicalDecodingContext and
relation name, the change action be enough to decide if the table is
publishable or not? If done this way, it can avoid some more
processing, no?
6. Please run pgindent and pgperltidy on the new source code and new
TAP test file respectively.
[1]
HEAD:
postgres=# BEGIN;
BEGIN
Time: 0.110 ms
postgres=*# insert into tbl_t1 select i,repeat('xyzzy',
i),repeat('abcba', i),repeat('dfds', i) from generate_series(0,99999)
i;
INSERT 0 100000
Time: 379488.265 ms (06:19.488)
postgres=*#
ubuntu:~/postgres/pg17/bin$ du -sh
/home/ubuntu/postgres/pg17/bin/db17/pg_replslot/mysub
837M /home/ubuntu/postgres/pg17/bin/db17/pg_replslot/mysub
ubuntu:~/postgres/pg17/bin$ du -sh /home/ubuntu/postgres/pg17/bin/db17
2.6G /home/ubuntu/postgres/pg17/bin/db17
PATCHED:
postgres=# BEGIN;
BEGIN
Time: 0.105 ms
postgres=*# insert into tbl_t1 select i,repeat('xyzzy',
i),repeat('abcba', i),repeat('dfds', i) from generate_series(0,99999)
i;
INSERT 0 100000
Time: 380044.554 ms (06:20.045)
ubuntu:~/postgres$ du -sh /home/ubuntu/postgres/pg17/bin/db17/pg_replslot/mysub
8.0K /home/ubuntu/postgres/pg17/bin/db17/pg_replslot/mysub
ubuntu:~/postgres$ du -sh /home/ubuntu/postgres/pg17/bin/db17
1.8G /home/ubuntu/postgres/pg17/bin/db17
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Reduce useless changes before reassembly during logical replication
2024-01-17 06:15 Reduce useless changes before reassembly during logical replication li jie <[email protected]>
2024-01-18 06:42 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
@ 2024-01-18 09:17 ` Amit Kapila <[email protected]>
2024-01-18 11:14 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
2024-02-22 08:11 ` Re: Reduce useless changes before reassembly during logical replication Andy Fan <[email protected]>
0 siblings, 2 replies; 8+ messages in thread
From: Amit Kapila @ 2024-01-18 09:17 UTC (permalink / raw)
To: Bharath Rupireddy <[email protected]>; +Cc: li jie <[email protected]>; PostgreSQL Hackers <[email protected]>
On Thu, Jan 18, 2024 at 12:12 PM Bharath Rupireddy
<[email protected]> wrote:
>
> On Wed, Jan 17, 2024 at 11:45 AM li jie <[email protected]> wrote:
> >
> > Hi hackers,
> >
> > During logical replication, if there is a large write transaction, some
> > spill files will be written to disk, depending on the setting of
> > logical_decoding_work_mem.
> >
> > This behavior can effectively avoid OOM, but if the transaction
> > generates a lot of change before commit, a large number of files may
> > fill the disk. For example, you can update a TB-level table.
> >
> > However, I found an inelegant phenomenon. If the modified large table is not
> > published, its changes will also be written with a large number of spill files.
> > Look at an example below:
>
> Thanks. I agree that decoding and queuing the changes of unpublished
> tables' data into reorder buffer is an unnecessary task for walsender.
> It takes processing efforts (CPU overhead), consumes disk space and
> uses memory configured via logical_decoding_work_mem for a replication
> connection inefficiently.
>
This is all true but note that in successful cases (where the table is
published) all the work done by FilterByTable(accessing caches,
transaction-related stuff) can add noticeable overhead as anyway we do
that later in pgoutput_change(). I think I gave the same comment
earlier as well but didn't see any satisfactory answer or performance
data for successful cases to back this proposal. Note, users can
configure to stream_in_progress transactions in which case they
shouldn't see such a big problem. However, I agree that if we can find
some solution where there is no noticeable overhead then that would be
worth considering.
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Reduce useless changes before reassembly during logical replication
2024-01-17 06:15 Reduce useless changes before reassembly during logical replication li jie <[email protected]>
2024-01-18 06:42 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
2024-01-18 09:17 ` Re: Reduce useless changes before reassembly during logical replication Amit Kapila <[email protected]>
@ 2024-01-18 11:14 ` Bharath Rupireddy <[email protected]>
1 sibling, 0 replies; 8+ messages in thread
From: Bharath Rupireddy @ 2024-01-18 11:14 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: li jie <[email protected]>; PostgreSQL Hackers <[email protected]>
On Thu, Jan 18, 2024 at 2:47 PM Amit Kapila <[email protected]> wrote:
>
> On Thu, Jan 18, 2024 at 12:12 PM Bharath Rupireddy
> <[email protected]> wrote:
> >
> > On Wed, Jan 17, 2024 at 11:45 AM li jie <[email protected]> wrote:
> > >
> > > Hi hackers,
> > >
> > > During logical replication, if there is a large write transaction, some
> > > spill files will be written to disk, depending on the setting of
> > > logical_decoding_work_mem.
> > >
> > > This behavior can effectively avoid OOM, but if the transaction
> > > generates a lot of change before commit, a large number of files may
> > > fill the disk. For example, you can update a TB-level table.
> > >
> > > However, I found an inelegant phenomenon. If the modified large table is not
> > > published, its changes will also be written with a large number of spill files.
> > > Look at an example below:
> >
> > Thanks. I agree that decoding and queuing the changes of unpublished
> > tables' data into reorder buffer is an unnecessary task for walsender.
> > It takes processing efforts (CPU overhead), consumes disk space and
> > uses memory configured via logical_decoding_work_mem for a replication
> > connection inefficiently.
> >
>
> This is all true but note that in successful cases (where the table is
> published) all the work done by FilterByTable(accessing caches,
> transaction-related stuff) can add noticeable overhead as anyway we do
> that later in pgoutput_change().
Right. Overhead for published tables need to be studied. A possible
way is to mark the checks performed in
FilterByTable/filter_by_table_cb and skip the same checks in
pgoutput_change. I'm not sure if this works without any issues though.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Reduce useless changes before reassembly during logical replication
2024-01-17 06:15 Reduce useless changes before reassembly during logical replication li jie <[email protected]>
2024-01-18 06:42 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
2024-01-18 09:17 ` Re: Reduce useless changes before reassembly during logical replication Amit Kapila <[email protected]>
@ 2024-02-22 08:11 ` Andy Fan <[email protected]>
1 sibling, 0 replies; 8+ messages in thread
From: Andy Fan @ 2024-02-22 08:11 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Bharath Rupireddy <[email protected]>; li jie <[email protected]>; [email protected]
Hi,
>
> This is all true but note that in successful cases (where the table is
> published) all the work done by FilterByTable(accessing caches,
> transaction-related stuff) can add noticeable overhead as anyway we do
> that later in pgoutput_change(). I think I gave the same comment
> earlier as well but didn't see any satisfactory answer or performance
> data for successful cases to back this proposal.
I did some benchmark yesterday at [1] and found it adds 20% cpu time.
then come out a basic idea, I think it deserves a share. "transaction
related stuff" comes from the syscache/systable access except the
HistorySansphot. and the syscache is required in the following
sistuations:
1. relfilenode (from wal) -> relid.
2. relid -> namespaceid (to check if the relid is a toast relation).
3. if toast, get its origianl relid.
4. access the data from pg_publication_tables.
5. see if the relid is a partition, if yes, we may get its root
relation.
Acutally we already has a RelationSyncCache for #4, and it *only* need
to access syscache when replicate_valid is false, I think this case
should be rare, but the caller doesn't know it, so the caller must
prepare the transaction stuff in advance even in the most case they are
not used. So I think we can get a optimization here.
then the attached patch is made.
Author: yizhi.fzh <[email protected]>
Date: Wed Feb 21 18:40:03 2024 +0800
Make get_rel_sync_entry less depending on transaction state.
get_rel_sync_entry needs transaction only a replicate_valid = false
entry is found, this should be some rare case. However the caller can't
know if a entry is valid, so they have to prepare the transaction state
before calling this function. Such preparation is expensive.
This patch makes the get_rel_sync_entry can manage a transaction stage
only if necessary. so the callers don't need to prepare it blindly.
Then comes to #1, acutally we have RelfilenumberMapHash as a cache, when
the cache is hit (suppose this is a usual case), no transaction stuff
related. I have two ideas then:
1. Optimize the cache hit sistuation like what we just did for
get_rel_sync_entry for the all the 5 kinds of data and only pay the
effort for cache miss case. for the data for #2, #3, #5, all the keys
are relid, so I think a same HTAB should be OK.
2. add the content for #1, #2, #3, #5 to wal when wal_level is set to
logical.
In either case, the changes for get_rel_sync_entry should be needed.
> Note, users can
> configure to stream_in_progress transactions in which case they
> shouldn't see such a big problem.
People would see the changes is spilled to disk, but the CPU cost for
Reorder should be still paid I think.
[1] https://www.postgresql.org/message-id/87o7cadqj3.fsf%40163.com
--
Best Regards
Andy Fan
Attachments:
[text/x-diff] v1-0001-Make-get_rel_sync_entry-less-depending-on-transac.patch (4.2K, ../../[email protected]/2-v1-0001-Make-get_rel_sync_entry-less-depending-on-transac.patch)
download | inline diff:
From 90b8df330df049bd1d7e881dc6e9b108c17b0924 Mon Sep 17 00:00:00 2001
From: "yizhi.fzh" <[email protected]>
Date: Wed, 21 Feb 2024 18:40:03 +0800
Subject: [PATCH v1 1/1] Make get_rel_sync_entry less depending on transaction
state.
get_rel_sync_entry needs transaction only a replicate_valid = false
entry is found, this should be some rare case. However the caller can't
know if a entry is valid, so they have to prepare the transaction state
before calling this function. Such preparation is expensive.
This patch makes the get_rel_sync_entry can manage a transaction stage
only if necessary. so the callers don't need to prepare it blindly.
---
src/backend/replication/pgoutput/pgoutput.c | 60 ++++++++++++++++++++-
1 file changed, 59 insertions(+), 1 deletion(-)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 998f92d671..25e55590a2 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -13,6 +13,7 @@
#include "postgres.h"
#include "access/tupconvert.h"
+#include "access/relation.h"
#include "catalog/partition.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_rel.h"
@@ -214,6 +215,11 @@ static void init_rel_sync_cache(MemoryContext cachectx);
static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data,
Relation relation);
+static RelationSyncEntry *get_rel_sync_entry_by_relid(PGOutputData *data,
+ Oid relid);
+static RelationSyncEntry *get_rel_sync_entry_internal(PGOutputData *data,
+ Relation relation,
+ Oid relid);
static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
static void rel_sync_cache_publication_cb(Datum arg, int cacheid,
uint32 hashvalue);
@@ -1962,11 +1968,29 @@ set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
*/
static RelationSyncEntry *
get_rel_sync_entry(PGOutputData *data, Relation relation)
+{
+ return get_rel_sync_entry_internal(data, relation, InvalidOid);
+}
+
+static RelationSyncEntry *
+__attribute__ ((unused))
+get_rel_sync_entry_by_relid(PGOutputData *data, Oid relid)
+{
+ return get_rel_sync_entry_internal(data, NULL, relid);
+}
+
+static RelationSyncEntry *
+get_rel_sync_entry_internal(PGOutputData *data, Relation relation, Oid oid)
{
RelationSyncEntry *entry;
bool found;
MemoryContext oldctx;
- Oid relid = RelationGetRelid(relation);
+ Oid relid = OidIsValid(oid) ? oid: RelationGetRelid(relation);
+ bool started_xact = false, using_subtxn;
+
+ /* either oid or relation is provided. */
+ Assert(OidIsValid(oid) || RelationIsValid(relation));
+ Assert(!(OidIsValid(oid) && RelationIsValid(relation)));
Assert(RelationSyncCache != NULL);
@@ -1993,6 +2017,23 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->attrmap = NULL;
}
+ if (!entry->replicate_valid && !IsTransactionOrTransactionBlock())
+ {
+ /*
+ * Validating the entry needs to access syscache, which must
+ * be in a transaction state, if that's not ready, start one.
+ */
+
+ using_subtxn = IsTransactionOrTransactionBlock();
+
+ if (using_subtxn)
+ BeginInternalSubTransaction(__func__);
+ else
+ StartTransactionCommand();
+
+ started_xact = true;
+ }
+
/* Validate the entry */
if (!entry->replicate_valid)
{
@@ -2198,9 +2239,19 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
if (entry->pubactions.pubinsert || entry->pubactions.pubupdate ||
entry->pubactions.pubdelete)
{
+ bool rel_opened = false;
+
+ if (!RelationIsValid(relation))
+ {
+ relation = relation_open(oid, AccessShareLock);
+ rel_opened = true;
+ }
/* Initialize the tuple slot and map */
init_tuple_slot(data, relation, entry);
+ if (rel_opened)
+ relation_close(relation, AccessShareLock);
+
/* Initialize the row filter */
pgoutput_row_filter_init(data, rel_publications, entry);
@@ -2215,6 +2266,13 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->replicate_valid = true;
}
+ if (started_xact)
+ {
+ AbortCurrentTransaction();
+ if (using_subtxn)
+ RollbackAndReleaseCurrentSubTransaction();
+ }
+
return entry;
}
--
2.34.1
^ permalink raw reply [nested|flat] 8+ messages in thread
end of thread, other threads:[~2024-02-22 08:11 UTC | newest]
Thread overview: 8+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-03-12 21:35 [PATCH 08/10] Default to zstd.. Justin Pryzby <[email protected]>
2023-01-09 22:29 [PATCH v5 1/2] fix MAINTAIN privs for partitions Nathan Bossart <[email protected]>
2023-01-09 22:29 [PATCH v6 1/2] fix MAINTAIN privs for partitions Nathan Bossart <[email protected]>
2024-01-17 06:15 Reduce useless changes before reassembly during logical replication li jie <[email protected]>
2024-01-18 06:42 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
2024-01-18 09:17 ` Re: Reduce useless changes before reassembly during logical replication Amit Kapila <[email protected]>
2024-01-18 11:14 ` Re: Reduce useless changes before reassembly during logical replication Bharath Rupireddy <[email protected]>
2024-02-22 08:11 ` Re: Reduce useless changes before reassembly during logical replication Andy Fan <[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