|
| 1 | +use strict; |
| 2 | +use warnings; |
| 3 | + |
| 4 | +use PostgresNode; |
| 5 | +use TestLib; |
| 6 | +use Test::Moretests=> 1; |
| 7 | + |
| 8 | +my$shard1 = get_new_node("shard1"); |
| 9 | +$shard1->init; |
| 10 | +$shard1->append_conf('postgresql.conf',qq( |
| 11 | +max_prepared_transactions = 30 |
| 12 | +postgres_fdw.use_tsdtm = on |
| 13 | +)); |
| 14 | +$shard1->start; |
| 15 | + |
| 16 | +my$shard2 = get_new_node("shard2"); |
| 17 | +$shard2->init; |
| 18 | +$shard2->append_conf('postgresql.conf',qq( |
| 19 | +max_prepared_transactions = 30 |
| 20 | +postgres_fdw.use_tsdtm = on |
| 21 | +)); |
| 22 | +$shard2->start; |
| 23 | + |
| 24 | +############################################################################### |
| 25 | +# Prepare nodes |
| 26 | +############################################################################### |
| 27 | + |
| 28 | +my@shards = ($shard1,$shard2); |
| 29 | + |
| 30 | +foreachmy$node (@shards) |
| 31 | +{ |
| 32 | +$node->safe_psql('postgres',qq[ |
| 33 | +CREATE EXTENSION postgres_fdw; |
| 34 | +CREATE TABLE accounts(id integer primary key, amount integer); |
| 35 | +]); |
| 36 | + |
| 37 | +foreachmy$neighbor (@shards) |
| 38 | +{ |
| 39 | +nextif ($neighboreq$node); |
| 40 | + |
| 41 | +my$port =$neighbor->port; |
| 42 | +my$host =$neighbor->host; |
| 43 | + |
| 44 | +$node->safe_psql('postgres',qq[ |
| 45 | +CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw |
| 46 | +options(dbname 'postgres', host '$host', port '$port'); |
| 47 | +CREATE FOREIGN TABLE |
| 48 | +accounts_fdw(id integer, amount integer) |
| 49 | +server shard_$port options(table_name 'accounts'); |
| 50 | +CREATE USER MAPPING for stas SERVER shard_$port |
| 51 | +options (user 'stas'); |
| 52 | +]); |
| 53 | +} |
| 54 | + |
| 55 | +} |
| 56 | + |
| 57 | +$shard1->psql('postgres',"insert into accounts select id, 0 from generate_series(1, 20020) as id;"); |
| 58 | +$shard2->psql('postgres',"insert into accounts select id, 0 from generate_series(1, 20020) as id;"); |
| 59 | + |
| 60 | + |
| 61 | +# diag("shard1: @{[$shard1->connstr('postgres')]}"); |
| 62 | +# diag("shard1: @{[$shard2->connstr('postgres')]}"); |
| 63 | +# sleep(3600); |
| 64 | + |
| 65 | + |
| 66 | +############################################################################### |
| 67 | +# pgbench scripts |
| 68 | +############################################################################### |
| 69 | + |
| 70 | +my$bank = File::Temp->new(); |
| 71 | +append_to_file($bank,q{ |
| 72 | +\set id random(1, 20000) |
| 73 | +BEGIN; |
| 74 | +UPDATE accounts SET amount = amount - 1 WHERE id = :id; |
| 75 | +UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1); |
| 76 | +UPDATE accounts_fdw SET amount = amount - 1 WHERE id = :id; |
| 77 | +UPDATE accounts_fdw SET amount = amount + 1 WHERE id = (:id + 1); |
| 78 | +COMMIT; |
| 79 | +}); |
| 80 | + |
| 81 | +############################################################################### |
| 82 | +# Concurrent global transactions |
| 83 | +############################################################################### |
| 84 | + |
| 85 | +my ($hashes,$hash1,$hash2); |
| 86 | +my$stability_errors = 0; |
| 87 | +my$selects = 0; |
| 88 | +my$seconds = 30; |
| 89 | + |
| 90 | +my$pgb_handle1 =$shard1->pgbench_async(-n,-c=> 5,-T=>$seconds,-f=>$bank,'postgres' ); |
| 91 | +# my $pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' ); |
| 92 | +my$started =time(); |
| 93 | +while (time() -$started <$seconds) |
| 94 | +{ |
| 95 | +foreachmy$node ($shard1,$shard2) |
| 96 | +{ |
| 97 | +($hash1,$_,$hash2) =split"\n",$node->safe_psql('postgres',qq[ |
| 98 | +begin isolation level repeatable read; |
| 99 | +select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t; |
| 100 | +select pg_sleep(1); |
| 101 | +select md5(array_agg((t.*)::text)::text) from (select * from accounts_fdw order by id) as t; |
| 102 | +commit; |
| 103 | +]); |
| 104 | + |
| 105 | +if ($hash1ne$hash2) |
| 106 | +{ |
| 107 | +diag("oops"); |
| 108 | +$stability_errors++; |
| 109 | +} |
| 110 | +elsif ($hash1eq''or$hash2eq'') |
| 111 | +{ |
| 112 | +die; |
| 113 | +} |
| 114 | +else |
| 115 | +{ |
| 116 | +diag("got hash$hash1"); |
| 117 | +$selects++; |
| 118 | +} |
| 119 | +} |
| 120 | +} |
| 121 | + |
| 122 | +$shard1->pgbench_await($pgb_handle1); |
| 123 | +# $shard2->pgbench_await($pgb_handle2); |
| 124 | + |
| 125 | +die"no real queries happend"unless ($selects > 0 ); |
| 126 | + |
| 127 | +is($stability_errors, 0,'snapshot is stable during concurrent global and local transactions'); |
| 128 | + |
| 129 | +$shard1->stop; |
| 130 | +$shard2->stop; |