|
| 1 | +use strict; |
| 2 | +use warnings; |
| 3 | + |
| 4 | +use PostgresNode; |
| 5 | +use TestLib; |
| 6 | +use Test::More tests => 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 | +foreach my $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 | + foreach my $neighbor (@shards) |
| 38 | + { |
| 39 | + next if ($neighbor eq $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 | + foreach my $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 ($hash1 ne $hash2) |
| 106 | + { |
| 107 | + diag("oops"); |
| 108 | + $stability_errors++; |
| 109 | + } |
| 110 | + elsif ($hash1 eq '' or $hash2 eq '') |
| 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; |
0 commit comments