Merge remote-tracking branch 'origin/master'

* Sane destination choosing in rebalance and set_replevel;
* synchronous_commit on -> local in both data and copy subs. It is safe, because
  we are moving to spider model and don't have sync streaming replication
  anywhere (during replica move we don't need sync replication).
* Keeping replica protecting triggers still commented as fix is not merged into
  PG yet.

Author: arssher

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;
 

src/copypart.c

@@ -363,7 +363,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,37 @@ get_prev_node(const char *part_name, int32 node_id, bool *part_exists)
 	return prev;
 }
 
+/*
+ * Node 'node' holds primary or replica of part_name?
+ */
+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",