--- /dev/null
+# Copyright (c) 2022-2023, PostgreSQL Global Development Group
+
+# Test logical replication behavior with subscriber using available index
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname = 'tap_sub';
+my $result = '';
+
+# =============================================================================
+# Testcase start: Subscription can use index with multiple rows and columns
+#
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int, y text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE test_replica_id_full REPLICA IDENTITY FULL");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int, y text)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX test_replica_id_full_idx ON test_replica_id_full(x,y)");
+
+# insert some initial data within the range 0-9 for x and y
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_replica_id_full SELECT (i%10), (i%10)::text FROM generate_series(0,10) i"
+);
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE test_replica_id_full");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# delete 2 rows
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM test_replica_id_full WHERE x IN (5, 6)");
+
+# update 2 rows
+$node_publisher->safe_psql('postgres',
+ "UPDATE test_replica_id_full SET x = 100, y = '200' WHERE x IN (1, 2)");
+
+# wait until the index is used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->poll_query_until(
+ 'postgres', q{select (idx_scan = 4) from pg_stat_all_indexes where indexrelname = 'test_replica_id_full_idx';}
+) or die "Timed out while waiting for check subscriber tap_sub_rep_full updates 4 rows via index";
+
+# make sure that the subscriber has the correct data after the UPDATE
+$result = $node_subscriber->safe_psql('postgres',
+ "select count(*) from test_replica_id_full WHERE (x = 100 and y = '200')");
+is($result, qq(2), 'ensure subscriber has the correct data at the end of the test');
+
+# make sure that the subscriber has the correct data after the first DELETE
+$result = $node_subscriber->safe_psql('postgres',
+ "select count(*) from test_replica_id_full where x in (5, 6)");
+is($result, qq(0), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+
+# Testcase end: Subscription can use index with multiple rows and columns
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Subscription can use index on partitioned tables
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE users_table_part(user_id bigint, value_1 int, value_2 int) PARTITION BY RANGE (value_1)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE users_table_part_0 PARTITION OF users_table_part FOR VALUES FROM (0) TO (10)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE users_table_part_1 PARTITION OF users_table_part FOR VALUES FROM (10) TO (20)"
+);
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE users_table_part REPLICA IDENTITY FULL");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE users_table_part_0 REPLICA IDENTITY FULL");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE users_table_part_1 REPLICA IDENTITY FULL");
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE users_table_part(user_id bigint, value_1 int, value_2 int) PARTITION BY RANGE (value_1)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE users_table_part_0 PARTITION OF users_table_part FOR VALUES FROM (0) TO (10)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE users_table_part_1 PARTITION OF users_table_part FOR VALUES FROM (10) TO (20)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX users_table_part_idx ON users_table_part(user_id, value_1)"
+);
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO users_table_part SELECT (i%100), (i%20), i FROM generate_series(0,100) i"
+);
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE users_table_part");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# update rows, moving them to other partitions
+$node_publisher->safe_psql('postgres',
+ "UPDATE users_table_part SET value_1 = 0 WHERE user_id = 4");
+
+# delete rows from different partitions
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM users_table_part WHERE user_id = 1 and value_1 = 1");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM users_table_part WHERE user_id = 12 and value_1 = 12");
+
+# wait until the index is used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->poll_query_until(
+ 'postgres', q{select sum(idx_scan)=3 from pg_stat_all_indexes where indexrelname ilike 'users_table_part_%';}
+) or die "Timed out while waiting for check subscriber tap_sub_rep_full updates partitioned table";
+
+# make sure that the subscriber has the correct data
+$result = $node_subscriber->safe_psql('postgres',
+ "select sum(user_id+value_1+value_2) from users_table_part");
+is($result, qq(10907), 'ensure subscriber has the correct data at the end of the test');
+$result = $node_subscriber->safe_psql('postgres',
+ "select count(DISTINCT(user_id,value_1, value_2)) from users_table_part");
+is($result, qq(99), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE users_table_part");
+
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE users_table_part");
+
+# Testcase end: Subscription can use index on partitioned tables
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Subscription will not use indexes with only expressions or
+# partial index
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE people (firstname text, lastname text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE people REPLICA IDENTITY FULL");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE people (firstname text, lastname text)");
+
+# index with only an expression
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX people_names_expr_only ON people ((firstname || ' ' || lastname))");
+
+# partial index
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX people_names_partial ON people(firstname) WHERE (firstname = 'first_name_1')");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO people SELECT 'first_name_' || i::text, 'last_name_' || i::text FROM generate_series(0,200) i");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE people");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# update 2 rows
+$node_publisher->safe_psql('postgres',
+ "UPDATE people SET firstname = 'no-name' WHERE firstname = 'first_name_1'");
+$node_publisher->safe_psql('postgres',
+ "UPDATE people SET firstname = 'no-name' WHERE firstname = 'first_name_2' AND lastname = 'last_name_2'");
+
+# make sure none of the indexes is used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres',
+ "select sum(idx_scan) from pg_stat_all_indexes where indexrelname IN ('people_names_expr_only', 'people_names_partial')");
+is($result, qq(0), 'ensure subscriber tap_sub_rep_full updates two rows via seq. scan with index on expressions');
+
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM people WHERE firstname = 'first_name_3'");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM people WHERE firstname = 'first_name_4' AND lastname = 'last_name_4'");
+
+# make sure the index is not used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres',
+ "select sum(idx_scan) from pg_stat_all_indexes where indexrelname IN ('people_names_expr_only', 'people_names_partial')");
+is($result, qq(0), 'ensure subscriber tap_sub_rep_full updates two rows via seq. scan with index on expressions');
+
+# make sure that the subscriber has the correct data
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM people");
+is($result, qq(199), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE people");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE people");
+
+# Testcase end: Subscription will not use indexes with only expressions or
+# partial index
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Subscription can use index having expressions and columns
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE people (firstname text, lastname text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE people REPLICA IDENTITY FULL");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE people (firstname text, lastname text)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX people_names ON people (firstname, lastname, (firstname || ' ' || lastname))");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO people SELECT 'first_name_' || i::text, 'last_name_' || i::text FROM generate_series(0, 20) i");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE people");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# update 1 row
+$node_publisher->safe_psql('postgres',
+ "UPDATE people SET firstname = 'no-name' WHERE firstname = 'first_name_1'");
+
+# delete the updated row
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM people WHERE firstname = 'no-name'");
+
+# wait until the index is used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->poll_query_until(
+ 'postgres', q{select idx_scan=2 from pg_stat_all_indexes where indexrelname = 'people_names';}
+) or die "Timed out while waiting for check subscriber tap_sub_rep_full deletes two rows via index scan with index on expressions and columns";
+
+# make sure that the subscriber has the correct data
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM people");
+is($result, qq(20), 'ensure subscriber has the correct data at the end of the test');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM people WHERE firstname = 'no-name'");
+is($result, qq(0), 'ensure subscriber has the correct data at the end of the test');
+
+# now, drop the index with the expression, we'll use sequential scan
+$node_subscriber->safe_psql('postgres',
+ "DROP INDEX people_names");
+
+# delete 1 row
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM people WHERE lastname = 'last_name_18'");
+
+# make sure that the subscriber has the correct data
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM people WHERE lastname = 'last_name_18'");
+is($result, qq(0), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE people");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE people");
+
+# Testcase end: Subscription can use index having expressions and columns
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Null values and missing column
+
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int)"
+);
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE test_replica_id_full REPLICA IDENTITY FULL");
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int, y int)"
+);
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE INDEX test_replica_id_full_idx ON test_replica_id_full(x,y)"
+);
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE test_replica_id_full");
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# load some data, and update 2 tuples
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_replica_id_full VALUES (1), (2), (3)");
+$node_publisher->safe_psql('postgres',
+ "UPDATE test_replica_id_full SET x = x + 1 WHERE x = 1");
+
+# check if the index is used even when the index has NULL values
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->poll_query_until(
+ 'postgres', q{select idx_scan=1 from pg_stat_all_indexes where indexrelname = 'test_replica_id_full_idx';}
+) or die "Timed out while waiting for check subscriber tap_sub_rep_full updates test_replica_id_full table";
+
+# make sure that the subscriber has the correct data
+$result = $node_subscriber->safe_psql('postgres',
+ "select sum(x) from test_replica_id_full WHERE y IS NULL");
+is($result, qq(7), 'ensure subscriber has the correct data at the end of the test');
+
+# make sure that the subscriber has the correct data
+$result = $node_subscriber->safe_psql('postgres',
+ "select count(*) from test_replica_id_full WHERE y IS NULL");
+is($result, qq(3), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+
+# Testcase end: Null values And missing column
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Subscription using a unique index when Pub/Sub has different
+# data
+#
+# The subscriber has duplicate tuples that publisher does not have. When
+# publisher updates/deletes 1 row, subscriber uses indexes and updates/deletes
+# exactly 1 row.
+#
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int, y int)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE test_replica_id_full REPLICA IDENTITY FULL");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE test_replica_id_full (x int, y int)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE UNIQUE INDEX test_replica_id_full_idxy ON test_replica_id_full(x,y)");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_replica_id_full SELECT i, i FROM generate_series(0,21) i");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_rep_full FOR TABLE test_replica_id_full");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_rep_full CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_rep_full"
+);
+
+# wait for initial table synchronization to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# duplicate the data in subscriber for y column
+$node_subscriber->safe_psql('postgres',
+ "INSERT INTO test_replica_id_full SELECT i+100, i FROM generate_series(0,21) i");
+
+# now, we update only 1 row on the publisher and expect the subscriber to only
+# update 1 row although there are two tuples with y = 15 on the subscriber
+$node_publisher->safe_psql('postgres',
+ "UPDATE test_replica_id_full SET x = 2000 WHERE y = 15");
+
+# wait until the index is used on the subscriber
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->poll_query_until(
+ 'postgres', q{select (idx_scan = 1) from pg_stat_all_indexes where indexrelname = 'test_replica_id_full_idxy';}
+) or die "Timed out while waiting for check subscriber tap_sub_rep_full updates one row via index";
+
+# make sure that the subscriber has the correct data
+# we only updated 1 row
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM test_replica_id_full WHERE x = 2000");
+is($result, qq(1), 'ensure subscriber has the correct data at the end of the test');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_rep_full");
+$node_publisher->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_rep_full");
+$node_subscriber->safe_psql('postgres', "DROP TABLE test_replica_id_full");
+
+# Testcase start: Subscription using a unique index when Pub/Sub has different
+# data
+# =============================================================================
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
projects / users / rhaas / postgres.git / commitdiff