src/test/subscription/t/016_stream_subxact.pl

   1
   2 # Copyright (c) 2021-2022, PostgreSQL Global Development Group
   3
   4 # Test streaming of large transaction containing large subtransactions
   5 use strict;
   6 use warnings;
   7 use PostgreSQL::Test::Cluster;
   8 use PostgreSQL::Test::Utils;
   9 use Test::More tests => 2;
  10
  11 # Create publisher node
  12 my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
  13 $node_publisher->init(allows_streaming => 'logical');
  14 $node_publisher->append_conf('postgresql.conf',
  15         'logical_decoding_work_mem = 64kB');
  16 $node_publisher->start;
  17
  18 # Create subscriber node
  19 my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
  20 $node_subscriber->init(allows_streaming => 'logical');
  21 $node_subscriber->start;
  22
  23 # Create some preexisting content on publisher
  24 $node_publisher->safe_psql('postgres',
  25         "CREATE TABLE test_tab (a int primary key, b varchar)");
  26 $node_publisher->safe_psql('postgres',
  27         "INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar')");
  28
  29 # Setup structure on subscriber
  30 $node_subscriber->safe_psql('postgres',
  31         "CREATE TABLE test_tab (a int primary key, b text, c timestamptz DEFAULT now(), d bigint DEFAULT 999)"
  32 );
  33
  34 # Setup logical replication
  35 my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
  36 $node_publisher->safe_psql('postgres',
  37         "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
  38
  39 my $appname = 'tap_sub';
  40 $node_subscriber->safe_psql('postgres',
  41         "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
  42 );
  43
  44 $node_publisher->wait_for_catchup($appname);
  45
  46 # Also wait for initial table sync to finish
  47 my $synced_query =
  48   "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
  49 $node_subscriber->poll_query_until('postgres', $synced_query)
  50   or die "Timed out while waiting for subscriber to synchronize data";
  51
  52 my $result =
  53   $node_subscriber->safe_psql('postgres',
  54         "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 # Insert, update and delete enough rows to exceed 64kB limit.
  58 $node_publisher->safe_psql(
  59         'postgres', q{
  60 BEGIN;
  61 INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(    3,  500) s(i);
  62 UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
  63 DELETE FROM test_tab WHERE mod(a,3) = 0;
  64 SAVEPOINT s1;
  65 INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501,  1000) s(i);
  66 UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
  67 DELETE FROM test_tab WHERE mod(a,3) = 0;
  68 SAVEPOINT s2;
  69 INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001,  1500) s(i);
  70 UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
  71 DELETE FROM test_tab WHERE mod(a,3) = 0;
  72 SAVEPOINT s3;
  73 INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501,  2000) s(i);
  74 UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
  75 DELETE FROM test_tab WHERE mod(a,3) = 0;
  76 SAVEPOINT s4;
  77 INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001, 2500) s(i);
  78 UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
  79 DELETE FROM test_tab WHERE mod(a,3) = 0;
  80 COMMIT;
  81 });
  82
  83 $node_publisher->wait_for_catchup($appname);
  84
  85 $result =
  86   $node_subscriber->safe_psql('postgres',
  87         "SELECT count(*), count(c), count(d = 999) FROM test_tab");
  88 is($result, qq(1667|1667|1667),
  89         'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
  90 );
  91
  92 $node_subscriber->stop;
  93 $node_publisher->stop;