|
| 1 | +# Test behavior with different schema on subscriber |
| 2 | +use strict; |
| 3 | +use warnings; |
| 4 | +use PostgresNode; |
| 5 | +use TestLib; |
| 6 | +use Test::Moretests=> 3; |
| 7 | + |
| 8 | +subwait_for_caught_up |
| 9 | +{ |
| 10 | +my ($node,$appname) =@_; |
| 11 | + |
| 12 | +$node->poll_query_until('postgres', |
| 13 | +"SELECT pg_current_wal_lsn() <= replay_lsn FROM pg_stat_replication WHERE application_name = '$appname';" |
| 14 | +)ordie"Timed out while waiting for subscriber to catch up"; |
| 15 | +} |
| 16 | + |
| 17 | +# Create publisher node |
| 18 | +my$node_publisher = get_new_node('publisher'); |
| 19 | +$node_publisher->init(allows_streaming=>'logical'); |
| 20 | +$node_publisher->start; |
| 21 | + |
| 22 | +# Create subscriber node |
| 23 | +my$node_subscriber = get_new_node('subscriber'); |
| 24 | +$node_subscriber->init(allows_streaming=>'logical'); |
| 25 | +$node_subscriber->start; |
| 26 | + |
| 27 | +# Create some preexisting content on publisher |
| 28 | +$node_publisher->safe_psql('postgres', |
| 29 | +"CREATE TABLE test_tab (a int primary key, b varchar)"); |
| 30 | +$node_publisher->safe_psql('postgres', |
| 31 | +"INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar')"); |
| 32 | + |
| 33 | +# Setup structure on subscriber |
| 34 | +$node_subscriber->safe_psql('postgres',"CREATE TABLE test_tab (a int primary key, b text, c timestamptz DEFAULT now(), d bigint DEFAULT 999)"); |
| 35 | + |
| 36 | +# Setup logical replication |
| 37 | +my$publisher_connstr =$node_publisher->connstr .' dbname=postgres'; |
| 38 | +$node_publisher->safe_psql('postgres',"CREATE PUBLICATION tap_pub FOR TABLE test_tab"); |
| 39 | + |
| 40 | +my$appname ='tap_sub'; |
| 41 | +$node_subscriber->safe_psql('postgres', |
| 42 | +"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub" |
| 43 | +); |
| 44 | + |
| 45 | +wait_for_caught_up($node_publisher,$appname); |
| 46 | + |
| 47 | +# Also wait for initial table sync to finish |
| 48 | +my$synced_query = |
| 49 | +"SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');"; |
| 50 | +$node_subscriber->poll_query_until('postgres',$synced_query) |
| 51 | +ordie"Timed out while waiting for subscriber to synchronize data"; |
| 52 | + |
| 53 | +my$result = |
| 54 | +$node_subscriber->safe_psql('postgres',"SELECT count(*), count(c), count(d = 999) FROM test_tab"); |
| 55 | +is($result,qq(2|2|2),'check initial data was copied to subscriber'); |
| 56 | + |
| 57 | +# Update the rows on the publisher and check the additional columns on |
| 58 | +# subscriber didn't change |
| 59 | +$node_publisher->safe_psql('postgres',"UPDATE test_tab SET b = md5(b)"); |
| 60 | + |
| 61 | +wait_for_caught_up($node_publisher,$appname); |
| 62 | + |
| 63 | +$result = |
| 64 | +$node_subscriber->safe_psql('postgres',"SELECT count(*), count(c), count(d = 999) FROM test_tab"); |
| 65 | +is($result,qq(2|2|2),'check extra columns contain local defaults'); |
| 66 | + |
| 67 | +# Change the local values of the extra columns on the subscriber, |
| 68 | +# update publisher, and check that subscriber retains the expected |
| 69 | +# values |
| 70 | +$node_subscriber->safe_psql('postgres',"UPDATE test_tab SET c = 'epoch'::timestamptz + 987654321 * interval '1s'"); |
| 71 | +$node_publisher->safe_psql('postgres',"UPDATE test_tab SET b = md5(a::text)"); |
| 72 | + |
| 73 | +wait_for_caught_up($node_publisher,$appname); |
| 74 | + |
| 75 | +$result = |
| 76 | +$node_subscriber->safe_psql('postgres',"SELECT count(*), count(extract(epoch from c) = 987654321), count(d = 999) FROM test_tab"); |
| 77 | +is($result,qq(2|2|2),'check extra columns contain locally changed data'); |
| 78 | + |
| 79 | +$node_subscriber->stop; |
| 80 | +$node_publisher->stop; |