Merge pull request #939 from citusdata/bugfix/colocated_multi_shards

Use co-located shard ID in multi-shard transactions
2016-11-02 11:18:46 +01:00 · 2016-11-02 11:18:46 +01:00 · d19c4869d6
parent 74d8e4f640 f6b3af7a49
commit d19c4869d6
9 changed files with 356 additions and 113 deletions
--- a/src/backend/distributed/transaction/multi_shard_transaction.c
+++ b/src/backend/distributed/transaction/multi_shard_transaction.c
@ -13,10 +13,13 @@
 #include "libpq-fe.h"
 #include "postgres.h"

+#include "distributed/colocation_utils.h"
 #include "distributed/commit_protocol.h"
 #include "distributed/connection_cache.h"
 #include "distributed/master_metadata_utility.h"
+#include "distributed/metadata_cache.h"
 #include "distributed/multi_shard_transaction.h"
+#include "distributed/shardinterval_utils.h"
 #include "distributed/worker_manager.h"
 #include "nodes/pg_list.h"
 #include "storage/ipc.h"
@ -182,7 +185,17 @@ BeginTransactionOnShardPlacements(uint64 shardId, char *userName)
 ShardConnections *
 GetShardConnections(int64 shardId, bool *shardConnectionsFound)
 {
-	return GetShardHashConnections(shardConnectionHash, shardId, shardConnectionsFound);
+	ShardConnections *shardConnections = NULL;
+
+	ShardInterval *shardInterval = LoadShardInterval(shardId);
+	List *colocatedShardIds = ColocatedShardIntervalList(shardInterval);
+	ShardInterval *baseShardInterval = LowestShardIntervalById(colocatedShardIds);
+	int64 baseShardId = baseShardInterval->shardId;
+
+	shardConnections = GetShardHashConnections(shardConnectionHash, baseShardId,
+											   shardConnectionsFound);
+
+	return shardConnections;
 }


--- a/src/backend/distributed/utils/shardinterval_utils.c
+++ b/src/backend/distributed/utils/shardinterval_utils.c
@ -28,6 +28,31 @@ static ShardInterval * SearchCachedShardInterval(Datum partitionColumnValue,
 												 FmgrInfo *compareFunction);


+/*
+ * LowestShardIntervalById returns the shard interval with the lowest shard
+ * ID from a list of shard intervals.
+ */
+ShardInterval *
+LowestShardIntervalById(List *shardIntervalList)
+{
+	ShardInterval *lowestShardInterval = NULL;
+	ListCell *shardIntervalCell = NULL;
+
+	foreach(shardIntervalCell, shardIntervalList)
+	{
+		ShardInterval *shardInterval = (ShardInterval *) lfirst(shardIntervalCell);
+
+		if (lowestShardInterval == NULL ||
+			lowestShardInterval->shardId > shardInterval->shardId)
+		{
+			lowestShardInterval = shardInterval;
+		}
+	}
+
+	return lowestShardInterval;
+}
+
+
 /*
 * CompareShardIntervals acts as a helper function to compare two shard intervals
 * by their minimum values, using the value's type comparison function.
--- a/src/include/distributed/shardinterval_utils.h
+++ b/src/include/distributed/shardinterval_utils.h
@ -23,6 +23,7 @@ typedef struct ShardIntervalCompareFunctionCacheEntry
 	FmgrInfo *functionInfo;
 } ShardIntervalCompareFunctionCacheEntry;

+extern ShardInterval * LowestShardIntervalById(List *shardIntervalList);
 extern int CompareShardIntervals(const void *leftElement, const void *rightElement,
 								 FmgrInfo *typeCompareFunction);
 extern int CompareShardIntervalsById(const void *leftElement, const void *rightElement);
--- a/src/test/regress/expected/multi_foreign_key.out
+++ b/src/test/regress/expected/multi_foreign_key.out
@ -101,6 +101,16 @@ CONTEXT:  while executing command on localhost:57637
 -- test delete from referenced table while there is NO corresponding value in referencing table
 DELETE FROM referencing_table WHERE ref_id = 1;
 DELETE FROM referenced_table WHERE id = 1;
+-- test cascading truncate
+INSERT INTO referenced_table VALUES(2, 2);
+INSERT INTO referencing_table VALUES(2, 2);
+TRUNCATE referenced_table CASCADE;
+NOTICE:  truncate cascades to table "referencing_table"
+SELECT * FROM referencing_table;
+ id | ref_id 
+----+--------
+(0 rows)
+
 -- drop table for next tests
 DROP TABLE referencing_table;
 DROP TABLE referenced_table;
@ -120,6 +130,7 @@ SELECT create_distributed_table('referencing_table', 'ref_id', 'hash');
 
 (1 row)

+-- single shard cascading delete
 INSERT INTO referenced_table VALUES(1, 1);
 INSERT INTO referencing_table VALUES(1, 1);
 DELETE FROM referenced_table WHERE id = 1;
@ -133,6 +144,34 @@ SELECT * FROM referenced_table;
 ----+-------------
 (0 rows)

+-- multi shard cascading delete
+INSERT INTO referenced_table VALUES(2, 2);
+INSERT INTO referencing_table VALUES(2, 2);
+SELECT master_modify_multiple_shards('DELETE FROM referenced_table');
+ master_modify_multiple_shards 
+-------------------------------
+                             1
+(1 row)
+
+SELECT * FROM referencing_table;
+ id | ref_id 
+----+--------
+(0 rows)
+
+-- multi shard cascading delete with alter table
+INSERT INTO referenced_table VALUES(3, 3);
+INSERT INTO referencing_table VALUES(3, 3);
+BEGIN;
+ALTER TABLE referencing_table ADD COLUMN x int DEFAULT 0;
+NOTICE:  using one-phase commit for distributed DDL commands
+HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
+SELECT master_modify_multiple_shards('DELETE FROM referenced_table');
+ master_modify_multiple_shards 
+-------------------------------
+                             1
+(1 row)
+
+COMMIT;
 DROP TABLE referencing_table;
 DROP TABLE referenced_table;
 -- test ON DELETE NO ACTION + DEFERABLE + INITIALLY DEFERRED
--- a/src/test/regress/expected/multi_insert_select.out
+++ b/src/test/regress/expected/multi_insert_select.out
@ -80,14 +80,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300004
-DEBUG:  sent COMMIT over connection 13300004
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- see that our first multi shard INSERT...SELECT works expected
 SET client_min_messages TO INFO;
 DEBUG:  StartTransactionCommand
@ -283,14 +283,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300004
-DEBUG:  sent COMMIT over connection 13300004
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 user_id | time | value_1 | value_2 | value_3 | value_4 
 ---------+------+---------+---------+---------+---------
       9 |      |      90 |         |    9000 |        
@ -363,14 +363,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- group by column not exists on the SELECT target list
 INSERT INTO agg_events (value_3_agg, value_4_agg, value_1_agg, user_id) 
 SELECT
@ -574,14 +574,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- add one more level subqueris on top of subquery JOINs
 INSERT INTO agg_events
            (user_id, value_4_agg)
@ -659,14 +659,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- subqueries in WHERE clause
 INSERT INTO raw_events_second
            (user_id)
@ -711,14 +711,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300007
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300005
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300006
-DEBUG:  sent COMMIT over connection 13300004
-DEBUG:  sent COMMIT over connection 13300004
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- some UPSERTS
 INSERT INTO agg_events AS ae 
            (
@ -758,14 +758,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- upserts with returning
 INSERT INTO agg_events AS ae 
            ( 
@ -806,14 +806,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 user_id | value_1_agg 
 ---------+-------------
       7 |            
@ -848,14 +848,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 --  FILTER CLAUSE
 INSERT INTO agg_events (user_id, value_1_agg)
 SELECT
@ -886,14 +886,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- a test with reference table JOINs
 INSERT INTO
  agg_events (user_id, value_1_agg)
@ -929,14 +929,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- a note on the outer joins is that
 -- we filter out outer join results
 -- where partition column returns
@ -1024,14 +1024,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 SET client_min_messages TO INFO;
 DEBUG:  StartTransactionCommand
 DEBUG:  StartTransaction
@ -1094,14 +1094,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- we don't want to see constraint vialotions, so truncate first
 SET client_min_messages TO INFO;
 DEBUG:  StartTransactionCommand
@ -1168,14 +1168,14 @@ DEBUG:  Plan is router executable
 DEBUG:  CommitTransactionCommand
 DEBUG:  CommitTransaction
 DEBUG:  name: unnamed; blockState:       STARTED; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300008
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300011
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300009
-DEBUG:  sent COMMIT over connection 13300010
-DEBUG:  sent COMMIT over connection 13300010
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300001
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300000
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300002
+DEBUG:  sent COMMIT over connection 13300003
+DEBUG:  sent COMMIT over connection 13300003
 -- We do not support any set operations
 INSERT INTO
  raw_events_first(user_id)
@ -1455,6 +1455,62 @@ DEBUG:  StartTransactionCommand
 DEBUG:  StartTransaction
 DEBUG:  name: unnamed; blockState:       DEFAULT; state: INPROGR, xid/subid/cid: 0/1/0, nestlvl: 1, children: 
 DEBUG:  ProcessUtility
+-- avoid constraint violations
+TRUNCATE raw_events_first;
+-- Altering a table and selecting from it using a multi-shard statement
+-- in the same transaction is allowed because we will use the same
+-- connections for all co-located placements.
+BEGIN;
+ALTER TABLE raw_events_second DROP COLUMN value_4;
+NOTICE:  using one-phase commit for distributed DDL commands
+HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
+INSERT INTO raw_events_first SELECT * FROM raw_events_second; 
+ROLLBACK;
+-- Alterating a table and selecting from it using a single-shard statement
+-- in the same transaction is disallowed because we will use a different
+-- connection.
+BEGIN;
+ALTER TABLE raw_events_second DROP COLUMN value_4;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 100; 
+ERROR:  single-shard DML commands must not appear in transaction blocks which contain multi-shard data modifications
+ROLLBACK;
+-- Insert after copy is currently disallowed because of the way the 
+-- transaction modification state is currently handled. Copy still
+-- goes through despite rollback.
+BEGIN;
+COPY raw_events_second (user_id, value_1) FROM STDIN DELIMITER ',';
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+ERROR:  multi-shard data modifications must not appear in transaction blocks which contain single-shard DML commands
+ROLLBACK;
+-- Insert after copy is currently allowed for single-shard operation.
+-- Since the COPY commits immediately, the result is visible in the
+-- next operation. Copy goes through despite rollback, while insert
+-- rolls back.
+BEGIN;
+COPY raw_events_second (user_id, value_1) FROM STDIN DELIMITER ',';
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 101;
+SELECT user_id FROM raw_events_first WHERE user_id = 101;
+ user_id 
+---------
+     101
+(1 row)
+
+ROLLBACK;
+-- Copy after insert is disallowed since the insert is not immediately
+-- committed and the copy uses different connections that will not yet
+-- see the result of the insert.
+BEGIN;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+COPY raw_events_first (user_id, value_1) FROM STDIN DELIMITER ',';
+ERROR:  distributed copy operations must not appear in transaction blocks containing other distributed modifications
+CONTEXT:  COPY raw_events_first, line 1: "102,102"
+ROLLBACK;
+BEGIN;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 100;
+COPY raw_events_first (user_id, value_1) FROM STDIN DELIMITER ',';
+ERROR:  distributed copy operations must not appear in transaction blocks containing other distributed modifications
+CONTEXT:  COPY raw_events_first, line 1: "103,103"
+ROLLBACK;
 -- Views does not work
 CREATE VIEW test_view AS SELECT * FROM raw_events_first;
 INSERT INTO raw_events_second SELECT * FROM test_view;
--- a/src/test/regress/expected/multi_modifying_xacts.out
+++ b/src/test/regress/expected/multi_modifying_xacts.out
@ -260,6 +260,24 @@ SELECT * FROM labs WHERE id = 12;
 12 | fsociety
 (1 row)

+-- multi-shard operations can co-exist with DDL in a transactional way
+BEGIN;
+ALTER TABLE labs ADD COLUMN motto text;
+SELECT master_modify_multiple_shards('DELETE FROM labs');
+ master_modify_multiple_shards 
+-------------------------------
+                             5
+(1 row)
+
+ALTER TABLE labs ADD COLUMN score float;
+ROLLBACK;
+-- should have rolled everything back
+SELECT * FROM labs WHERE id = 12;
+ id |   name   
+----+----------
+ 12 | fsociety
+(1 row)
+
 -- now, for some special failures...
 CREATE TABLE objects (
 	id bigint PRIMARY KEY,
--- a/src/test/regress/sql/multi_foreign_key.sql
+++ b/src/test/regress/sql/multi_foreign_key.sql
@ -81,6 +81,12 @@ DELETE FROM referenced_table WHERE id = 1;
 DELETE FROM referencing_table WHERE ref_id = 1;
 DELETE FROM referenced_table WHERE id = 1;

+-- test cascading truncate
+INSERT INTO referenced_table VALUES(2, 2);
+INSERT INTO referencing_table VALUES(2, 2);
+TRUNCATE referenced_table CASCADE;
+SELECT * FROM referencing_table;
+
 -- drop table for next tests
 DROP TABLE referencing_table;
 DROP TABLE referenced_table;
@ -91,11 +97,28 @@ CREATE TABLE referenced_table(id int UNIQUE, test_column int, PRIMARY KEY(id, te
 CREATE TABLE referencing_table(id int, ref_id int, FOREIGN KEY(ref_id) REFERENCES referenced_table(id) ON DELETE CASCADE);
 SELECT create_distributed_table('referenced_table', 'id', 'hash');
 SELECT create_distributed_table('referencing_table', 'ref_id', 'hash');
+
+-- single shard cascading delete
 INSERT INTO referenced_table VALUES(1, 1);
 INSERT INTO referencing_table VALUES(1, 1);
 DELETE FROM referenced_table WHERE id = 1;
 SELECT * FROM referencing_table;
 SELECT * FROM referenced_table;
+
+-- multi shard cascading delete
+INSERT INTO referenced_table VALUES(2, 2);
+INSERT INTO referencing_table VALUES(2, 2);
+SELECT master_modify_multiple_shards('DELETE FROM referenced_table');
+SELECT * FROM referencing_table;
+
+-- multi shard cascading delete with alter table
+INSERT INTO referenced_table VALUES(3, 3);
+INSERT INTO referencing_table VALUES(3, 3);
+BEGIN;
+ALTER TABLE referencing_table ADD COLUMN x int DEFAULT 0;
+SELECT master_modify_multiple_shards('DELETE FROM referenced_table');
+COMMIT;
+
 DROP TABLE referencing_table;
 DROP TABLE referenced_table;

--- a/src/test/regress/sql/multi_insert_select.sql
+++ b/src/test/regress/sql/multi_insert_select.sql
@ -603,6 +603,64 @@ GROUP  BY grouping sets ( ( user_id ), ( value_1 ), ( user_id, value_1 ), ( ) );
 -- set back to INFO
 SET client_min_messages TO INFO;

+-- avoid constraint violations
+TRUNCATE raw_events_first;
+
+-- Altering a table and selecting from it using a multi-shard statement
+-- in the same transaction is allowed because we will use the same
+-- connections for all co-located placements.
+BEGIN;
+ALTER TABLE raw_events_second DROP COLUMN value_4;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second; 
+ROLLBACK;
+
+-- Alterating a table and selecting from it using a single-shard statement
+-- in the same transaction is disallowed because we will use a different
+-- connection.
+BEGIN;
+ALTER TABLE raw_events_second DROP COLUMN value_4;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 100; 
+ROLLBACK;
+
+-- Insert after copy is currently disallowed because of the way the 
+-- transaction modification state is currently handled. Copy still
+-- goes through despite rollback.
+BEGIN;
+COPY raw_events_second (user_id, value_1) FROM STDIN DELIMITER ',';
+100,100
+\.
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+ROLLBACK;
+
+-- Insert after copy is currently allowed for single-shard operation.
+-- Since the COPY commits immediately, the result is visible in the
+-- next operation. Copy goes through despite rollback, while insert
+-- rolls back.
+BEGIN;
+COPY raw_events_second (user_id, value_1) FROM STDIN DELIMITER ',';
+101,101
+\.
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 101;
+SELECT user_id FROM raw_events_first WHERE user_id = 101;
+ROLLBACK;
+
+-- Copy after insert is disallowed since the insert is not immediately
+-- committed and the copy uses different connections that will not yet
+-- see the result of the insert.
+BEGIN;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second;
+COPY raw_events_first (user_id, value_1) FROM STDIN DELIMITER ',';
+102,102
+\.
+ROLLBACK;
+
+BEGIN;
+INSERT INTO raw_events_first SELECT * FROM raw_events_second WHERE user_id = 100;
+COPY raw_events_first (user_id, value_1) FROM STDIN DELIMITER ',';
+103,103
+\.
+ROLLBACK;
+
 -- Views does not work
 CREATE VIEW test_view AS SELECT * FROM raw_events_first;
 INSERT INTO raw_events_second SELECT * FROM test_view;
--- a/src/test/regress/sql/multi_modifying_xacts.sql
+++ b/src/test/regress/sql/multi_modifying_xacts.sql
@ -202,6 +202,16 @@ COMMIT;
 \d labs
 SELECT * FROM labs WHERE id = 12;

+-- multi-shard operations can co-exist with DDL in a transactional way
+BEGIN;
+ALTER TABLE labs ADD COLUMN motto text;
+SELECT master_modify_multiple_shards('DELETE FROM labs');
+ALTER TABLE labs ADD COLUMN score float;
+ROLLBACK;
+
+-- should have rolled everything back
+SELECT * FROM labs WHERE id = 12;
+
 -- now, for some special failures...
 CREATE TABLE objects (
 	id bigint PRIMARY KEY,