Check for pressence of partition at destination node in rebalance and set_replelvel

Author: knizhnik

shard.sql

@@ -157,7 +157,7 @@ BEGIN
 		PERFORM shardman.eliminate_sub(prev_lname);
 		EXECUTE format(
 			'CREATE SUBSCRIPTION %I connection %L
-		PUBLICATION %I with (create_slot = false, slot_name = %L, copy_data = false, synchronous_commit = on);',
+		PUBLICATION %I with (create_slot = false, slot_name = %L, copy_data = false, synchronous_commit = local);',
 		prev_lname, prev_connstr, prev_lname, prev_lname);
 		-- If we have prev, we are replica
 		PERFORM shardman.readonly_replica_on(p_name::regclass);
@@ -177,7 +177,7 @@ BEGIN
 	PERFORM shardman.eliminate_sub(lname);
 	EXECUTE format(
 		'CREATE SUBSCRIPTION %I connection %L
-		PUBLICATION %I with (create_slot = false, slot_name = %L, copy_data = false, synchronous_commit = on);',
+		PUBLICATION %I with (create_slot = false, slot_name = %L, copy_data = false, synchronous_commit = local);',
 		lname, dst_connstr, lname, lname);
 END $$ LANGUAGE plpgsql STRICT;
 
@@ -664,8 +664,8 @@ CREATE FUNCTION readonly_table_on(relation regclass)
 	RETURNS void AS $$
 BEGIN
 	-- Create go away trigger to prevent any modifications
-	-- PERFORM shardman.readonly_table_off(relation);
-	-- PERFORM shardman.create_modification_triggers(relation, 'shardman_readonly', 'shardman.go_away()');
+	PERFORM shardman.readonly_table_off(relation);
+	PERFORM shardman.create_modification_triggers(relation, 'shardman_readonly', 'shardman.go_away()');
 END
 $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION go_away() RETURNS TRIGGER AS $$
@@ -688,8 +688,8 @@ CREATE FUNCTION readonly_replica_on(relation regclass)
 BEGIN
 	RAISE DEBUG '[SHMN] table % made read-only for all but apply workers',
 		relation;
-	-- PERFORM shardman.readonly_replica_off(relation);
-	-- PERFORM shardman.create_modification_triggers(relation, 'shardman_readonly_replica', 'shardman.ror_go_away()');
+	PERFORM shardman.readonly_replica_off(relation);
+	PERFORM shardman.create_modification_triggers(relation, 'shardman_readonly_replica', 'shardman.ror_go_away()');
 END $$ LANGUAGE plpgsql STRICT;
 -- This function is impudent because it is used as both stmt and row trigger.
 -- The idea is that we must never reach RETURN NEW after stmt row trigger,

src/copypart.c

@@ -370,7 +370,7 @@ init_cp_state(CopyPartState *cps)
 		" including storage);"
 		" drop subscription if exists %s cascade;"
 		" create subscription %s connection '%s' publication %s with"
-		"   (create_slot = false, slot_name = '%s');",
+		"   (create_slot = false, slot_name = '%s', synchronous_commit = local);",
 		cps->part_name,
 		cps->part_name, cps->relation,
 		cps->logname,

src/include/pg_shardman.h

@@ -131,5 +131,5 @@ extern int32 get_prev_node(const char *part_name, int32 node_id, bool *part_exis
 extern char *get_partition_relation(const char *part_name);
 extern Partition *get_parts(const char *relation, uint64 *num_parts);
 extern RepCount *get_repcount(const char *relation, uint64 *num_parts);
-
+extern bool node_has_partition(int32 node, const char *part_name);
 #endif							/* PG_SHARDMAN_H */

src/pg_shardman.c

@@ -1121,6 +1121,34 @@ get_prev_node(const char *part_name, int32 node_id, bool *part_exists)
 	return prev;
 }
 
+bool
+node_has_partition(int32 node, const char *part_name)
+{
+	char *sql;
+	int64 count = 0;
+	SPI_XACT_STATUS;
+
+	SPI_PROLOG;
+
+	sql = psprintf("select count(*) from shardman.partitions"
+				   " where part_name = '%s' and owner = %d", part_name, node);
+
+	if (SPI_execute(sql, true, 0) < 0)
+	{
+		shmn_elog(FATAL, "Stmt failed : %s", sql);
+	}
+	if (SPI_processed != 0)
+	{
+		bool isnull;
+		count = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0],
+											SPI_tuptable->tupdesc,
+											1,
+											&isnull));
+	}
+	SPI_EPILOG;
+	return count != 0;
+}
+
 /*
  * Get relation name of partition part_name. Memory is palloc'ed.
  * NULL is returned, if there is no such partition.

src/shard.c

@@ -220,12 +220,16 @@ rebalance(Cmd *cmd)
 	for (part_idx = 0, worker_idx = 0; part_idx < num_parts; part_idx++)
 	{
 		Partition part = parts[part_idx];
-		int32 worker = workers[worker_idx];
+		int32 worker;
 		MovePartState *mps = palloc0(sizeof(MovePartState));
 
+		do {
+			worker = workers[worker_idx];
+			worker_idx = (worker_idx + 1) % num_workers;
+		} while (node_has_partition(worker, part.part_name));
+
 		init_mp_state(mps, part.part_name, part.owner, worker);
 		tasks[part_idx] = (CopyPartState *) mps;
-		worker_idx = (worker_idx + 1) % num_workers;
 	}
 
 	exec_tasks(tasks, num_parts);
@@ -286,8 +290,14 @@ set_replevel(Cmd *cmd)
 			if (rc.count < replevel)
 			{
 				CreateReplicaState *crs = palloc0(sizeof(CreateReplicaState));
-				int32 dst_node = workers[rand() % num_workers];
+				int32 dst_node;
+
+				do {
+					dst_node = workers[rand() % num_workers];
+				} while (node_has_partition(dst_node, rc.part_name));
+
 				init_cr_state(crs, rc.part_name, dst_node);
+
 				tasks[ntasks] = (CopyPartState *) crs;
 				ntasks++;
 				shmn_elog(DEBUG1, "Adding replica for shard %s on node %d",