Fix data inconsistency between publisher and subscriber.

We were not updating the partition map cache in the subscriber even when
the corresponding remote rel is changed. Due to this data was getting
incorrectly replicated for partition tables after the publisher has
changed the table schema.

Fix it by resetting the required entries in the partition map cache after
receiving a new relation mapping from the publisher.

Reported-by: Shi Yu
Author: Shi Yu, Hou Zhijie
Reviewed-by: Amit Langote, Amit Kapila
Backpatch-through: 13, where it was introduced
Discussion: https://postgr.es/m/OSZPR01MB6310F46CD425A967E4AEF736FDA49@OSZPR01MB6310.jpnprd01.prod.outlook.com

diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 283afa5d9d7157e9cd59066e9cf538770d05c3c4..026d2c2af47682a2b023d6ad9772c567a2592732 100644 (file)
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -453,6 +453,40 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid)
    }
 }
 
+/*
+ * Reset the entries in the partition map that refer to remoterel.
+ *
+ * Called when new relation mapping is sent by the publisher to update our
+ * expected view of incoming data from said publisher.
+ *
+ * Note that we don't update the remoterel information in the entry here,
+ * we will update the information in logicalrep_partition_open to avoid
+ * unnecessary work.
+ */
+void
+logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel)
+{
+   HASH_SEQ_STATUS status;
+   LogicalRepPartMapEntry *part_entry;
+   LogicalRepRelMapEntry *entry;
+
+   if (LogicalRepPartMap == NULL)
+       return;
+
+   hash_seq_init(&status, LogicalRepPartMap);
+   while ((part_entry = (LogicalRepPartMapEntry *) hash_seq_search(&status)) != NULL)
+   {
+       entry = &part_entry->relmapentry;
+
+       if (entry->remoterel.remoteid != remoterel->remoteid)
+           continue;
+
+       logicalrep_relmap_free_entry(entry);
+
+       memset(entry, 0, sizeof(LogicalRepRelMapEntry));
+   }
+}
+
 /*
  * Initialize the partition map cache.
  */

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c04abd79e74e606f6bed42c1e9829ac5d624c3b8..8eafd61cdbd9234303aa1f62a71b415ba0a0e22e 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -612,6 +612,9 @@ apply_handle_relation(StringInfo s)
 
    rel = logicalrep_read_rel(s);
    logicalrep_relmap_update(rel);
+
+   /* Also reset all entries in the partition map that refer to remoterel. */
+   logicalrep_partmap_reset_relmap(rel);
 }
 
 /*

diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index e369b27e7f58438b1b9f960b45e777aee0a3a5ab..847e6f184656e64c81cfff0678ed57319a42c5e9 100644 (file)
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -33,6 +33,7 @@ typedef struct LogicalRepRelMapEntry
 } LogicalRepRelMapEntry;
 
 extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
+extern void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel);
 
 extern LogicalRepRelMapEntry *logicalrep_rel_open(LogicalRepRelId remoteid,
                                                  LOCKMODE lockmode);

diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl
index 4eef52df771fbb5642974f9c124c2fbc3b6dcb84..8a1ec55f24feb725c9c6128f8dd9e6dd816dc6b5 100644 (file)
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 69;
+use Test::More tests => 70;
 
 # setup
 
@@ -838,3 +838,18 @@ $node_publisher->wait_for_catchup('sub2');
 $result = $node_subscriber2->safe_psql('postgres',
    "SELECT a, b, c FROM tab5 ORDER BY 1");
 is($result, qq(3|1|), 'updates of tab5 replicated correctly after altering table on subscriber');
+
+# Test that replication into the partitioned target table continues to
+# work correctly when the published table is altered.
+$node_publisher->safe_psql(
+   'postgres', q{
+   ALTER TABLE tab5 DROP COLUMN b, ADD COLUMN c INT;
+   ALTER TABLE tab5 ADD COLUMN b INT;});
+
+$node_publisher->safe_psql('postgres', "UPDATE tab5 SET c = 1 WHERE a = 3");
+
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+   "SELECT a, b, c FROM tab5 ORDER BY 1");
+is($result, qq(3||1), 'updates of tab5 replicated correctly after altering table on publisher');