Merge 0884783de1 into 94151c9aef

src/backend/distributed/commands/multi_copy.c

@@ -72,6 +72,7 @@
 #include "nodes/makefuncs.h"
 #include "tsearch/ts_locale.h"
 #include "utils/builtins.h"
+#include "utils/inval.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
 #include "utils/memutils.h"
@@ -1751,6 +1752,14 @@ CitusCopyDestReceiverStartup(DestReceiver *dest, int operation,
 	LockShardListMetadata(shardIntervalList, ShareLock);
 	LockShardListResources(shardIntervalList, ShareLock);
 
+	/*
+	 * We might have some concurrent metadata changes. In order to get the changes,
+	 * we first need to accept the cache invalidation messages and then access the
+	 * cache entry to trigger the cache entry rebuild.
+	 */
+	AcceptInvalidationMessages();
+	cacheEntry = DistributedTableCacheEntry(tableId);
+
 	/* keep the table metadata to avoid looking it up for every tuple */
 	copyDest->tableMetadata = cacheEntry;
 

src/test/regress/expected/isolation_add_node_vs_copy.out

@@ -0,0 +1,83 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1-begin s1-add-second-worker s2-begin s2-copy-to-reference-table s1-commit s2-commit s2-print-content s1-remove-second-worker s1-begin s1-add-second-worker s2-begin s2-copy-to-reference-table s1-commit s2-commit s2-print-content
+create_reference_table
+
+               
+step s1-begin: 
+    BEGIN;
+
+step s1-add-second-worker: 
+   	SELECT nodename, nodeport, isactive FROM master_add_node('localhost', 57638);
+
+nodename       nodeport       isactive       
+
+localhost      57638          t              
+step s2-begin: 
+    BEGIN;
+
+step s2-copy-to-reference-table: 
+	COPY test_reference_table FROM PROGRAM 'echo "1\n2\n3\n4\n5"';
+ <waiting ...>
+step s1-commit: 
+    COMMIT;
+
+step s2-copy-to-reference-table: <... completed>
+step s2-commit: 
+    COMMIT;
+
+step s2-print-content: 
+	SELECT 
+		nodeport, success, result 
+	FROM 
+		run_command_on_placements('test_reference_table', 'select count(*) from %s')
+	ORDER BY
+		nodeport;
+
+nodeport       success        result         
+
+57637          t              5              
+57638          t              5              
+step s1-remove-second-worker: 
+   	SELECT master_remove_node('localhost', 57638);
+
+master_remove_node
+
+               
+step s1-begin: 
+    BEGIN;
+
+step s1-add-second-worker: 
+   	SELECT nodename, nodeport, isactive FROM master_add_node('localhost', 57638);
+
+nodename       nodeport       isactive       
+
+localhost      57638          t              
+step s2-begin: 
+    BEGIN;
+
+step s2-copy-to-reference-table: 
+	COPY test_reference_table FROM PROGRAM 'echo "1\n2\n3\n4\n5"';
+ <waiting ...>
+step s1-commit: 
+    COMMIT;
+
+step s2-copy-to-reference-table: <... completed>
+step s2-commit: 
+    COMMIT;
+
+step s2-print-content: 
+	SELECT 
+		nodeport, success, result 
+	FROM 
+		run_command_on_placements('test_reference_table', 'select count(*) from %s')
+	ORDER BY
+		nodeport;
+
+nodeport       success        result         
+
+57637          t              10             
+57638          t              10             
+nodename       nodeport       isactive       
+
+localhost      57638          t              

src/test/regress/isolation_schedule

@@ -2,3 +2,4 @@ test: isolation_cluster_management
 test: isolation_dml_vs_repair
 test: isolation_concurrent_dml isolation_data_migration
 test: isolation_drop_shards
+test: isolation_add_node_vs_copy

src/test/regress/specs/isolation_add_node_vs_copy.spec

@@ -0,0 +1,72 @@
+# remove one of the nodes for the purpose of the test
+setup
+{	
+	SELECT master_remove_node('localhost', 57638);
+	
+    CREATE TABLE test_reference_table (test_id integer);
+    SELECT create_reference_table('test_reference_table');
+}
+
+# ensure that both nodes exists for the remaining of the isolation tests
+teardown
+{
+	DROP TABLE test_reference_table;
+	SELECT nodename, nodeport, isactive FROM master_add_node('localhost', 57637);
+	SELECT nodename, nodeport, isactive FROM master_add_node('localhost', 57638);
+}
+
+session "s1"
+
+step "s1-begin"
+{
+    BEGIN;
+}
+
+step "s1-add-second-worker"
+{
+   	SELECT nodename, nodeport, isactive FROM master_add_node('localhost', 57638);
+}
+
+step "s1-remove-second-worker"
+{
+   	SELECT master_remove_node('localhost', 57638);
+}
+
+step "s1-commit"
+{
+    COMMIT;
+}
+
+session "s2"
+
+step "s2-begin"
+{
+    BEGIN;
+}
+
+step "s2-copy-to-reference-table" 
+{
+	COPY test_reference_table FROM PROGRAM 'echo "1\n2\n3\n4\n5"';
+}
+
+step "s2-commit"
+{
+    COMMIT;
+}
+
+step "s2-print-content"
+{
+	SELECT 
+		nodeport, success, result 
+	FROM 
+		run_command_on_placements('test_reference_table', 'select count(*) from %s')
+	ORDER BY
+		nodeport;
+}
+
+
+# verify that copy gets the invalidation and re-builts its metadata cache
+# note that we need to run the same test twice to ensure that metadata is cached
+# otherwise the test would be useless since the cache would be empty and the 
+# metadata data is gathered from the tables directly
+permutation "s1-begin" "s1-add-second-worker" "s2-begin" "s2-copy-to-reference-table" "s1-commit" "s2-commit" "s2-print-content" "s1-remove-second-worker" "s1-begin" "s1-add-second-worker" "s2-begin" "s2-copy-to-reference-table" "s1-commit" "s2-commit" "s2-print-content"