From 9d009b5a884b3f9dad7e9b7ba4797c65ad8b9d94 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Thu, 29 Jan 2026 18:56:13 +0530 Subject: [PATCH v38] handle EXCEPT TABLE correctly with partitioned tables When a publication is created with EXCEPT TABLE, adjust logical replication so that data synchronization and change replication correctly respect exclusions for partitioned tables. On the subscriber side, extend fetch_remote_table_info() to compute the effective set of relations used for the initial COPY. When exclusions are present, the root partitioned table can no longer be used directly; instead, derive the list of non-excluded leaf partitions and combine them with UNION ALL. When no exclusions exist, retain the existing behavior and copy from the root relation as before. This is based on approach 1 discussed at: https://www.postgresql.org/message-id/CAJpy0uD81HRrMYr7S-6AV4W2PtbGKM-nf2D89zsoMHJ9jZssUg@mail.gmail.com This patch is a topup patch on top of 0001 patch. --- doc/src/sgml/ref/create_publication.sgml | 17 +- src/backend/catalog/pg_publication.c | 4 +- src/backend/commands/subscriptioncmds.c | 2 + src/backend/replication/logical/tablesync.c | 185 +++++++++++++++++- src/backend/replication/pgoutput/pgoutput.c | 9 +- src/include/replication/worker_internal.h | 6 + .../t/037_rep_changes_except_table.pl | 156 ++++++++------- src/tools/pgindent/typedefs.list | 1 + 8 files changed, 285 insertions(+), 95 deletions(-) diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index 1e091bb3c6d..730b9c4bced 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -205,16 +205,15 @@ CREATE PUBLICATION name tables are excluded. - For partitioned tables, when publish_via_partition_root - is set to true, specifying a root partitioned table in - EXCEPT TABLE excludes it and all its partitions from - replication. Specifying a leaf partition has no effect, as its changes are - still replicated via the root partitioned table. When + For partitioned tables, when a table is specified in EXCEPT TABLE, then + changes to that table and all of its partitions (that is, the entire + partition subtree rooted at that table) are not replicated. This behavior + is the same regardless of whether publish_via_partition_root is set to - false, specifying a root partitioned table has no - effect, as changes are replicated via the leaf partitions. Specifying a - leaf partition excludes only that partition from replication. The optional - * has no meaning for partitioned tables. + true or false. The + publish_via_partition_root setting only determines + which relation is used as the publishing relation for replicated changes, + and does not affect exclusion semantics. diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 09c69005122..e72e49bd97b 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -960,8 +960,8 @@ GetAllPublicationRelations(Publication *pub, char relkind) if (relkind == RELKIND_RELATION) exceptlist = GetAllPublicationExcludedTables(pubid, pubviaroot ? - PUBLICATION_PART_ALL : - PUBLICATION_PART_ROOT); + PUBLICATION_PART_ROOT : + PUBLICATION_PART_LEAF); classRel = table_open(RelationRelationId, AccessShareLock); diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 0b3c8499b49..804ae2f349e 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -2940,6 +2940,8 @@ fetch_relation_list(WalReceiverConn *wrconn, List *publications) pub_names.data); } + elog(LOG, "fetch_relation_list: executing query to fetch effectiverelations: \n%s", + cmd.data); pfree(pub_names.data); res = walrcv_exec(wrconn, cmd.data, column_count, tableRow); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 19a3c21a863..2e8466b3ab7 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -716,21 +716,28 @@ copy_read_data(void *outbuf, int minread, int maxread) * message provides during replication. * * This function also returns (a) the relation qualifications to be used in - * the COPY command, and (b) whether the remote relation has published any - * generated column. + * the COPY command, (b) whether the remote relation has published any + * generated column, and (c) computes the effective set of relations to be used + * as COPY sources when exclusions are present. When no exclusions exist, the + * list remains empty and the root relation is used as-is. When exclusions + * exist, the list contains leaf relations that are not excluded and must be + * combined using UNION ALL. */ static void fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, - List **qual, bool *gencol_published) + List **qual, bool *gencol_published, + List **effective_relations) { WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[] = {OIDOID, CHAROID, CHAROID}; + Oid tableRow[] = {OIDOID, CHAROID, CHAROID, BOOLOID}; Oid attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID, BOOLOID}; Oid qualRow[] = {TEXTOID}; + Oid filtertableRow[] = {TEXTOID, TEXTOID}; bool isnull; int natt; + bool is_partition; StringInfo pub_names = NULL; Bitmapset *included_cols = NULL; int server_version = walrcv_server_version(LogRepWorkerWalRcvConn); @@ -740,7 +747,7 @@ fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, /* First fetch Oid and replica identity. */ initStringInfo(&cmd); - appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind" + appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind, c.relispartition" " FROM pg_catalog.pg_class c" " INNER JOIN pg_catalog.pg_namespace n" " ON (c.relnamespace = n.oid)" @@ -770,6 +777,8 @@ fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, Assert(!isnull); lrel->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull)); Assert(!isnull); + is_partition = DatumGetBool(slot_getattr(slot, 4, &isnull)); + Assert(!isnull); ExecDropSingleTupleTableSlot(slot); walrcv_clear_result(res); @@ -954,6 +963,110 @@ fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, walrcv_clear_result(res); + if (server_version >= 190000 && !is_partition && + lrel->relkind == RELKIND_PARTITIONED_TABLE) + { + resetStringInfo(&cmd); + + /* + * This query recursively traverses the inheritance (partition) tree + * starting from the given table OID and determines which leaf + * relations should be included for replication. Exclusion propagates + * from parent to child, and a relation is also treated as excluded if + * it is explicitly marked with prexcept = true in pg_publication_rel + * for the specified publications. The final result returns only + * non excluded leaf relations. + */ + appendStringInfo(&cmd, + "WITH RECURSIVE branch_search AS (\n" + " SELECT %u::oid AS oid, false AS is_excluded\n" + " UNION ALL\n" + " SELECT i.inhrelid AS oid,\n" + " parent.is_excluded\n" + " OR EXISTS (\n" + " SELECT 1\n" + " FROM pg_publication_rel pr\n" + " JOIN pg_publication p ON p.oid = pr.prpubid\n" + " WHERE pr.prrelid = i.inhrelid\n" + " AND pr.prexcept = true\n" + " AND p.pubname IN ( %s )\n" + " ) AS is_excluded\n" + " FROM pg_inherits i\n" + " JOIN branch_search parent\n" + " ON i.inhparent = parent.oid\n" + ")\n" + " SELECT n.nspname AS schemaname, c.relname AS relname\n" + " FROM (\n" + " SELECT bs.*, bool_or(bs.is_excluded) OVER () AS has_exclusion\n" + " FROM branch_search bs\n" + " ) bs\n" + " JOIN pg_class c ON c.oid = bs.oid\n" + " JOIN pg_namespace n ON n.oid = c.relnamespace\n" + " WHERE bs.is_excluded = false\n" + " AND (\n" + " (bs.has_exclusion = false AND bs.oid = %u::oid)\n" + " OR (bs.has_exclusion = true AND NOT EXISTS (\n" + " SELECT 1\n" + " FROM pg_inherits\n" + " WHERE inhparent = bs.oid\n" + " ))\n" + " );", + lrel->remoteid, + pub_names->data, + lrel->remoteid + ); + + elog(LOG, "Executing query to get the tables:\n%s", cmd.data); + res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, + lengthof(filtertableRow), filtertableRow); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get non excluded table list for table \"%s.%s\" from publisher: %s", + nspname, relname, res->err)); + + /* + * Store the tables as a list of schemaname and tablename. + */ + natt = 0; + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + QualifiedRelationName *relinfo = palloc_object(QualifiedRelationName); + + relinfo->schemaname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + relinfo->relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + *effective_relations = lappend(*effective_relations, relinfo); + + ExecClearTuple(slot); + } + + ExecDropSingleTupleTableSlot(slot); + + /* + * If there is exactly one item in the exclusion list and it equals + * the table being processed, that means no actual exclusion occurred. + */ + if (list_length(*effective_relations) == 1) + { + QualifiedRelationName *relinfo; + + relinfo = linitial(*effective_relations); + if (strcmp(nspname, relinfo->schemaname) == 0 && + strcmp(relname, relinfo->relname) == 0) + { + pfree(relinfo->schemaname); + pfree(relinfo->relname); + list_free_deep(*effective_relations); + *effective_relations = NIL; + } + } + } + /* * Get relation's row filter expressions. DISTINCT avoids the same * expression of a table in multiple publications from being included @@ -1043,6 +1156,7 @@ copy_table(Relation rel) LogicalRepRelMapEntry *relmapentry; LogicalRepRelation lrel; List *qual = NIL; + List *effective_relations = NIL; WalRcvExecResult *res; StringInfoData cmd; CopyFromState cstate; @@ -1054,7 +1168,7 @@ copy_table(Relation rel) /* Get the publisher relation info. */ fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)), RelationGetRelationName(rel), &lrel, &qual, - &gencol_published); + &gencol_published, &effective_relations); /* Put the relation into relmap. */ logicalrep_relmap_update(&lrel); @@ -1066,12 +1180,64 @@ copy_table(Relation rel) /* Start copy on the publisher. */ initStringInfo(&cmd); + + if (effective_relations && list_length(effective_relations)) + { + bool first = true; + + /* + * Build a single COPY command to synchronize all resolved relations + * into the root table. + * + * The array 'effective_relations' contains the leaf tables of + * partition hierarchies, with excluded subtrees removed according to + * the EXCEPT clauses. This applies only when + * 'publish_via_partition_root' is enabled, since the initial sync must + * route all changes to the root table. + * + * We construct a UNION ALL query that combines data from multiple leaf + * relations into one sub-COPY statement, ensuring all rows are copied + * consistently into the root table. + */ + appendStringInfoString(&cmd, "COPY (\n"); + foreach_ptr(QualifiedRelationName, relinfo, effective_relations) + { + if (!first) + appendStringInfoString(&cmd, "UNION ALL\n"); + + first = false; + + appendStringInfoString(&cmd, "SELECT "); + + /* If the table has columns, then specify the columns */ + if (lrel.natts) + { + for (int i = 0; i < lrel.natts; i++) + { + if (i > 0) + appendStringInfoString(&cmd, ", "); + + appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i])); + } + } + else + appendStringInfoString(&cmd, " * "); + + appendStringInfo(&cmd, " FROM %s\n", + quote_qualified_identifier(relinfo->schemaname, + relinfo->relname)); + } + + appendStringInfoString(&cmd, ")\n"); + appendStringInfoString(&cmd, "TO STDOUT"); + } /* Regular or partitioned table with no row filter or generated columns */ - if ((lrel.relkind == RELKIND_RELATION || lrel.relkind == RELKIND_PARTITIONED_TABLE) - && qual == NIL && !gencol_published) + else if ((lrel.relkind == RELKIND_RELATION || + lrel.relkind == RELKIND_PARTITIONED_TABLE) && + qual == NIL && !gencol_published) { appendStringInfo(&cmd, "COPY %s", - quote_qualified_identifier(lrel.nspname, lrel.relname)); + quote_qualified_identifier(lrel.nspname, lrel.relname)); /* If the table has columns, then specify the columns */ if (lrel.natts) @@ -1157,6 +1323,7 @@ copy_table(Relation rel) } res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL); + elog(LOG, "Tablesync worker: Executing query to get the initial sync data:\n%s", cmd.data); pfree(cmd.data); if (res->status != WALRCV_OK_COPY_OUT) ereport(ERROR, diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 05802482c10..407d9304101 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -2209,8 +2209,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) * TABLE list: * * 1. If pubviaroot is set and the relation is a partition, check - * whether the partition root is included in the EXCEPT TABLE - * list. If so, do not publish the change. + * whether the current relation or any of the ancestors is included + * in the EXCEPT TABLE list. If so, do not publish the change. * * 2. If pubviaroot is not set, check whether the relation itself * is included in the EXCEPT TABLE list. If so, do not publish the @@ -2228,6 +2228,11 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) { List *ancestors = get_partition_ancestors(relid); + GetRelationPublications(relid, NULL, &exceptpubids); + + foreach_oid(ancestor, ancestors) + GetRelationPublications(ancestor, NULL, &exceptpubids); + pub_relid = llast_oid(ancestors); ancestor_level = list_length(ancestors); } diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index c1285fdd1bc..cab97531276 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -234,6 +234,12 @@ typedef struct ParallelApplyWorkerInfo ParallelApplyWorkerShared *shared; } ParallelApplyWorkerInfo; +typedef struct QualifiedRelationName +{ + char *schemaname; + char *relname; +} QualifiedRelationName; + /* Main memory context for apply worker. Permanent during worker lifetime. */ extern PGDLLIMPORT MemoryContext ApplyContext; diff --git a/src/test/subscription/t/037_rep_changes_except_table.pl b/src/test/subscription/t/037_rep_changes_except_table.pl index 95904ddd005..80b6a70fca8 100644 --- a/src/test/subscription/t/037_rep_changes_except_table.pl +++ b/src/test/subscription/t/037_rep_changes_except_table.pl @@ -88,8 +88,11 @@ $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_schema"); $node_publisher->safe_psql( 'postgres', qq( CREATE TABLE sch1.t1(a int) PARTITION BY RANGE(a); - CREATE TABLE sch1.part1 PARTITION OF sch1.t1 FOR VALUES FROM (0) TO (5); - CREATE TABLE sch1.part2 PARTITION OF sch1.t1 FOR VALUES FROM (6) TO (10); + CREATE TABLE sch1.part1 PARTITION OF sch1.t1 FOR VALUES FROM (0) TO (100); + CREATE TABLE sch1.part2(a int) PARTITION BY RANGE(a); + CREATE TABLE sch1.part2_1 PARTITION OF sch1.part2 FOR VALUES FROM (101) TO (150); + CREATE TABLE sch1.part2_2 PARTITION OF sch1.part2 FOR VALUES FROM (151) TO (200); + ALTER TABLE sch1.t1 ATTACH PARTITION sch1.part2 FOR VALUES FROM (101) TO (200); )); $node_subscriber->safe_psql( @@ -97,142 +100,149 @@ $node_subscriber->safe_psql( CREATE TABLE sch1.t1(a int); CREATE TABLE sch1.part1(a int); CREATE TABLE sch1.part2(a int); + CREATE TABLE sch1.part2_1(a int); + CREATE TABLE sch1.part2_2(a int); )); -# EXCEPT TABLE (sch1.part1) with publish_via_partition_root = false -# Excluding a partition while publish_via_partition_root = false prevents -# replication of rows inserted into the partitioned table for that particular -# partition. +# Excluding the root partitioned table excludes all its partitions as well when +# publish_via_partition_root = false. $node_publisher->safe_psql( 'postgres', qq( - CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.part1) WITH (publish_via_partition_root = false); - INSERT INTO sch1.t1 VALUES (1), (6); + CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.t1) WITH (publish_via_partition_root = false); + INSERT INTO sch1.t1 VALUES (1), (101), (151); )); $node_subscriber->safe_psql('postgres', "CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part" ); $node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); $node_publisher->safe_psql('postgres', - "INSERT INTO sch1.t1 VALUES (2), (7);"); + "SELECT slot_name FROM pg_replication_slot_advance('test_slot', pg_current_wal_lsn());" +); +$node_publisher->safe_psql('postgres', + "INSERT INTO sch1.t1 VALUES (2), (102), (152)"); + +# Verify that data inserted to the partitioned table is not published when it is +# excluded with publish_via_partition_root = true. +$result = $node_publisher->safe_psql('postgres', + "SELECT count(*) = 0 FROM pg_logical_slot_get_binary_changes('test_slot', NULL, NULL, 'proto_version', '1', 'publication_names', 'tap_pub_part')" +); $node_publisher->wait_for_catchup('tap_sub_part'); +# Check that no rows are replicated to subscriber $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.t1"); -is($result, qq(), 'check rows on partitioned table'); +is($result, qq(), 'check rows on root table'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part1"); -is($result, qq(), 'check rows on excluded partition'); +is($result, qq(), 'check rows on table sch1.part1'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2"); -is( $result, qq(6 -7), 'check rows on other partition'); +is($result, qq(), 'check rows on table sch1.part2'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_1"); +is($result, qq(), 'check rows on table sch1.part2_1'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_2"); +is($result, qq(), 'check rows on table sch1.part2_2'); -$node_publisher->safe_psql('postgres', "TRUNCATE sch1.t1"); -$node_publisher->wait_for_catchup('tap_sub_part'); $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_part"); $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_part;"); -# EXCEPT TABLE (sch1.t1) with publish_via_partition_root = false -# Excluding the partitioned table still allows rows inserted into the -# partitioned table to be replicated via its partitions. +# Excluding the root partitioned table excludes all its partitions as well when +# publish_via_partition_root = true. $node_publisher->safe_psql( 'postgres', qq( - CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.t1) WITH (publish_via_partition_root = false); - INSERT INTO sch1.t1 VALUES (1), (6); + CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.t1) WITH (publish_via_partition_root = true); )); $node_subscriber->safe_psql('postgres', "CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part" ); $node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); $node_publisher->safe_psql('postgres', - "INSERT INTO sch1.t1 VALUES (2), (7);"); + "SELECT slot_name FROM pg_replication_slot_advance('test_slot', pg_current_wal_lsn());" +); +$node_publisher->safe_psql('postgres', + "INSERT INTO sch1.t1 VALUES (3), (103), (153);"); + +# Verify that data inserted to the partitioned table is not published when it is +# excluded with publish_via_partition_root = true. +$result = $node_publisher->safe_psql('postgres', + "SELECT count(*) = 0 FROM pg_logical_slot_get_binary_changes('test_slot', NULL, NULL, 'proto_version', '1', 'publication_names', 'tap_pub_part')" +); $node_publisher->wait_for_catchup('tap_sub_part'); +# Check that no rows are replicated to subscriber $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.t1"); -is($result, qq(), 'check rows on partitioned table'); +is($result, qq(), 'check rows on root table'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part1"); -is( $result, qq(1 -2), 'check rows on first partition'); +is($result, qq(), 'check rows on table sch1.part1'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2"); -is( $result, qq(6 -7), 'check rows on second partition'); +is($result, qq(), 'check rows on table sch1.part2'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_1"); +is($result, qq(), 'check rows on table sch1.part2_1'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_2"); +is($result, qq(), 'check rows on table sch1.part2_2'); -$node_publisher->safe_psql('postgres', "TRUNCATE sch1.t1"); -$node_publisher->wait_for_catchup('tap_sub_part'); $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_part"); $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_part;"); $node_publisher->safe_psql('postgres', "SELECT slot_name FROM pg_replication_slot_advance('test_slot', pg_current_wal_lsn());" ); -# EXCEPT TABLE (sch1.t1) with publish_via_partition_root = true -# When the partitioned table is excluded and publish_via_partition_root is true, -# no rows from the table or its partitions are replicated. +# Excluding one of the child partition table with +# publish_via_partition_root = true should replicate the other partitions. $node_publisher->safe_psql( 'postgres', qq( - CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.t1) WITH (publish_via_partition_root); - INSERT INTO sch1.t1 VALUES (1), (6); + TRUNCATE sch1.t1; + INSERT INTO sch1.t1 VALUES (3), (103), (153); + CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.part2) WITH (publish_via_partition_root = true); +)); +$node_subscriber->safe_psql( + 'postgres', qq( + TRUNCATE sch1.t1; + CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part; )); -$node_subscriber->safe_psql('postgres', - "CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part" -); $node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); $node_publisher->safe_psql('postgres', - "INSERT INTO sch1.t1 VALUES (2), (7);"); -$node_publisher->wait_for_catchup('tap_sub_part'); + "SELECT slot_name FROM pg_replication_slot_advance('test_slot', pg_current_wal_lsn());" +); # Verify that data inserted to the partitioned table is not published when it is # excluded with publish_via_partition_root = true. $result = $node_publisher->safe_psql('postgres', "SELECT count(*) = 0 FROM pg_logical_slot_get_binary_changes('test_slot', NULL, NULL, 'proto_version', '1', 'publication_names', 'tap_pub_part')" ); -is($result, qq(t), 'check no changes for excluded table in replication slot'); +$node_publisher->wait_for_catchup('tap_sub_part'); +# Check that table data 103 and 153 which is present in sch1.sch1.part2 should +# not be replicated. $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.t1"); -is($result, qq(), 'check rows on partitioned table'); +is($result, qq(3), 'check rows on root table'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part1"); -is($result, qq(), 'check rows on first partition'); +is($result, qq(), 'check rows on table sch1.part1'); $result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2"); -is($result, qq(), 'check rows on second partition'); - -$node_publisher->safe_psql('postgres', "TRUNCATE sch1.t1"); -$node_publisher->wait_for_catchup('tap_sub_part'); -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_part"); -$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_part;"); - -# EXCEPT TABLE (sch1.part1) with publish_via_partition_root = true -# When a partition is excluded but publish_via_partition_root is true, -# rows published through the partitioned table can still be replicated. -$node_publisher->safe_psql( - 'postgres', qq( - CREATE PUBLICATION tap_pub_part FOR ALL TABLES EXCEPT TABLE (sch1.part1) WITH (publish_via_partition_root); - INSERT INTO sch1.t1 VALUES (1), (6) -)); -$node_subscriber->safe_psql('postgres', - "CREATE SUBSCRIPTION tap_sub_part CONNECTION '$publisher_connstr' PUBLICATION tap_pub_part" -); -$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub_part'); -$node_publisher->safe_psql('postgres', - "INSERT INTO sch1.t1 VALUES (2), (7);"); -$node_publisher->wait_for_catchup('tap_sub_part'); +is($result, qq(), 'check rows on table sch1.part2'); $result = - $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.t1 ORDER BY a"); -is( $result, qq(1 -2 -6 -7), 'check rows on partitioned table'); + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_1"); +is($result, qq(), 'check rows on table sch1.part2_1'); -$result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part1"); -is($result, qq(), 'check rows on excluded partition'); +$result = + $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2_2"); +is($result, qq(), 'check rows on table sch1.part2_2'); -$result = $node_subscriber->safe_psql('postgres', "SELECT * FROM sch1.part2"); -is($result, qq(), 'check rows on other partition'); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_part"); +$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_part;"); -$node_subscriber->stop('fast'); $node_publisher->stop('fast'); done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 34374df0d67..bfe2964cba1 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2418,6 +2418,7 @@ QTNode QUERYTYPE QualCost QualItem +QualifiedRelationName Query QueryCompletion QueryDesc -- 2.43.0