Ensure all preceding writes are visible in data migration

src/backend/distributed/commands/create_distributed_table.c

@@ -762,6 +762,17 @@ CopyLocalDataIntoShards(Oid distributedRelationId)
 	/* take an ExclusiveLock to block all operations except SELECT */
 	distributedRelation = heap_open(distributedRelationId, ExclusiveLock);
 
+	/*
+	 * All writes have finished, make sure that we can see them by using the
+	 * latest snapshot. We use GetLatestSnapshot instead of
+	 * GetTransactionSnapshot since the latter would not reveal all writes
+	 * in serializable or repeatable read mode. Note that subsequent reads
+	 * from the distributed table would reveal those writes, temporarily
+	 * violating the isolation level. However, this seems preferable over
+	 * dropping the writes entirely.
+	 */
+	PushActiveSnapshot(GetLatestSnapshot());
+
 	/* get the table columns */
 	tupleDescriptor = RelationGetDescr(distributedRelation);
 	slot = MakeSingleTupleTableSlot(tupleDescriptor);
@@ -827,6 +838,8 @@ CopyLocalDataIntoShards(Oid distributedRelationId)
 	ExecDropSingleTupleTableSlot(slot);
 	FreeExecutorState(estate);
 	heap_close(distributedRelation, NoLock);
+
+	PopActiveSnapshot();
 }
 
 

src/test/regress/expected/isolation_data_migration.out

@@ -0,0 +1,151 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s2-begin s2-copy s1-create_distributed_table s2-commit s2-select
+step s2-begin: 
+	BEGIN;
+
+step s2-copy: 
+	COPY migration_table FROM PROGRAM 'echo 1,hello' WITH CSV;
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+ <waiting ...>
+step s2-commit: 
+	COMMIT;
+
+step s1-create_distributed_table: <... completed>
+create_distributed_table
+
+               
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          
+
+starting permutation: s1-begin s1-create_distributed_table s2-copy s1-commit s2-select
+step s1-begin: 
+	BEGIN;
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+
+create_distributed_table
+
+               
+step s2-copy: 
+	COPY migration_table FROM PROGRAM 'echo 1,hello' WITH CSV;
+ <waiting ...>
+step s1-commit: 
+	COMMIT;
+
+step s2-copy: <... completed>
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          
+
+starting permutation: s2-begin s2-insert s1-create_distributed_table s2-commit s2-select
+step s2-begin: 
+	BEGIN;
+
+step s2-insert: 
+	INSERT INTO migration_table VALUES (1, 'hello');
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+ <waiting ...>
+step s2-commit: 
+	COMMIT;
+
+step s1-create_distributed_table: <... completed>
+create_distributed_table
+
+               
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          
+
+starting permutation: s1-begin s1-create_distributed_table s2-insert s1-commit s2-select
+step s1-begin: 
+	BEGIN;
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+
+create_distributed_table
+
+               
+step s2-insert: 
+	INSERT INTO migration_table VALUES (1, 'hello');
+ <waiting ...>
+step s1-commit: 
+	COMMIT;
+
+step s2-insert: <... completed>
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          
+
+starting permutation: s1-begin-serializable s2-copy s1-create_distributed_table s1-commit s2-select
+step s1-begin-serializable: 
+	BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+	SELECT 1;
+
+?column?       
+
+1              
+step s2-copy: 
+	COPY migration_table FROM PROGRAM 'echo 1,hello' WITH CSV;
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+
+create_distributed_table
+
+               
+step s1-commit: 
+	COMMIT;
+
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          
+
+starting permutation: s1-begin-serializable s2-insert s1-create_distributed_table s1-commit s2-select
+step s1-begin-serializable: 
+	BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+	SELECT 1;
+
+?column?       
+
+1              
+step s2-insert: 
+	INSERT INTO migration_table VALUES (1, 'hello');
+
+step s1-create_distributed_table: 
+	SELECT create_distributed_table('migration_table', 'test_id');
+
+create_distributed_table
+
+               
+step s1-commit: 
+	COMMIT;
+
+step s2-select: 
+	SELECT * FROM migration_table ORDER BY test_id;
+
+test_id        data           
+
+1              hello          

src/test/regress/isolation_schedule

@@ -1,4 +1,4 @@
 test: isolation_cluster_management
 test: isolation_dml_vs_repair
-test: isolation_concurrent_dml
+test: isolation_concurrent_dml isolation_data_migration
 test: isolation_drop_shards

src/test/regress/specs/isolation_data_migration.spec

@@ -0,0 +1,73 @@
+setup
+{
+	CREATE TABLE migration_table (test_id integer NOT NULL, data text);
+}
+
+teardown
+{
+	DROP TABLE migration_table;
+}
+
+session "s1"
+
+step "s1-begin"
+{
+	BEGIN;
+}
+
+step "s1-begin-serializable"
+{
+	BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+	SELECT 1;
+}
+
+step "s1-create_distributed_table"
+{
+	SELECT create_distributed_table('migration_table', 'test_id');
+}
+
+step "s1-commit"
+{
+	COMMIT;
+}
+
+session "s2"
+
+step "s2-begin"
+{
+	BEGIN;
+}
+
+step "s2-copy"
+{
+	COPY migration_table FROM PROGRAM 'echo 1,hello' WITH CSV;
+}
+
+step "s2-insert"
+{
+	INSERT INTO migration_table VALUES (1, 'hello');
+}
+
+step "s2-commit"
+{
+	COMMIT;
+}
+
+step "s2-select"
+{
+	SELECT * FROM migration_table ORDER BY test_id;
+}
+
+# verify that local COPY is picked up by create_distributed_table once it commits
+permutation "s2-begin" "s2-copy" "s1-create_distributed_table" "s2-commit" "s2-select"
+# verify that COPY is distributed once create_distributed_table commits
+permutation "s1-begin" "s1-create_distributed_table" "s2-copy" "s1-commit" "s2-select"
+
+# verify that local INSERT is picked up by create_distributed_table once it commits
+permutation "s2-begin" "s2-insert" "s1-create_distributed_table" "s2-commit" "s2-select"
+# verify that INSERT is distributed once create_distributed_table commits
+permutation "s1-begin" "s1-create_distributed_table" "s2-insert" "s1-commit" "s2-select"
+
+# verify that changes are picked up even in serializable mode
+permutation "s1-begin-serializable" "s2-copy" "s1-create_distributed_table" "s1-commit" "s2-select"
+permutation "s1-begin-serializable" "s2-insert" "s1-create_distributed_table" "s1-commit" "s2-select"