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

   1
   2 # Copyright (c) 2021-2022, PostgreSQL Global Development Group
   3
   4 # Test logical replication with partitioned tables
   5 use strict;
   6 use warnings;
   7 use PostgreSQL::Test::Cluster;
   8 use PostgreSQL::Test::Utils;
   9 use Test::More tests => 63;
  10
  11 # setup
  12
  13 my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
  14 $node_publisher->init(allows_streaming => 'logical');
  15 $node_publisher->start;
  16
  17 my $node_subscriber1 = PostgreSQL::Test::Cluster->new('subscriber1');
  18 $node_subscriber1->init(allows_streaming => 'logical');
  19 $node_subscriber1->start;
  20
  21 my $node_subscriber2 = PostgreSQL::Test::Cluster->new('subscriber2');
  22 $node_subscriber2->init(allows_streaming => 'logical');
  23 $node_subscriber2->start;
  24
  25 my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
  26
  27 # publisher
  28 $node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub1");
  29 $node_publisher->safe_psql('postgres',
  30         "CREATE PUBLICATION pub_all FOR ALL TABLES");
  31 $node_publisher->safe_psql('postgres',
  32         "CREATE TABLE tab1 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
  33 $node_publisher->safe_psql('postgres',
  34         "CREATE TABLE tab1_1 (b text, a int NOT NULL)");
  35 $node_publisher->safe_psql('postgres',
  36         "ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3)");
  37 $node_publisher->safe_psql('postgres',
  38         "CREATE TABLE tab1_2 PARTITION OF tab1 FOR VALUES IN (4, 5, 6)");
  39 $node_publisher->safe_psql('postgres',
  40         "CREATE TABLE tab1_def PARTITION OF tab1 DEFAULT");
  41 $node_publisher->safe_psql('postgres',
  42         "ALTER PUBLICATION pub1 ADD TABLE tab1, tab1_1");
  43
  44 # subscriber1
  45 #
  46 # This is partitioned differently from the publisher.  tab1_2 is
  47 # subpartitioned.  This tests the tuple routing code on the
  48 # subscriber.
  49 $node_subscriber1->safe_psql('postgres',
  50         "CREATE TABLE tab1 (c text, a int PRIMARY KEY, b text) PARTITION BY LIST (a)"
  51 );
  52 $node_subscriber1->safe_psql('postgres',
  53         "CREATE TABLE tab1_1 (b text, c text DEFAULT 'sub1_tab1', a int NOT NULL)"
  54 );
  55 $node_subscriber1->safe_psql('postgres',
  56         "ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3)");
  57 $node_subscriber1->safe_psql('postgres',
  58         "CREATE TABLE tab1_2 PARTITION OF tab1 (c DEFAULT 'sub1_tab1') FOR VALUES IN (4, 5, 6) PARTITION BY LIST (a)"
  59 );
  60 $node_subscriber1->safe_psql('postgres',
  61         "CREATE TABLE tab1_2_1 (c text, b text, a int NOT NULL)");
  62 $node_subscriber1->safe_psql('postgres',
  63         "ALTER TABLE tab1_2 ATTACH PARTITION tab1_2_1 FOR VALUES IN (5)");
  64 $node_subscriber1->safe_psql('postgres',
  65         "CREATE TABLE tab1_2_2 PARTITION OF tab1_2 FOR VALUES IN (4, 6)");
  66 $node_subscriber1->safe_psql('postgres',
  67         "CREATE TABLE tab1_def PARTITION OF tab1 (c DEFAULT 'sub1_tab1') DEFAULT"
  68 );
  69 $node_subscriber1->safe_psql('postgres',
  70         "CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
  71 );
  72
  73 # Add set of AFTER replica triggers for testing that they are fired
  74 # correctly.  This uses a table that records details of all trigger
  75 # activities.  Triggers are marked as enabled for a subset of the
  76 # partition tree.
  77 $node_subscriber1->safe_psql(
  78         'postgres', qq{
  79 CREATE TABLE sub1_trigger_activity (tgtab text, tgop text,
  80   tgwhen text, tglevel text, olda int, newa int);
  81 CREATE FUNCTION sub1_trigger_activity_func() RETURNS TRIGGER AS \$\$
  82 BEGIN
  83   IF (TG_OP = 'INSERT') THEN
  84     INSERT INTO public.sub1_trigger_activity
  85       SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a;
  86   ELSIF (TG_OP = 'UPDATE') THEN
  87     INSERT INTO public.sub1_trigger_activity
  88       SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a;
  89   END IF;
  90   RETURN NULL;
  91 END;
  92 \$\$ LANGUAGE plpgsql;
  93 CREATE TRIGGER sub1_tab1_log_op_trigger
  94   AFTER INSERT OR UPDATE ON tab1
  95   FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
  96 ALTER TABLE ONLY tab1 ENABLE REPLICA TRIGGER sub1_tab1_log_op_trigger;
  97 CREATE TRIGGER sub1_tab1_2_log_op_trigger
  98   AFTER INSERT OR UPDATE ON tab1_2
  99   FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
 100 ALTER TABLE ONLY tab1_2 ENABLE REPLICA TRIGGER sub1_tab1_2_log_op_trigger;
 101 CREATE TRIGGER sub1_tab1_2_2_log_op_trigger
 102   AFTER INSERT OR UPDATE ON tab1_2_2
 103   FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
 104 ALTER TABLE ONLY tab1_2_2 ENABLE REPLICA TRIGGER sub1_tab1_2_2_log_op_trigger;
 105 });
 106
 107 # subscriber 2
 108 #
 109 # This does not use partitioning.  The tables match the leaf tables on
 110 # the publisher.
 111 $node_subscriber2->safe_psql('postgres',
 112         "CREATE TABLE tab1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1', b text)"
 113 );
 114 $node_subscriber2->safe_psql('postgres',
 115         "CREATE TABLE tab1_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_1', b text)"
 116 );
 117 $node_subscriber2->safe_psql('postgres',
 118         "CREATE TABLE tab1_2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_2', b text)"
 119 );
 120 $node_subscriber2->safe_psql('postgres',
 121         "CREATE TABLE tab1_def (a int PRIMARY KEY, b text, c text DEFAULT 'sub2_tab1_def')"
 122 );
 123 $node_subscriber2->safe_psql('postgres',
 124         "CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub_all"
 125 );
 126
 127 # Add set of AFTER replica triggers for testing that they are fired
 128 # correctly, using the same method as the first subscriber.
 129 $node_subscriber2->safe_psql(
 130         'postgres', qq{
 131 CREATE TABLE sub2_trigger_activity (tgtab text,
 132   tgop text, tgwhen text, tglevel text, olda int, newa int);
 133 CREATE FUNCTION sub2_trigger_activity_func() RETURNS TRIGGER AS \$\$
 134 BEGIN
 135   IF (TG_OP = 'INSERT') THEN
 136     INSERT INTO public.sub2_trigger_activity
 137       SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a;
 138   ELSIF (TG_OP = 'UPDATE') THEN
 139     INSERT INTO public.sub2_trigger_activity
 140       SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a;
 141   END IF;
 142   RETURN NULL;
 143 END;
 144 \$\$ LANGUAGE plpgsql;
 145 CREATE TRIGGER sub2_tab1_log_op_trigger
 146   AFTER INSERT OR UPDATE ON tab1
 147   FOR EACH ROW EXECUTE PROCEDURE sub2_trigger_activity_func();
 148 ALTER TABLE ONLY tab1 ENABLE REPLICA TRIGGER sub2_tab1_log_op_trigger;
 149 CREATE TRIGGER sub2_tab1_2_log_op_trigger
 150   AFTER INSERT OR UPDATE ON tab1_2
 151   FOR EACH ROW EXECUTE PROCEDURE sub2_trigger_activity_func();
 152 ALTER TABLE ONLY tab1_2 ENABLE REPLICA TRIGGER sub2_tab1_2_log_op_trigger;
 153 });
 154
 155 # Wait for initial sync of all subscriptions
 156 my $synced_query =
 157   "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
 158 $node_subscriber1->poll_query_until('postgres', $synced_query)
 159   or die "Timed out while waiting for subscriber to synchronize data";
 160 $node_subscriber2->poll_query_until('postgres', $synced_query)
 161   or die "Timed out while waiting for subscriber to synchronize data";
 162
 163 # Tests for replication using leaf partition identity and schema
 164
 165 # insert
 166 $node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (1)");
 167 $node_publisher->safe_psql('postgres', "INSERT INTO tab1_1 (a) VALUES (3)");
 168 $node_publisher->safe_psql('postgres', "INSERT INTO tab1_2 VALUES (5)");
 169 $node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (0)");
 170
 171 $node_publisher->wait_for_catchup('sub1');
 172 $node_publisher->wait_for_catchup('sub2');
 173
 174 my $result = $node_subscriber1->safe_psql('postgres',
 175         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 176 is( $result, qq(sub1_tab1|0
 177 sub1_tab1|1
 178 sub1_tab1|3
 179 sub1_tab1|5), 'inserts into tab1 and its partitions replicated');
 180
 181 $result = $node_subscriber1->safe_psql('postgres',
 182         "SELECT a FROM tab1_2_1 ORDER BY 1");
 183 is($result, qq(5), 'inserts into tab1_2 replicated into tab1_2_1 correctly');
 184
 185 $result = $node_subscriber1->safe_psql('postgres',
 186         "SELECT a FROM tab1_2_2 ORDER BY 1");
 187 is($result, qq(), 'inserts into tab1_2 replicated into tab1_2_2 correctly');
 188
 189 $result = $node_subscriber2->safe_psql('postgres',
 190         "SELECT c, a FROM tab1_1 ORDER BY 1, 2");
 191 is( $result, qq(sub2_tab1_1|1
 192 sub2_tab1_1|3), 'inserts into tab1_1 replicated');
 193
 194 $result = $node_subscriber2->safe_psql('postgres',
 195         "SELECT c, a FROM tab1_2 ORDER BY 1, 2");
 196 is($result, qq(sub2_tab1_2|5), 'inserts into tab1_2 replicated');
 197
 198 # The AFTER trigger of tab1_2 should have recorded one INSERT.
 199 $result = $node_subscriber2->safe_psql('postgres',
 200         "SELECT * FROM sub2_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
 201 );
 202 is( $result,
 203         qq(tab1_2|INSERT|AFTER|ROW||5),
 204         'check replica insert after trigger applied on subscriber');
 205
 206 $result = $node_subscriber2->safe_psql('postgres',
 207         "SELECT c, a FROM tab1_def ORDER BY 1, 2");
 208 is($result, qq(sub2_tab1_def|0), 'inserts into tab1_def replicated');
 209
 210 # update (replicated as update)
 211 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 2 WHERE a = 1");
 212 # All of the following cause an update to be applied to a partitioned
 213 # table on subscriber1: tab1_2 is leaf partition on publisher, whereas
 214 # it's sub-partitioned on subscriber1.
 215 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
 216 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 4 WHERE a = 6");
 217 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 4");
 218
 219 $node_publisher->wait_for_catchup('sub1');
 220 $node_publisher->wait_for_catchup('sub2');
 221
 222 $result = $node_subscriber1->safe_psql('postgres',
 223         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 224 is( $result, qq(sub1_tab1|0
 225 sub1_tab1|2
 226 sub1_tab1|3
 227 sub1_tab1|6), 'update of tab1_1, tab1_2 replicated');
 228
 229 $result = $node_subscriber1->safe_psql('postgres',
 230         "SELECT a FROM tab1_2_1 ORDER BY 1");
 231 is($result, qq(), 'updates of tab1_2 replicated into tab1_2_1 correctly');
 232
 233 $result = $node_subscriber1->safe_psql('postgres',
 234         "SELECT a FROM tab1_2_2 ORDER BY 1");
 235 is($result, qq(6), 'updates of tab1_2 replicated into tab1_2_2 correctly');
 236
 237 # The AFTER trigger should have recorded the UPDATEs of tab1_2_2.
 238 $result = $node_subscriber1->safe_psql('postgres',
 239         "SELECT * FROM sub1_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
 240 );
 241 is( $result, qq(tab1_2_2|INSERT|AFTER|ROW||6
 242 tab1_2_2|UPDATE|AFTER|ROW|4|6
 243 tab1_2_2|UPDATE|AFTER|ROW|6|4),
 244         'check replica update after trigger applied on subscriber');
 245
 246 $result = $node_subscriber2->safe_psql('postgres',
 247         "SELECT c, a FROM tab1_1 ORDER BY 1, 2");
 248 is( $result, qq(sub2_tab1_1|2
 249 sub2_tab1_1|3), 'update of tab1_1 replicated');
 250
 251 $result = $node_subscriber2->safe_psql('postgres',
 252         "SELECT c, a FROM tab1_2 ORDER BY 1, 2");
 253 is($result, qq(sub2_tab1_2|6), 'tab1_2 updated');
 254
 255 # The AFTER trigger should have recorded the updates of tab1_2.
 256 $result = $node_subscriber2->safe_psql('postgres',
 257         "SELECT * FROM sub2_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
 258 );
 259 is( $result, qq(tab1_2|INSERT|AFTER|ROW||5
 260 tab1_2|UPDATE|AFTER|ROW|4|6
 261 tab1_2|UPDATE|AFTER|ROW|5|6
 262 tab1_2|UPDATE|AFTER|ROW|6|4),
 263         'check replica update after trigger applied on subscriber');
 264
 265 $result = $node_subscriber2->safe_psql('postgres',
 266         "SELECT c, a FROM tab1_def ORDER BY 1");
 267 is($result, qq(sub2_tab1_def|0), 'tab1_def unchanged');
 268
 269 # update (replicated as delete+insert)
 270 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 1 WHERE a = 0");
 271 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 4 WHERE a = 1");
 272
 273 $node_publisher->wait_for_catchup('sub1');
 274 $node_publisher->wait_for_catchup('sub2');
 275
 276 $result = $node_subscriber1->safe_psql('postgres',
 277         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 278 is( $result, qq(sub1_tab1|2
 279 sub1_tab1|3
 280 sub1_tab1|4
 281 sub1_tab1|6),
 282         'update of tab1 (delete from tab1_def + insert into tab1_1) replicated');
 283
 284 $result = $node_subscriber1->safe_psql('postgres',
 285         "SELECT a FROM tab1_2_2 ORDER BY 1");
 286 is( $result, qq(4
 287 6), 'updates of tab1 (delete + insert) replicated into tab1_2_2 correctly');
 288
 289 $result = $node_subscriber2->safe_psql('postgres',
 290         "SELECT c, a FROM tab1_1 ORDER BY 1, 2");
 291 is( $result, qq(sub2_tab1_1|2
 292 sub2_tab1_1|3), 'tab1_1 unchanged');
 293
 294 $result = $node_subscriber2->safe_psql('postgres',
 295         "SELECT c, a FROM tab1_2 ORDER BY 1, 2");
 296 is( $result, qq(sub2_tab1_2|4
 297 sub2_tab1_2|6), 'insert into tab1_2 replicated');
 298
 299 $result = $node_subscriber2->safe_psql('postgres',
 300         "SELECT a FROM tab1_def ORDER BY 1");
 301 is($result, qq(), 'delete from tab1_def replicated');
 302
 303 # delete
 304 $node_publisher->safe_psql('postgres',
 305         "DELETE FROM tab1 WHERE a IN (2, 3, 5)");
 306 $node_publisher->safe_psql('postgres', "DELETE FROM tab1_2");
 307
 308 $node_publisher->wait_for_catchup('sub1');
 309 $node_publisher->wait_for_catchup('sub2');
 310
 311 $result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1");
 312 is($result, qq(), 'delete from tab1_1, tab1_2 replicated');
 313
 314 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_1");
 315 is($result, qq(), 'delete from tab1_1 replicated');
 316
 317 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_2");
 318 is($result, qq(), 'delete from tab1_2 replicated');
 319
 320 # truncate
 321 $node_subscriber1->safe_psql('postgres',
 322         "INSERT INTO tab1 (a) VALUES (1), (2), (5)");
 323 $node_subscriber2->safe_psql('postgres', "INSERT INTO tab1_2 (a) VALUES (2)");
 324 $node_publisher->safe_psql('postgres', "TRUNCATE tab1_2");
 325
 326 $node_publisher->wait_for_catchup('sub1');
 327 $node_publisher->wait_for_catchup('sub2');
 328
 329 $result =
 330   $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
 331 is( $result, qq(1
 332 2), 'truncate of tab1_2 replicated');
 333
 334 $result =
 335   $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_2 ORDER BY 1");
 336 is($result, qq(), 'truncate of tab1_2 replicated');
 337
 338 $node_publisher->safe_psql('postgres', "TRUNCATE tab1");
 339
 340 $node_publisher->wait_for_catchup('sub1');
 341 $node_publisher->wait_for_catchup('sub2');
 342
 343 $result =
 344   $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
 345 is($result, qq(), 'truncate of tab1_1 replicated');
 346 $result =
 347   $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
 348 is($result, qq(), 'truncate of tab1 replicated');
 349
 350 # Check that subscriber handles cases where update/delete target tuple
 351 # is missing.  We have to look for the DEBUG1 log messages about that,
 352 # so temporarily bump up the log verbosity.
 353 $node_subscriber1->append_conf('postgresql.conf',
 354         "log_min_messages = debug1");
 355 $node_subscriber1->reload;
 356
 357 $node_publisher->safe_psql('postgres',
 358         "INSERT INTO tab1 VALUES (1, 'foo'), (4, 'bar'), (10, 'baz')");
 359
 360 $node_publisher->wait_for_catchup('sub1');
 361 $node_publisher->wait_for_catchup('sub2');
 362
 363 $node_subscriber1->safe_psql('postgres', "DELETE FROM tab1");
 364
 365 # Note that the current location of the log file is not grabbed immediately
 366 # after reloading the configuration, but after sending one SQL command to
 367 # the node so as we are sure that the reloading has taken effect.
 368 my $log_location = -s $node_subscriber1->logfile;
 369
 370 $node_publisher->safe_psql('postgres',
 371         "UPDATE tab1 SET b = 'quux' WHERE a = 4");
 372 $node_publisher->safe_psql('postgres', "DELETE FROM tab1");
 373
 374 $node_publisher->wait_for_catchup('sub1');
 375 $node_publisher->wait_for_catchup('sub2');
 376
 377 my $logfile = slurp_file($node_subscriber1->logfile(), $log_location);
 378 ok( $logfile =~
 379           qr/logical replication did not find row to be updated in replication target relation's partition "tab1_2_2"/,
 380         'update target row is missing in tab1_2_2');
 381 ok( $logfile =~
 382           qr/logical replication did not find row to be deleted in replication target relation "tab1_1"/,
 383         'delete target row is missing in tab1_1');
 384 ok( $logfile =~
 385           qr/logical replication did not find row to be deleted in replication target relation "tab1_2_2"/,
 386         'delete target row is missing in tab1_2_2');
 387 ok( $logfile =~
 388           qr/logical replication did not find row to be deleted in replication target relation "tab1_def"/,
 389         'delete target row is missing in tab1_def');
 390
 391 $node_subscriber1->append_conf('postgresql.conf',
 392         "log_min_messages = warning");
 393 $node_subscriber1->reload;
 394
 395 # Tests for replication using root table identity and schema
 396
 397 # publisher
 398 $node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
 399 $node_publisher->safe_psql('postgres',
 400         "CREATE TABLE tab2 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
 401 $node_publisher->safe_psql('postgres',
 402         "CREATE TABLE tab2_1 (b text, a int NOT NULL)");
 403 $node_publisher->safe_psql('postgres',
 404         "ALTER TABLE tab2 ATTACH PARTITION tab2_1 FOR VALUES IN (0, 1, 2, 3)");
 405 $node_publisher->safe_psql('postgres',
 406         "CREATE TABLE tab2_2 PARTITION OF tab2 FOR VALUES IN (5, 6)");
 407
 408 $node_publisher->safe_psql('postgres',
 409         "CREATE TABLE tab3 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
 410 $node_publisher->safe_psql('postgres',
 411         "CREATE TABLE tab3_1 PARTITION OF tab3 FOR VALUES IN (0, 1, 2, 3, 5, 6)");
 412 $node_publisher->safe_psql('postgres',
 413         "ALTER PUBLICATION pub_all SET (publish_via_partition_root = true)");
 414 # Note: tab3_1's parent is not in the publication, in which case its
 415 # changes are published using own identity. For tab2, even though both parent
 416 # and child tables are present but changes will be replicated via the parent's
 417 # identity and only once.
 418 $node_publisher->safe_psql('postgres',
 419         "CREATE PUBLICATION pub_viaroot FOR TABLE tab2, tab2_1, tab3_1 WITH (publish_via_partition_root = true)"
 420 );
 421
 422 # prepare data for the initial sync
 423 $node_publisher->safe_psql('postgres', "INSERT INTO tab2 VALUES (1)");
 424
 425 # subscriber 1
 426 $node_subscriber1->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
 427 $node_subscriber1->safe_psql('postgres',
 428         "CREATE TABLE tab2 (a int PRIMARY KEY, c text DEFAULT 'sub1_tab2', b text) PARTITION BY RANGE (a)"
 429 );
 430 $node_subscriber1->safe_psql('postgres',
 431         "CREATE TABLE tab2_1 (c text DEFAULT 'sub1_tab2', b text, a int NOT NULL)"
 432 );
 433 $node_subscriber1->safe_psql('postgres',
 434         "ALTER TABLE tab2 ATTACH PARTITION tab2_1 FOR VALUES FROM (0) TO (10)");
 435 $node_subscriber1->safe_psql('postgres',
 436         "CREATE TABLE tab3_1 (c text DEFAULT 'sub1_tab3_1', b text, a int NOT NULL PRIMARY KEY)"
 437 );
 438 $node_subscriber1->safe_psql('postgres',
 439         "CREATE SUBSCRIPTION sub_viaroot CONNECTION '$publisher_connstr' PUBLICATION pub_viaroot"
 440 );
 441
 442 # subscriber 2
 443 $node_subscriber2->safe_psql('postgres', "DROP TABLE tab1");
 444 $node_subscriber2->safe_psql('postgres',
 445         "CREATE TABLE tab1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1', b text) PARTITION BY HASH (a)"
 446 );
 447 # Note: tab1's partitions are named tab1_1 and tab1_2 on the publisher.
 448 $node_subscriber2->safe_psql('postgres',
 449         "CREATE TABLE tab1_part1 (b text, c text, a int NOT NULL)");
 450 $node_subscriber2->safe_psql('postgres',
 451         "ALTER TABLE tab1 ATTACH PARTITION tab1_part1 FOR VALUES WITH (MODULUS 2, REMAINDER 0)"
 452 );
 453 $node_subscriber2->safe_psql('postgres',
 454         "CREATE TABLE tab1_part2 PARTITION OF tab1 FOR VALUES WITH (MODULUS 2, REMAINDER 1)"
 455 );
 456 $node_subscriber2->safe_psql('postgres',
 457         "CREATE TABLE tab2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab2', b text)"
 458 );
 459 $node_subscriber2->safe_psql('postgres',
 460         "CREATE TABLE tab3 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3', b text)"
 461 );
 462 $node_subscriber2->safe_psql('postgres',
 463         "CREATE TABLE tab3_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3_1', b text)"
 464 );
 465 # Publication that sub2 points to now publishes via root, so must update
 466 # subscription target relations.
 467 $node_subscriber2->safe_psql('postgres',
 468         "ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");
 469
 470 # Wait for initial sync of all subscriptions
 471 $node_subscriber1->poll_query_until('postgres', $synced_query)
 472   or die "Timed out while waiting for subscriber to synchronize data";
 473 $node_subscriber2->poll_query_until('postgres', $synced_query)
 474   or die "Timed out while waiting for subscriber to synchronize data";
 475
 476 # check that data is synced correctly
 477 $result = $node_subscriber1->safe_psql('postgres',
 478         "SELECT c, a FROM tab2");
 479 is( $result, qq(sub1_tab2|1), 'initial data synced for pub_viaroot');
 480
 481 # insert
 482 $node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (1), (0)");
 483 $node_publisher->safe_psql('postgres', "INSERT INTO tab1_1 (a) VALUES (3)");
 484 $node_publisher->safe_psql('postgres', "INSERT INTO tab1_2 VALUES (5)");
 485 $node_publisher->safe_psql('postgres',
 486         "INSERT INTO tab2 VALUES (0), (3), (5)");
 487 $node_publisher->safe_psql('postgres',
 488         "INSERT INTO tab3 VALUES (1), (0), (3), (5)");
 489
 490 $node_publisher->wait_for_catchup('sub_viaroot');
 491 $node_publisher->wait_for_catchup('sub2');
 492
 493 $result = $node_subscriber1->safe_psql('postgres',
 494         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 495 is( $result, qq(sub1_tab2|0
 496 sub1_tab2|1
 497 sub1_tab2|3
 498 sub1_tab2|5), 'inserts into tab2 replicated');
 499
 500 $result = $node_subscriber1->safe_psql('postgres',
 501         "SELECT c, a FROM tab3_1 ORDER BY 1, 2");
 502 is( $result, qq(sub1_tab3_1|0
 503 sub1_tab3_1|1
 504 sub1_tab3_1|3
 505 sub1_tab3_1|5), 'inserts into tab3_1 replicated');
 506
 507 $result = $node_subscriber2->safe_psql('postgres',
 508         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 509 is( $result, qq(sub2_tab1|0
 510 sub2_tab1|1
 511 sub2_tab1|3
 512 sub2_tab1|5), 'inserts into tab1 replicated');
 513
 514 $result = $node_subscriber2->safe_psql('postgres',
 515         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 516 is( $result, qq(sub2_tab2|0
 517 sub2_tab2|1
 518 sub2_tab2|3
 519 sub2_tab2|5), 'inserts into tab2 replicated');
 520
 521 $result = $node_subscriber2->safe_psql('postgres',
 522         "SELECT c, a FROM tab3 ORDER BY 1, 2");
 523 is( $result, qq(sub2_tab3|0
 524 sub2_tab3|1
 525 sub2_tab3|3
 526 sub2_tab3|5), 'inserts into tab3 replicated');
 527
 528 # update (replicated as update)
 529 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
 530 $node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 6 WHERE a = 5");
 531 $node_publisher->safe_psql('postgres', "UPDATE tab3 SET a = 6 WHERE a = 5");
 532
 533 $node_publisher->wait_for_catchup('sub_viaroot');
 534 $node_publisher->wait_for_catchup('sub2');
 535
 536 $result = $node_subscriber1->safe_psql('postgres',
 537         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 538 is( $result, qq(sub1_tab2|0
 539 sub1_tab2|1
 540 sub1_tab2|3
 541 sub1_tab2|6), 'update of tab2 replicated');
 542
 543 $result = $node_subscriber1->safe_psql('postgres',
 544         "SELECT c, a FROM tab3_1 ORDER BY 1, 2");
 545 is( $result, qq(sub1_tab3_1|0
 546 sub1_tab3_1|1
 547 sub1_tab3_1|3
 548 sub1_tab3_1|6), 'update of tab3_1 replicated');
 549
 550 $result = $node_subscriber2->safe_psql('postgres',
 551         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 552 is( $result, qq(sub2_tab1|0
 553 sub2_tab1|1
 554 sub2_tab1|3
 555 sub2_tab1|6), 'inserts into tab1 replicated');
 556
 557 $result = $node_subscriber2->safe_psql('postgres',
 558         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 559 is( $result, qq(sub2_tab2|0
 560 sub2_tab2|1
 561 sub2_tab2|3
 562 sub2_tab2|6), 'inserts into tab2 replicated');
 563
 564 $result = $node_subscriber2->safe_psql('postgres',
 565         "SELECT c, a FROM tab3 ORDER BY 1, 2");
 566 is( $result, qq(sub2_tab3|0
 567 sub2_tab3|1
 568 sub2_tab3|3
 569 sub2_tab3|6), 'inserts into tab3 replicated');
 570
 571 # update (replicated as delete+insert)
 572 $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 2 WHERE a = 6");
 573 $node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 2 WHERE a = 6");
 574 $node_publisher->safe_psql('postgres', "UPDATE tab3 SET a = 2 WHERE a = 6");
 575
 576 $node_publisher->wait_for_catchup('sub_viaroot');
 577 $node_publisher->wait_for_catchup('sub2');
 578
 579 $result = $node_subscriber1->safe_psql('postgres',
 580         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 581 is( $result, qq(sub1_tab2|0
 582 sub1_tab2|1
 583 sub1_tab2|2
 584 sub1_tab2|3), 'update of tab2 replicated');
 585
 586 $result = $node_subscriber1->safe_psql('postgres',
 587         "SELECT c, a FROM tab3_1 ORDER BY 1, 2");
 588 is( $result, qq(sub1_tab3_1|0
 589 sub1_tab3_1|1
 590 sub1_tab3_1|2
 591 sub1_tab3_1|3), 'update of tab3_1 replicated');
 592
 593 $result = $node_subscriber2->safe_psql('postgres',
 594         "SELECT c, a FROM tab1 ORDER BY 1, 2");
 595 is( $result, qq(sub2_tab1|0
 596 sub2_tab1|1
 597 sub2_tab1|2
 598 sub2_tab1|3), 'update of tab1 replicated');
 599
 600 $result = $node_subscriber2->safe_psql('postgres',
 601         "SELECT c, a FROM tab2 ORDER BY 1, 2");
 602 is( $result, qq(sub2_tab2|0
 603 sub2_tab2|1
 604 sub2_tab2|2
 605 sub2_tab2|3), 'update of tab2 replicated');
 606
 607 $result = $node_subscriber2->safe_psql('postgres',
 608         "SELECT c, a FROM tab3 ORDER BY 1, 2");
 609 is( $result, qq(sub2_tab3|0
 610 sub2_tab3|1
 611 sub2_tab3|2
 612 sub2_tab3|3), 'update of tab3 replicated');
 613
 614 # delete
 615 $node_publisher->safe_psql('postgres', "DELETE FROM tab1");
 616 $node_publisher->safe_psql('postgres', "DELETE FROM tab2");
 617 $node_publisher->safe_psql('postgres', "DELETE FROM tab3");
 618
 619 $node_publisher->wait_for_catchup('sub_viaroot');
 620 $node_publisher->wait_for_catchup('sub2');
 621
 622 $result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2");
 623 is($result, qq(), 'delete tab2 replicated');
 624
 625 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1");
 626 is($result, qq(), 'delete from tab1 replicated');
 627
 628 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2");
 629 is($result, qq(), 'delete from tab2 replicated');
 630
 631 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3");
 632 is($result, qq(), 'delete from tab3 replicated');
 633
 634 # truncate
 635 $node_publisher->safe_psql('postgres',
 636         "INSERT INTO tab1 VALUES (1), (2), (5)");
 637 $node_publisher->safe_psql('postgres',
 638         "INSERT INTO tab2 VALUES (1), (2), (5)");
 639 # these will NOT be replicated
 640 $node_publisher->safe_psql('postgres', "TRUNCATE tab1_2, tab2_1, tab3_1");
 641
 642 $node_publisher->wait_for_catchup('sub_viaroot');
 643 $node_publisher->wait_for_catchup('sub2');
 644
 645 $result =
 646   $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2 ORDER BY 1");
 647 is( $result, qq(1
 648 2
 649 5), 'truncate of tab2_1 NOT replicated');
 650
 651 $result =
 652   $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
 653 is( $result, qq(1
 654 2
 655 5), 'truncate of tab1_2 NOT replicated');
 656
 657 $result =
 658   $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2 ORDER BY 1");
 659 is( $result, qq(1
 660 2
 661 5), 'truncate of tab2_1 NOT replicated');
 662
 663 $node_publisher->safe_psql('postgres', "TRUNCATE tab1, tab2, tab3");
 664
 665 $node_publisher->wait_for_catchup('sub_viaroot');
 666 $node_publisher->wait_for_catchup('sub2');
 667
 668 $result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2");
 669 is($result, qq(), 'truncate of tab2 replicated');
 670
 671 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1");
 672 is($result, qq(), 'truncate of tab1 replicated');
 673
 674 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2");
 675 is($result, qq(), 'truncate of tab2 replicated');
 676
 677 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3");
 678 is($result, qq(), 'truncate of tab3 replicated');
 679
 680 $result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3_1");
 681 is($result, qq(), 'truncate of tab3_1 replicated');
 682
 683 # check that the map to convert tuples from leaf partition to the root
 684 # table is correctly rebuilt when a new column is added
 685 $node_publisher->safe_psql('postgres',
 686         "ALTER TABLE tab2 DROP b, ADD COLUMN c text DEFAULT 'pub_tab2', ADD b text"
 687 );
 688 $node_publisher->safe_psql('postgres',
 689         "INSERT INTO tab2 (a, b) VALUES (1, 'xxx'), (3, 'yyy'), (5, 'zzz')");
 690 $node_publisher->safe_psql('postgres',
 691         "INSERT INTO tab2 (a, b, c) VALUES (6, 'aaa', 'xxx_c')");
 692
 693 $node_publisher->wait_for_catchup('sub_viaroot');
 694 $node_publisher->wait_for_catchup('sub2');
 695
 696 $result = $node_subscriber1->safe_psql('postgres',
 697         "SELECT c, a, b FROM tab2 ORDER BY 1, 2");
 698 is( $result, qq(pub_tab2|1|xxx
 699 pub_tab2|3|yyy
 700 pub_tab2|5|zzz
 701 xxx_c|6|aaa), 'inserts into tab2 replicated');
 702
 703 $result = $node_subscriber2->safe_psql('postgres',
 704         "SELECT c, a, b FROM tab2 ORDER BY 1, 2");
 705 is( $result, qq(pub_tab2|1|xxx
 706 pub_tab2|3|yyy
 707 pub_tab2|5|zzz
 708 xxx_c|6|aaa), 'inserts into tab2 replicated');
 709
 710 # Check that subscriber handles cases where update/delete target tuple
 711 # is missing.  We have to look for the DEBUG1 log messages about that,
 712 # so temporarily bump up the log verbosity.
 713 $node_subscriber1->append_conf('postgresql.conf',
 714         "log_min_messages = debug1");
 715 $node_subscriber1->reload;
 716
 717 $node_subscriber1->safe_psql('postgres', "DELETE FROM tab2");
 718
 719 # Note that the current location of the log file is not grabbed immediately
 720 # after reloading the configuration, but after sending one SQL command to
 721 # the node so as we are sure that the reloading has taken effect.
 722 $log_location = -s $node_subscriber1->logfile;
 723
 724 $node_publisher->safe_psql('postgres',
 725         "UPDATE tab2 SET b = 'quux' WHERE a = 5");
 726 $node_publisher->safe_psql('postgres', "DELETE FROM tab2 WHERE a = 1");
 727
 728 $node_publisher->wait_for_catchup('sub_viaroot');
 729 $node_publisher->wait_for_catchup('sub2');
 730
 731 $logfile = slurp_file($node_subscriber1->logfile(), $log_location);
 732 ok( $logfile =~
 733           qr/logical replication did not find row to be updated in replication target relation's partition "tab2_1"/,
 734         'update target row is missing in tab2_1');
 735 ok( $logfile =~
 736           qr/logical replication did not find row to be deleted in replication target relation "tab2_1"/,
 737         'delete target row is missing in tab2_1');
 738
 739 # No need for this until more tests are added.
 740 # $node_subscriber1->append_conf('postgresql.conf',
 741 #       "log_min_messages = warning");
 742 # $node_subscriber1->reload;