# Copyright (c) 2021-2022, PostgreSQL Global Development Group

# Test logical replication with partitioned tables
use strict;
use warnings;
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;

# setup

my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;

my $node_subscriber1 = PostgreSQL::Test::Cluster->new('subscriber1');
$node_subscriber1->init(allows_streaming => 'logical');
$node_subscriber1->start;

my $node_subscriber2 = PostgreSQL::Test::Cluster->new('subscriber2');
$node_subscriber2->init(allows_streaming => 'logical');
$node_subscriber2->start;

my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';

# publisher
$node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub1");
$node_publisher->safe_psql('postgres',
	"CREATE PUBLICATION pub_all FOR ALL TABLES");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab1 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab1_1 (b text, a int NOT NULL)");
$node_publisher->safe_psql('postgres',
	"ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab1_2 PARTITION OF tab1 FOR VALUES IN (4, 5, 6)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab1_def PARTITION OF tab1 DEFAULT");
$node_publisher->safe_psql('postgres',
	"ALTER PUBLICATION pub1 ADD TABLE tab1, tab1_1");

# subscriber1
#
# This is partitioned differently from the publisher.  tab1_2 is
# subpartitioned.  This tests the tuple routing code on the
# subscriber.
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1 (c text, a int PRIMARY KEY, b text) PARTITION BY LIST (a)"
);
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1_1 (b text, c text DEFAULT 'sub1_tab1', a int NOT NULL)"
);
$node_subscriber1->safe_psql('postgres',
	"ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3)");
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1_2 PARTITION OF tab1 (c DEFAULT 'sub1_tab1') FOR VALUES IN (4, 5, 6) PARTITION BY LIST (a)"
);
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1_2_1 (c text, b text, a int NOT NULL)");
$node_subscriber1->safe_psql('postgres',
	"ALTER TABLE tab1_2 ATTACH PARTITION tab1_2_1 FOR VALUES IN (5)");
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1_2_2 PARTITION OF tab1_2 FOR VALUES IN (4, 6)");
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab1_def PARTITION OF tab1 (c DEFAULT 'sub1_tab1') DEFAULT"
);
$node_subscriber1->safe_psql('postgres',
	"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
);

# Add set of AFTER replica triggers for testing that they are fired
# correctly.  This uses a table that records details of all trigger
# activities.  Triggers are marked as enabled for a subset of the
# partition tree.
$node_subscriber1->safe_psql(
	'postgres', qq{
CREATE TABLE sub1_trigger_activity (tgtab text, tgop text,
  tgwhen text, tglevel text, olda int, newa int);
CREATE FUNCTION sub1_trigger_activity_func() RETURNS TRIGGER AS \$\$
BEGIN
  IF (TG_OP = 'INSERT') THEN
    INSERT INTO public.sub1_trigger_activity
      SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a;
  ELSIF (TG_OP = 'UPDATE') THEN
    INSERT INTO public.sub1_trigger_activity
      SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a;
  END IF;
  RETURN NULL;
END;
\$\$ LANGUAGE plpgsql;
CREATE TRIGGER sub1_tab1_log_op_trigger
  AFTER INSERT OR UPDATE ON tab1
  FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
ALTER TABLE ONLY tab1 ENABLE REPLICA TRIGGER sub1_tab1_log_op_trigger;
CREATE TRIGGER sub1_tab1_2_log_op_trigger
  AFTER INSERT OR UPDATE ON tab1_2
  FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
ALTER TABLE ONLY tab1_2 ENABLE REPLICA TRIGGER sub1_tab1_2_log_op_trigger;
CREATE TRIGGER sub1_tab1_2_2_log_op_trigger
  AFTER INSERT OR UPDATE ON tab1_2_2
  FOR EACH ROW EXECUTE PROCEDURE sub1_trigger_activity_func();
ALTER TABLE ONLY tab1_2_2 ENABLE REPLICA TRIGGER sub1_tab1_2_2_log_op_trigger;
});

# subscriber 2
#
# This does not use partitioning.  The tables match the leaf tables on
# the publisher.
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1', b text)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_1', b text)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1_2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_2', b text)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1_def (a int PRIMARY KEY, b text, c text DEFAULT 'sub2_tab1_def')"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub_all"
);

# Add set of AFTER replica triggers for testing that they are fired
# correctly, using the same method as the first subscriber.
$node_subscriber2->safe_psql(
	'postgres', qq{
CREATE TABLE sub2_trigger_activity (tgtab text,
  tgop text, tgwhen text, tglevel text, olda int, newa int);
CREATE FUNCTION sub2_trigger_activity_func() RETURNS TRIGGER AS \$\$
BEGIN
  IF (TG_OP = 'INSERT') THEN
    INSERT INTO public.sub2_trigger_activity
      SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, NULL, NEW.a;
  ELSIF (TG_OP = 'UPDATE') THEN
    INSERT INTO public.sub2_trigger_activity
      SELECT TG_RELNAME, TG_OP, TG_WHEN, TG_LEVEL, OLD.a, NEW.a;
  END IF;
  RETURN NULL;
END;
\$\$ LANGUAGE plpgsql;
CREATE TRIGGER sub2_tab1_log_op_trigger
  AFTER INSERT OR UPDATE ON tab1
  FOR EACH ROW EXECUTE PROCEDURE sub2_trigger_activity_func();
ALTER TABLE ONLY tab1 ENABLE REPLICA TRIGGER sub2_tab1_log_op_trigger;
CREATE TRIGGER sub2_tab1_2_log_op_trigger
  AFTER INSERT OR UPDATE ON tab1_2
  FOR EACH ROW EXECUTE PROCEDURE sub2_trigger_activity_func();
ALTER TABLE ONLY tab1_2 ENABLE REPLICA TRIGGER sub2_tab1_2_log_op_trigger;
});

# Wait for initial sync of all subscriptions
$node_subscriber1->wait_for_subscription_sync;
$node_subscriber2->wait_for_subscription_sync;

# Tests for replication using leaf partition identity and schema

# insert
$node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (1)");
$node_publisher->safe_psql('postgres', "INSERT INTO tab1_1 (a) VALUES (3)");
$node_publisher->safe_psql('postgres', "INSERT INTO tab1_2 VALUES (5)");
$node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (0)");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

my $result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub1_tab1|0
sub1_tab1|1
sub1_tab1|3
sub1_tab1|5), 'inserts into tab1 and its partitions replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT a FROM tab1_2_1 ORDER BY 1");
is($result, qq(5), 'inserts into tab1_2 replicated into tab1_2_1 correctly');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT a FROM tab1_2_2 ORDER BY 1");
is($result, qq(), 'inserts into tab1_2 replicated into tab1_2_2 correctly');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1_1|1
sub2_tab1_1|3), 'inserts into tab1_1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_2 ORDER BY 1, 2");
is($result, qq(sub2_tab1_2|5), 'inserts into tab1_2 replicated');

# The AFTER trigger of tab1_2 should have recorded one INSERT.
$result = $node_subscriber2->safe_psql('postgres',
	"SELECT * FROM sub2_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
);
is( $result,
	qq(tab1_2|INSERT|AFTER|ROW||5),
	'check replica insert after trigger applied on subscriber');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_def ORDER BY 1, 2");
is($result, qq(sub2_tab1_def|0), 'inserts into tab1_def replicated');

# update (replicated as update)
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 2 WHERE a = 1");
# All of the following cause an update to be applied to a partitioned
# table on subscriber1: tab1_2 is leaf partition on publisher, whereas
# it's sub-partitioned on subscriber1.
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 4 WHERE a = 6");
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 4");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub1_tab1|0
sub1_tab1|2
sub1_tab1|3
sub1_tab1|6), 'update of tab1_1, tab1_2 replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT a FROM tab1_2_1 ORDER BY 1");
is($result, qq(), 'updates of tab1_2 replicated into tab1_2_1 correctly');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT a FROM tab1_2_2 ORDER BY 1");
is($result, qq(6), 'updates of tab1_2 replicated into tab1_2_2 correctly');

# The AFTER trigger should have recorded the UPDATEs of tab1_2_2.
$result = $node_subscriber1->safe_psql('postgres',
	"SELECT * FROM sub1_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
);
is( $result, qq(tab1_2_2|INSERT|AFTER|ROW||6
tab1_2_2|UPDATE|AFTER|ROW|4|6
tab1_2_2|UPDATE|AFTER|ROW|6|4),
	'check replica update after trigger applied on subscriber');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1_1|2
sub2_tab1_1|3), 'update of tab1_1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_2 ORDER BY 1, 2");
is($result, qq(sub2_tab1_2|6), 'tab1_2 updated');

# The AFTER trigger should have recorded the updates of tab1_2.
$result = $node_subscriber2->safe_psql('postgres',
	"SELECT * FROM sub2_trigger_activity ORDER BY tgtab, tgop, tgwhen, olda, newa;"
);
is( $result, qq(tab1_2|INSERT|AFTER|ROW||5
tab1_2|UPDATE|AFTER|ROW|4|6
tab1_2|UPDATE|AFTER|ROW|5|6
tab1_2|UPDATE|AFTER|ROW|6|4),
	'check replica update after trigger applied on subscriber');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_def ORDER BY 1");
is($result, qq(sub2_tab1_def|0), 'tab1_def unchanged');

# update (replicated as delete+insert)
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 1 WHERE a = 0");
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 4 WHERE a = 1");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub1_tab1|2
sub1_tab1|3
sub1_tab1|4
sub1_tab1|6),
	'update of tab1 (delete from tab1_def + insert into tab1_1) replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT a FROM tab1_2_2 ORDER BY 1");
is( $result, qq(4
6), 'updates of tab1 (delete + insert) replicated into tab1_2_2 correctly');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1_1|2
sub2_tab1_1|3), 'tab1_1 unchanged');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1_2 ORDER BY 1, 2");
is( $result, qq(sub2_tab1_2|4
sub2_tab1_2|6), 'insert into tab1_2 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT a FROM tab1_def ORDER BY 1");
is($result, qq(), 'delete from tab1_def replicated');

# delete
$node_publisher->safe_psql('postgres',
	"DELETE FROM tab1 WHERE a IN (2, 3, 5)");
$node_publisher->safe_psql('postgres', "DELETE FROM tab1_2");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1");
is($result, qq(), 'delete from tab1_1, tab1_2 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_1");
is($result, qq(), 'delete from tab1_1 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_2");
is($result, qq(), 'delete from tab1_2 replicated');

# truncate
$node_subscriber1->safe_psql('postgres',
	"INSERT INTO tab1 (a) VALUES (1), (2), (5)");
$node_subscriber2->safe_psql('postgres', "INSERT INTO tab1_2 (a) VALUES (2)");
$node_publisher->safe_psql('postgres', "TRUNCATE tab1_2");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$result =
  $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
is( $result, qq(1
2), 'truncate of tab1_2 replicated');

$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1_2 ORDER BY 1");
is($result, qq(), 'truncate of tab1_2 replicated');

$node_publisher->safe_psql('postgres', "TRUNCATE tab1");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$result =
  $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
is($result, qq(), 'truncate of tab1_1 replicated');
$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
is($result, qq(), 'truncate of tab1 replicated');

# Check that subscriber handles cases where update/delete target tuple
# is missing.  We have to look for the DEBUG1 log messages about that,
# so temporarily bump up the log verbosity.
$node_subscriber1->append_conf('postgresql.conf',
	"log_min_messages = debug1");
$node_subscriber1->reload;

$node_publisher->safe_psql('postgres',
	"INSERT INTO tab1 VALUES (1, 'foo'), (4, 'bar'), (10, 'baz')");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

$node_subscriber1->safe_psql('postgres', "DELETE FROM tab1");

# Note that the current location of the log file is not grabbed immediately
# after reloading the configuration, but after sending one SQL command to
# the node so as we are sure that the reloading has taken effect.
my $log_location = -s $node_subscriber1->logfile;

$node_publisher->safe_psql('postgres',
	"UPDATE tab1 SET b = 'quux' WHERE a = 4");
$node_publisher->safe_psql('postgres', "DELETE FROM tab1");

$node_publisher->wait_for_catchup('sub1');
$node_publisher->wait_for_catchup('sub2');

my $logfile = slurp_file($node_subscriber1->logfile(), $log_location);
ok( $logfile =~
	  qr/logical replication did not find row to be updated in replication target relation's partition "tab1_2_2"/,
	'update target row is missing in tab1_2_2');
ok( $logfile =~
	  qr/logical replication did not find row to be deleted in replication target relation "tab1_1"/,
	'delete target row is missing in tab1_1');
ok( $logfile =~
	  qr/logical replication did not find row to be deleted in replication target relation "tab1_2_2"/,
	'delete target row is missing in tab1_2_2');
ok( $logfile =~
	  qr/logical replication did not find row to be deleted in replication target relation "tab1_def"/,
	'delete target row is missing in tab1_def');

$node_subscriber1->append_conf('postgresql.conf',
	"log_min_messages = warning");
$node_subscriber1->reload;

# Tests for replication using root table identity and schema

# publisher
$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab2 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab2_1 (b text, a int NOT NULL)");
$node_publisher->safe_psql('postgres',
	"ALTER TABLE tab2 ATTACH PARTITION tab2_1 FOR VALUES IN (0, 1, 2, 3)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab2_2 PARTITION OF tab2 FOR VALUES IN (5, 6)");

$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab3 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab3_1 PARTITION OF tab3 FOR VALUES IN (0, 1, 2, 3, 5, 6)");

$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab4 (a int PRIMARY KEY) PARTITION BY LIST (a)");
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab4_1 PARTITION OF tab4 FOR VALUES IN (0, 1) PARTITION BY LIST (a)"
);
$node_publisher->safe_psql('postgres',
	"CREATE TABLE tab4_1_1 PARTITION OF tab4_1 FOR VALUES IN (0, 1)");

$node_publisher->safe_psql('postgres',
	"ALTER PUBLICATION pub_all SET (publish_via_partition_root = true)");
# Note: tab3_1's parent is not in the publication, in which case its
# changes are published using own identity. For tab2, even though both parent
# and child tables are present but changes will be replicated via the parent's
# identity and only once.
$node_publisher->safe_psql('postgres',
	"CREATE PUBLICATION pub_viaroot FOR TABLE tab2, tab2_1, tab3_1 WITH (publish_via_partition_root = true)"
);

# for tab4, we publish changes through the "middle" partitioned table
$node_publisher->safe_psql('postgres',
	"CREATE PUBLICATION pub_lower_level FOR TABLE tab4_1 WITH (publish_via_partition_root = true)"
);

# prepare data for the initial sync
$node_publisher->safe_psql('postgres', "INSERT INTO tab2 VALUES (1)");

# subscriber 1
$node_subscriber1->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab2 (a int PRIMARY KEY, c text DEFAULT 'sub1_tab2', b text) PARTITION BY RANGE (a)"
);
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab2_1 (c text DEFAULT 'sub1_tab2', b text, a int NOT NULL)"
);
$node_subscriber1->safe_psql('postgres',
	"ALTER TABLE tab2 ATTACH PARTITION tab2_1 FOR VALUES FROM (0) TO (10)");
$node_subscriber1->safe_psql('postgres',
	"CREATE TABLE tab3_1 (c text DEFAULT 'sub1_tab3_1', b text, a int NOT NULL PRIMARY KEY)"
);
$node_subscriber1->safe_psql('postgres',
	"CREATE SUBSCRIPTION sub_viaroot CONNECTION '$publisher_connstr' PUBLICATION pub_viaroot"
);

# subscriber 2
$node_subscriber2->safe_psql('postgres', "DROP TABLE tab1");
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1', b text) PARTITION BY HASH (a)"
);
# Note: tab1's partitions are named tab1_1 and tab1_2 on the publisher.
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1_part1 (b text, c text, a int NOT NULL)");
$node_subscriber2->safe_psql('postgres',
	"ALTER TABLE tab1 ATTACH PARTITION tab1_part1 FOR VALUES WITH (MODULUS 2, REMAINDER 0)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab1_part2 PARTITION OF tab1 FOR VALUES WITH (MODULUS 2, REMAINDER 1)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab2', b text)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab3 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3', b text)"
);
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab3_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3_1', b text)"
);

# Note: We create two separate tables, not a partitioned one, so that we can
# easily identity through which relation were the changes replicated.
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab4 (a int PRIMARY KEY)");
$node_subscriber2->safe_psql('postgres',
	"CREATE TABLE tab4_1 (a int PRIMARY KEY)");
# Publication that sub2 points to now publishes via root, so must update
# subscription target relations. We set the list of publications so that
# the FOR ALL TABLES publication is second (the list order matters).
$node_subscriber2->safe_psql('postgres',
	"ALTER SUBSCRIPTION sub2 SET PUBLICATION pub_lower_level, pub_all");

# Wait for initial sync of all subscriptions
$node_subscriber1->wait_for_subscription_sync;
$node_subscriber2->wait_for_subscription_sync;

# check that data is synced correctly
$result = $node_subscriber1->safe_psql('postgres', "SELECT c, a FROM tab2");
is($result, qq(sub1_tab2|1), 'initial data synced for pub_viaroot');

# insert
$node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (1), (0)");
$node_publisher->safe_psql('postgres', "INSERT INTO tab1_1 (a) VALUES (3)");
$node_publisher->safe_psql('postgres', "INSERT INTO tab1_2 VALUES (5)");
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab2 VALUES (0), (3), (5)");
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab3 VALUES (1), (0), (3), (5)");

# Insert a row into the leaf partition, should be replicated through the
# partition root (thanks to the FOR ALL TABLES partition).
$node_publisher->safe_psql('postgres', "INSERT INTO tab4 VALUES (0)");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub1_tab2|0
sub1_tab2|1
sub1_tab2|3
sub1_tab2|5), 'inserts into tab2 replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab3_1 ORDER BY 1, 2");
is( $result, qq(sub1_tab3_1|0
sub1_tab3_1|1
sub1_tab3_1|3
sub1_tab3_1|5), 'inserts into tab3_1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1|0
sub2_tab1|1
sub2_tab1|3
sub2_tab1|5), 'inserts into tab1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub2_tab2|0
sub2_tab2|1
sub2_tab2|3
sub2_tab2|5), 'inserts into tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab3 ORDER BY 1, 2");
is( $result, qq(sub2_tab3|0
sub2_tab3|1
sub2_tab3|3
sub2_tab3|5), 'inserts into tab3 replicated');

# tab4 change should be replicated through the root partition, which
# maps to the tab4 relation on subscriber.
$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4 ORDER BY 1");
is($result, qq(0), 'inserts into tab4 replicated');

$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4_1 ORDER BY 1");
is($result, qq(), 'inserts into tab4_1 replicated');


# now switch the order of publications in the list, try again, the result
# should be the same (no dependence on order of pulications)
$node_subscriber2->safe_psql('postgres',
	"ALTER SUBSCRIPTION sub2 SET PUBLICATION pub_all, pub_lower_level");

# make sure the subscription on the second subscriber is synced, before
# continuing
$node_subscriber2->wait_for_subscription_sync;

# Insert a change into the leaf partition, should be replicated through
# the partition root (thanks to the FOR ALL TABLES partition).
$node_publisher->safe_psql('postgres', "INSERT INTO tab4 VALUES (1)");

$node_publisher->wait_for_catchup('sub2');

# tab4 change should be replicated through the root partition, which
# maps to the tab4 relation on subscriber.
$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4 ORDER BY 1");
is( $result, qq(0
1), 'inserts into tab4 replicated');

$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4_1 ORDER BY 1");
is($result, qq(), 'inserts into tab4_1 replicated');


# update (replicated as update)
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
$node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 6 WHERE a = 5");
$node_publisher->safe_psql('postgres', "UPDATE tab3 SET a = 6 WHERE a = 5");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub1_tab2|0
sub1_tab2|1
sub1_tab2|3
sub1_tab2|6), 'update of tab2 replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab3_1 ORDER BY 1, 2");
is( $result, qq(sub1_tab3_1|0
sub1_tab3_1|1
sub1_tab3_1|3
sub1_tab3_1|6), 'update of tab3_1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1|0
sub2_tab1|1
sub2_tab1|3
sub2_tab1|6), 'inserts into tab1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub2_tab2|0
sub2_tab2|1
sub2_tab2|3
sub2_tab2|6), 'inserts into tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab3 ORDER BY 1, 2");
is( $result, qq(sub2_tab3|0
sub2_tab3|1
sub2_tab3|3
sub2_tab3|6), 'inserts into tab3 replicated');

# update (replicated as delete+insert)
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 2 WHERE a = 6");
$node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 2 WHERE a = 6");
$node_publisher->safe_psql('postgres', "UPDATE tab3 SET a = 2 WHERE a = 6");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub1_tab2|0
sub1_tab2|1
sub1_tab2|2
sub1_tab2|3), 'update of tab2 replicated');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a FROM tab3_1 ORDER BY 1, 2");
is( $result, qq(sub1_tab3_1|0
sub1_tab3_1|1
sub1_tab3_1|2
sub1_tab3_1|3), 'update of tab3_1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab1 ORDER BY 1, 2");
is( $result, qq(sub2_tab1|0
sub2_tab1|1
sub2_tab1|2
sub2_tab1|3), 'update of tab1 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab2 ORDER BY 1, 2");
is( $result, qq(sub2_tab2|0
sub2_tab2|1
sub2_tab2|2
sub2_tab2|3), 'update of tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a FROM tab3 ORDER BY 1, 2");
is( $result, qq(sub2_tab3|0
sub2_tab3|1
sub2_tab3|2
sub2_tab3|3), 'update of tab3 replicated');

# delete
$node_publisher->safe_psql('postgres', "DELETE FROM tab1");
$node_publisher->safe_psql('postgres', "DELETE FROM tab2");
$node_publisher->safe_psql('postgres', "DELETE FROM tab3");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2");
is($result, qq(), 'delete tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1");
is($result, qq(), 'delete from tab1 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2");
is($result, qq(), 'delete from tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3");
is($result, qq(), 'delete from tab3 replicated');

# truncate
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab1 VALUES (1), (2), (5)");
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab2 VALUES (1), (2), (5)");
# these will NOT be replicated
$node_publisher->safe_psql('postgres', "TRUNCATE tab1_2, tab2_1, tab3_1");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result =
  $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2 ORDER BY 1");
is( $result, qq(1
2
5), 'truncate of tab2_1 NOT replicated');

$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
is( $result, qq(1
2
5), 'truncate of tab1_2 NOT replicated');

$result =
  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2 ORDER BY 1");
is( $result, qq(1
2
5), 'truncate of tab2_1 NOT replicated');

$node_publisher->safe_psql('postgres', "TRUNCATE tab1, tab2, tab3");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres', "SELECT a FROM tab2");
is($result, qq(), 'truncate of tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1");
is($result, qq(), 'truncate of tab1 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab2");
is($result, qq(), 'truncate of tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3");
is($result, qq(), 'truncate of tab3 replicated');

$result = $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab3_1");
is($result, qq(), 'truncate of tab3_1 replicated');

# check that the map to convert tuples from leaf partition to the root
# table is correctly rebuilt when a new column is added
$node_publisher->safe_psql('postgres',
	"ALTER TABLE tab2 DROP b, ADD COLUMN c text DEFAULT 'pub_tab2', ADD b text"
);
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab2 (a, b) VALUES (1, 'xxx'), (3, 'yyy'), (5, 'zzz')");
$node_publisher->safe_psql('postgres',
	"INSERT INTO tab2 (a, b, c) VALUES (6, 'aaa', 'xxx_c')");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber1->safe_psql('postgres',
	"SELECT c, a, b FROM tab2 ORDER BY 1, 2");
is( $result, qq(pub_tab2|1|xxx
pub_tab2|3|yyy
pub_tab2|5|zzz
xxx_c|6|aaa), 'inserts into tab2 replicated');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT c, a, b FROM tab2 ORDER BY 1, 2");
is( $result, qq(pub_tab2|1|xxx
pub_tab2|3|yyy
pub_tab2|5|zzz
xxx_c|6|aaa), 'inserts into tab2 replicated');

# Check that subscriber handles cases where update/delete target tuple
# is missing.  We have to look for the DEBUG1 log messages about that,
# so temporarily bump up the log verbosity.
$node_subscriber1->append_conf('postgresql.conf',
	"log_min_messages = debug1");
$node_subscriber1->reload;

$node_subscriber1->safe_psql('postgres', "DELETE FROM tab2");

# Note that the current location of the log file is not grabbed immediately
# after reloading the configuration, but after sending one SQL command to
# the node so as we are sure that the reloading has taken effect.
$log_location = -s $node_subscriber1->logfile;

$node_publisher->safe_psql('postgres',
	"UPDATE tab2 SET b = 'quux' WHERE a = 5");
$node_publisher->safe_psql('postgres', "DELETE FROM tab2 WHERE a = 1");

$node_publisher->wait_for_catchup('sub_viaroot');
$node_publisher->wait_for_catchup('sub2');

$logfile = slurp_file($node_subscriber1->logfile(), $log_location);
ok( $logfile =~
	  qr/logical replication did not find row to be updated in replication target relation's partition "tab2_1"/,
	'update target row is missing in tab2_1');
ok( $logfile =~
	  qr/logical replication did not find row to be deleted in replication target relation "tab2_1"/,
	'delete target row is missing in tab2_1');

$node_subscriber1->append_conf('postgresql.conf',
	"log_min_messages = warning");
$node_subscriber1->reload;

# Test that replication continues to work correctly after altering the
# partition of a partitioned target table.

$node_publisher->safe_psql(
	'postgres', q{
	CREATE TABLE tab5 (a int NOT NULL, b int);
	CREATE UNIQUE INDEX tab5_a_idx ON tab5 (a);
	ALTER TABLE tab5 REPLICA IDENTITY USING INDEX tab5_a_idx;});

$node_subscriber2->safe_psql(
	'postgres', q{
	CREATE TABLE tab5 (a int NOT NULL, b int, c int) PARTITION BY LIST (a);
	CREATE TABLE tab5_1 PARTITION OF tab5 DEFAULT;
	CREATE UNIQUE INDEX tab5_a_idx ON tab5 (a);
	ALTER TABLE tab5 REPLICA IDENTITY USING INDEX tab5_a_idx;
	ALTER TABLE tab5_1 REPLICA IDENTITY USING INDEX tab5_1_a_idx;});

$node_subscriber2->safe_psql('postgres',
	"ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");

$node_subscriber2->wait_for_subscription_sync;

# Make partition map cache
$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (1, 1)");
$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 2 WHERE a = 1");

$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT a, b FROM tab5 ORDER BY 1");
is($result, qq(2|1), 'updates of tab5 replicated correctly');

# Change the column order of partition on subscriber
$node_subscriber2->safe_psql(
	'postgres', q{
	ALTER TABLE tab5 DETACH PARTITION tab5_1;
	ALTER TABLE tab5_1 DROP COLUMN b;
	ALTER TABLE tab5_1 ADD COLUMN b int;
	ALTER TABLE tab5 ATTACH PARTITION tab5_1 DEFAULT});

$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 3 WHERE a = 2");

$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');

# Test that replication works correctly as long as the leaf partition
# has the necessary REPLICA IDENTITY, even though the actual target
# partitioned table does not.
$node_subscriber2->safe_psql('postgres',
	"ALTER TABLE tab5 REPLICA IDENTITY NOTHING");

$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 4 WHERE a = 3");

$node_publisher->wait_for_catchup('sub2');

$result = $node_subscriber2->safe_psql('postgres',
	"SELECT a, b, c FROM tab5_1 ORDER BY 1");
is($result, qq(4||1), 'updates of tab5 replicated correctly');

done_testing();