Skip to content

Commit c91f71b

Browse files
tvondratvondra
committed
Fix publish_as_relid with multiple publications
Commit 83fd453 allowed publishing of changes via ancestors, for publications defined with publish_via_partition_root. But the way the ancestor was determined in get_rel_sync_entry() was incorrect, simply updating the same variable. So with multiple publications, replicating different ancestors, the outcome depended on the order of publications in the list - the value from the last loop was used, even if it wasn't the top-most ancestor. This is a probably rare situation, as in most cases publications do not overlap, so each partition has exactly one candidate ancestor to replicate as and there's no ambiguity. Fixed by tracking the "ancestor level" for each publication, and picking the top-most ancestor. Adds a test case, verifying the correct ancestor is used for publishing the changes and that this does not depend on order of publications in the list. Older releases have another bug in this loop - once all actions are replicated, the loop is terminated, on the assumption that inspecting additional publications is unecessary. But that misses the fact that those additional applications may replicate different ancestors. Fixed by removal of this break condition. We might still terminate the loop in some cases (e.g. when replicating all actions and the ancestor is the partition root). Backpatch to 13, where publish_via_partition_root was introduced. Initial report and fix by me, test added by Hou zj. Reviews and improvements by Amit Kapila. Author: Tomas Vondra, Hou zj, Amit Kapila Reviewed-by: Amit Kapila, Hou zj Discussion: https://postgr.es/m/d26d24dd-2fab-3c48-0162-2b7f84a9c893%40enterprisedb.com
1 parent d0083c1 commit c91f71b

File tree

5 files changed

+133
-9
lines changed

5 files changed

+133
-9
lines changed

‎src/backend/catalog/pg_publication.c

+19-2
Original file line numberDiff line numberDiff line change
@@ -277,16 +277,21 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
277277

278278
/*
279279
* Returns the relid of the topmost ancestor that is published via this
280-
* publication if any, otherwise returns InvalidOid.
280+
* publication if any and set its ancestor level to ancestor_level,
281+
* otherwise returns InvalidOid.
282+
*
283+
* The ancestor_level value allows us to compare the results for multiple
284+
* publications, and decide which value is higher up.
281285
*
282286
* Note that the list of ancestors should be ordered such that the topmost
283287
* ancestor is at the end of the list.
284288
*/
285289
Oid
286-
GetTopMostAncestorInPublication(Oid puboid, List *ancestors)
290+
GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
287291
{
288292
ListCell *lc;
289293
Oid topmost_relid = InvalidOid;
294+
int level = 0;
290295

291296
/*
292297
* Find the "topmost" ancestor that is in this publication.
@@ -297,13 +302,25 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors)
297302
List *apubids = GetRelationPublications(ancestor);
298303
List *aschemaPubids = NIL;
299304

305+
level++;
306+
300307
if (list_member_oid(apubids, puboid))
308+
{
301309
topmost_relid = ancestor;
310+
311+
if (ancestor_level)
312+
*ancestor_level = level;
313+
}
302314
else
303315
{
304316
aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor));
305317
if (list_member_oid(aschemaPubids, puboid))
318+
{
306319
topmost_relid = ancestor;
320+
321+
if (ancestor_level)
322+
*ancestor_level = level;
323+
}
307324
}
308325

309326
list_free(apubids);

‎src/backend/commands/publicationcmds.c

+2-1
Original file line numberDiff line numberDiff line change
@@ -323,7 +323,8 @@ contain_invalid_rfcolumn(Oid pubid, Relation relation, List *ancestors,
323323
*/
324324
if (pubviaroot && relation->rd_rel->relispartition)
325325
{
326-
publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors);
326+
publish_as_relid
327+
= GetTopMostAncestorInPublication(pubid, ancestors, NULL);
327328

328329
if (!OidIsValid(publish_as_relid))
329330
publish_as_relid = relid;

‎src/backend/replication/pgoutput/pgoutput.c

+40-3
Original file line numberDiff line numberDiff line change
@@ -1748,6 +1748,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
17481748
List *schemaPubids = GetSchemaPublications(schemaId);
17491749
ListCell *lc;
17501750
Oid publish_as_relid = relid;
1751+
int publish_ancestor_level = 0;
17511752
bool am_partition = get_rel_relispartition(relid);
17521753
char relkind = get_rel_relkind(relid);
17531754
List *rel_publications = NIL;
@@ -1815,11 +1816,28 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
18151816
Publication *pub = lfirst(lc);
18161817
bool publish = false;
18171818

1819+
/*
1820+
* Under what relid should we publish changes in this publication?
1821+
* We'll use the top-most relid across all publications. Also track
1822+
* the ancestor level for this publication.
1823+
*/
1824+
Oid pub_relid = relid;
1825+
int ancestor_level = 0;
1826+
1827+
/*
1828+
* If this is a FOR ALL TABLES publication, pick the partition root
1829+
* and set the ancestor level accordingly.
1830+
*/
18181831
if (pub->alltables)
18191832
{
18201833
publish = true;
18211834
if (pub->pubviaroot && am_partition)
1822-
publish_as_relid = llast_oid(get_partition_ancestors(relid));
1835+
{
1836+
List *ancestors = get_partition_ancestors(relid);
1837+
1838+
pub_relid = llast_oid(ancestors);
1839+
ancestor_level = list_length(ancestors);
1840+
}
18231841
}
18241842

18251843
if (!publish)
@@ -1835,16 +1853,21 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
18351853
if (am_partition)
18361854
{
18371855
Oid ancestor;
1856+
int level;
18381857
List *ancestors = get_partition_ancestors(relid);
18391858

18401859
ancestor = GetTopMostAncestorInPublication(pub->oid,
1841-
ancestors);
1860+
ancestors,
1861+
&level);
18421862

18431863
if (ancestor != InvalidOid)
18441864
{
18451865
ancestor_published = true;
18461866
if (pub->pubviaroot)
1847-
publish_as_relid = ancestor;
1867+
{
1868+
pub_relid = ancestor;
1869+
ancestor_level = level;
1870+
}
18481871
}
18491872
}
18501873

@@ -1868,6 +1891,20 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
18681891
entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
18691892

18701893
rel_publications = lappend(rel_publications, pub);
1894+
1895+
/*
1896+
* We want to publish the changes as the top-most ancestor
1897+
* across all publications. So we need to check if the
1898+
* already calculated level is higher than the new one. If
1899+
* yes, we can ignore the new value (as it's a child).
1900+
* Otherwise the new value is an ancestor, so we keep it.
1901+
*/
1902+
if (publish_ancestor_level > ancestor_level)
1903+
continue;
1904+
1905+
/* The new value is an ancestor, so let's keep it. */
1906+
publish_as_relid = pub_relid;
1907+
publish_ancestor_level = ancestor_level;
18711908
}
18721909
}
18731910

‎src/include/catalog/pg_publication.h

+2-1
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,8 @@ extern List *GetAllSchemaPublicationRelations(Oid puboid,
134134
extern List *GetPubPartitionOptionRelations(List *result,
135135
PublicationPartOpt pub_partopt,
136136
Oid relid);
137-
extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors);
137+
extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors,
138+
int *ancestor_level);
138139

139140
extern bool is_publishable_relation(Relation rel);
140141
extern bool is_schema_publication(Oid pubid);

‎src/test/subscription/t/013_partition.pl

+70-2
Original file line numberDiff line numberDiff line change
@@ -409,6 +409,14 @@ BEGIN
409409
"CREATE TABLE tab3 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
410410
$node_publisher->safe_psql('postgres',
411411
"CREATE TABLE tab3_1 PARTITION OF tab3 FOR VALUES IN (0, 1, 2, 3, 5, 6)");
412+
413+
$node_publisher->safe_psql('postgres',
414+
"CREATE TABLE tab4 (a int PRIMARY KEY) PARTITION BY LIST (a)");
415+
$node_publisher->safe_psql('postgres',
416+
"CREATE TABLE tab4_1 PARTITION OF tab4 FOR VALUES IN (0, 1) PARTITION BY LIST (a)");
417+
$node_publisher->safe_psql('postgres',
418+
"CREATE TABLE tab4_1_1 PARTITION OF tab4_1 FOR VALUES IN (0, 1)");
419+
412420
$node_publisher->safe_psql('postgres',
413421
"ALTER PUBLICATION pub_all SET (publish_via_partition_root = true)");
414422
# Note: tab3_1's parent is not in the publication, in which case its
@@ -419,6 +427,11 @@ BEGIN
419427
"CREATE PUBLICATION pub_viaroot FOR TABLE tab2, tab2_1, tab3_1 WITH (publish_via_partition_root = true)"
420428
);
421429

430+
# for tab4, we publish changes through the "middle" partitioned table
431+
$node_publisher->safe_psql('postgres',
432+
"CREATE PUBLICATION pub_lower_level FOR TABLE tab4_1 WITH (publish_via_partition_root = true)"
433+
);
434+
422435
# prepare data for the initial sync
423436
$node_publisher->safe_psql('postgres', "INSERT INTO tab2 VALUES (1)");
424437

@@ -462,10 +475,20 @@ BEGIN
462475
$node_subscriber2->safe_psql('postgres',
463476
"CREATE TABLE tab3_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3_1', b text)"
464477
);
478+
479+
# Note: We create two separate tables, not a partitioned one, so that we can
480+
# easily identity through which relation were the changes replicated.
481+
$node_subscriber2->safe_psql('postgres',
482+
"CREATE TABLE tab4 (a int PRIMARY KEY)"
483+
);
484+
$node_subscriber2->safe_psql('postgres',
485+
"CREATE TABLE tab4_1 (a int PRIMARY KEY)"
486+
);
465487
# Publication that sub2 points to now publishes via root, so must update
466-
# subscription target relations.
488+
# subscription target relations. We set the list of publications so that
489+
# the FOR ALL TABLES publication is second (the list order matters).
467490
$node_subscriber2->safe_psql('postgres',
468-
"ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");
491+
"ALTER SUBSCRIPTION sub2 SET PUBLICATION pub_lower_level, pub_all");
469492

470493
# Wait for initial sync of all subscriptions
471494
$node_subscriber1->poll_query_until('postgres', $synced_query)
@@ -487,6 +510,11 @@ BEGIN
487510
$node_publisher->safe_psql('postgres',
488511
"INSERT INTO tab3 VALUES (1), (0), (3), (5)");
489512

513+
# Insert a row into the leaf partition, should be replicated through the
514+
# partition root (thanks to the FOR ALL TABLES partition).
515+
$node_publisher->safe_psql('postgres',
516+
"INSERT INTO tab4 VALUES (0)");
517+
490518
$node_publisher->wait_for_catchup('sub_viaroot');
491519
$node_publisher->wait_for_catchup('sub2');
492520

@@ -525,6 +553,46 @@ BEGIN
525553
sub2_tab3|3
526554
sub2_tab3|5), 'inserts into tab3 replicated');
527555

556+
# tab4 change should be replicated through the root partition, which
557+
# maps to the tab4 relation on subscriber.
558+
$result = $node_subscriber2->safe_psql('postgres',
559+
"SELECT a FROM tab4 ORDER BY 1");
560+
is( $result, qq(0), 'inserts into tab4 replicated');
561+
562+
$result = $node_subscriber2->safe_psql('postgres',
563+
"SELECT a FROM tab4_1 ORDER BY 1");
564+
is( $result, qq(), 'inserts into tab4_1 replicated');
565+
566+
567+
# now switch the order of publications in the list, try again, the result
568+
# should be the same (no dependence on order of pulications)
569+
$node_subscriber2->safe_psql('postgres',
570+
"ALTER SUBSCRIPTION sub2 SET PUBLICATION pub_all, pub_lower_level");
571+
572+
# make sure the subscription on the second subscriber is synced, before
573+
# continuing
574+
$node_subscriber2->poll_query_until('postgres', $synced_query)
575+
or die "Timed out while waiting for subscriber to synchronize data";
576+
577+
# Insert a change into the leaf partition, should be replicated through
578+
# the partition root (thanks to the FOR ALL TABLES partition).
579+
$node_publisher->safe_psql('postgres',
580+
"INSERT INTO tab4 VALUES (1)");
581+
582+
$node_publisher->wait_for_catchup('sub2');
583+
584+
# tab4 change should be replicated through the root partition, which
585+
# maps to the tab4 relation on subscriber.
586+
$result = $node_subscriber2->safe_psql('postgres',
587+
"SELECT a FROM tab4 ORDER BY 1");
588+
is( $result, qq(0
589+
1), 'inserts into tab4 replicated');
590+
591+
$result = $node_subscriber2->safe_psql('postgres',
592+
"SELECT a FROM tab4_1 ORDER BY 1");
593+
is( $result, qq(), 'inserts into tab4_1 replicated');
594+
595+
528596
# update (replicated as update)
529597
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
530598
$node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 6 WHERE a = 5");

0 commit comments

Comments
 (0)