Add reference table failure tests

Fairly straightforward; verified that modifications fail atomically if
a worker is down or fails mid-transaction (i.e. all workers need to ack
modifications to reference tables in order to persist changes).

src/test/regress/expected/failure_ref_tables.out

@@ -0,0 +1,99 @@
+SET citus.next_shard_id TO 100500;
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+CREATE TABLE ref_table (key int, value int);
+SELECT create_reference_table('ref_table');
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+\copy ref_table FROM stdin delimiter ',';
+SELECT citus.clear_network_traffic();
+ clear_network_traffic 
+-----------------------
+ 
+(1 row)
+
+SELECT COUNT(*) FROM ref_table;
+ count 
+-------
+     4
+(1 row)
+
+-- verify behavior of single INSERT; should fail to execute
+SELECT citus.mitmproxy('conn.onQuery(query="^INSERT").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+INSERT INTO ref_table VALUES (5, 6);
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+SELECT COUNT(*) FROM ref_table WHERE key=5;
+ count 
+-------
+     0
+(1 row)
+
+-- verify behavior of UPDATE ... RETURNING; should not execute
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+UPDATE ref_table SET key=7 RETURNING value;
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+SELECT COUNT(*) FROM ref_table WHERE key=7;
+ count 
+-------
+     0
+(1 row)
+
+-- verify fix to #2214; should raise error and fail to execute
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM ref_table WHERE key=5;
+UPDATE ref_table SET key=value;
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+COMMIT;
+SELECT COUNT(*) FROM ref_table WHERE key=value;
+ count 
+-------
+     0
+(1 row)
+
+-- all shards should still be healthy
+SELECT COUNT(*) FROM pg_dist_shard_placement WHERE shardstate = 3;
+ count 
+-------
+     0
+(1 row)
+
+-- ==== Clean up, we're done here ====
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+DROP TABLE ref_table;

src/test/regress/failure_schedule

@@ -23,3 +23,4 @@ test: failure_insert_select_via_coordinator
 test: failure_multi_dml
 test: failure_vacuum
 test: failure_single_select
+test: failure_ref_tables

src/test/regress/sql/failure_ref_tables.sql

@@ -0,0 +1,47 @@
+SET citus.next_shard_id TO 100500;
+
+SELECT citus.mitmproxy('conn.allow()');
+
+CREATE TABLE ref_table (key int, value int);
+SELECT create_reference_table('ref_table');
+
+\copy ref_table FROM stdin delimiter ',';
+1,2
+2,3
+3,4
+4,5
+\.
+
+SELECT citus.clear_network_traffic();
+
+SELECT COUNT(*) FROM ref_table;
+
+-- verify behavior of single INSERT; should fail to execute
+SELECT citus.mitmproxy('conn.onQuery(query="^INSERT").kill()');
+INSERT INTO ref_table VALUES (5, 6);
+
+SELECT COUNT(*) FROM ref_table WHERE key=5;
+
+-- verify behavior of UPDATE ... RETURNING; should not execute
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+UPDATE ref_table SET key=7 RETURNING value;
+
+SELECT COUNT(*) FROM ref_table WHERE key=7;
+
+-- verify fix to #2214; should raise error and fail to execute
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+
+BEGIN;
+DELETE FROM ref_table WHERE key=5;
+UPDATE ref_table SET key=value;
+COMMIT;
+
+SELECT COUNT(*) FROM ref_table WHERE key=value;
+
+-- all shards should still be healthy
+SELECT COUNT(*) FROM pg_dist_shard_placement WHERE shardstate = 3;
+
+-- ==== Clean up, we're done here ====
+
+SELECT citus.mitmproxy('conn.allow()');
+DROP TABLE ref_table;