Updated LR data channel for move_primary

Author: arssher

bin/common.sh

@@ -18,11 +18,11 @@ master_port=5432
 # declare -a worker_datadirs=("${HOME}/postgres/data2")
 # declare -a worker_ports=("5433")
 
-declare -a worker_datadirs=("${HOME}/postgres/data2" "${HOME}/postgres/data3")
-declare -a worker_ports=("5433" "5434")
+# declare -a worker_datadirs=("${HOME}/postgres/data2" "${HOME}/postgres/data3")
+# declare -a worker_ports=("5433" "5434")
 
-# declare -a worker_datadirs=("${HOME}/postgres/data2" "${HOME}/postgres/data3" "${HOME}/postgres/data4")
-# declare -a worker_ports=("5433" "5434" "5435")
+declare -a worker_datadirs=("${HOME}/postgres/data2" "${HOME}/postgres/data3" "${HOME}/postgres/data4")
+declare -a worker_ports=("5433" "5434" "5435")
 
 #------------------------------------------------------------
 PATH="$PATH:${pgpath}bin/"
@@ -64,5 +64,10 @@ function run_demo()
     psql -p 5433 -c "INSERT INTO pt SELECT generate_series(1, 10), random();"
     psql -c "select shardman.add_node('port=5433');"
     psql -c "select shardman.add_node('port=5434');"
+    psql -p 5433 -c "drop table if exists pt_0;" # drop replica
     psql -c "select shardman.create_hash_partitions(2, 'pt', 'id', 2);"
+
+    psql -c "select shardman.add_node('port=5435');"
+    psql -c "select shardman.move_primary('pt_0', 4);"
+    psql -c "select shardman.create_replica('pt_0', 2);"
 }

shard.sql

@@ -106,6 +106,48 @@ CREATE TRIGGER new_primary AFTER INSERT ON shardman.partitions
 -- fire trigger only on worker nodes
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER new_primary;
 
+-- Executed on node with new primary, see mp_rebuild_lr
+CREATE FUNCTION primary_moved_create_data_pub(p_name name, src int, dst int)
+	RETURNS void AS $$
+DECLARE
+	-- Metadata is not yet updated, so taking nxt from src node
+	replica int := nxt FROM shardman.partitions
+				WHERE part_name = p_name AND owner = src;
+	new_pubname text := shardman.get_data_pubname(p_name, dst);
+	new_subname text := shardman.get_data_subname(p_name, dst, replica);
+BEGIN
+	PERFORM shardman.create_repslot(new_pubname);
+	-- Create publication for new data channel
+	EXECUTE format('DROP PUBLICATION IF EXISTS %I', new_pubname);
+	EXECUTE format('CREATE PUBLICATION %I FOR TABLE %I',
+				   new_pubname, p_name);
+	-- Make this channel sync
+	PERFORM shardman.ensure_sync_standby(new_subname);
+END $$ LANGUAGE plpgsql STRICT;
+
+-- Executed on nearest replica after primary moved, see mp_rebuild_lr
+CREATE FUNCTION primary_moved_create_data_sub(p_name name, src int, dst int)
+	RETURNS void AS $$
+DECLARE
+	-- Metadata is not yet updated, so taking nxt from src node
+	replica int := nxt FROM shardman.partitions
+				WHERE part_name = p_name AND owner = src;
+	new_pubname text := shardman.get_data_pubname(p_name, dst);
+	new_subname text := shardman.get_data_subname(p_name, dst, replica);
+	cp_logname text := shardman.get_cp_logname(p_name, src, dst);
+	new_connstr text := shardman.get_worker_node_connstr(dst);
+BEGIN
+	-- Drop subscription used for copy
+	PERFORM shardman.eliminate_sub(cp_logname);
+	-- Create subscription for new data channel
+	-- It should never exist at this moment, but just in case...
+	PERFORM shardman.eliminate_sub(new_subname);
+	EXECUTE format(
+		'CREATE SUBSCRIPTION %I connection %L
+		PUBLICATION %I with (create_slot = false, slot_name = %L, copy_data = false);',
+		new_subname, new_connstr, new_pubname, new_pubname);
+END $$ LANGUAGE plpgsql STRICT;
+
 -- Update metadata according to primary move
 CREATE FUNCTION primary_moved() RETURNS TRIGGER AS $$
 DECLARE
@@ -115,6 +157,8 @@ BEGIN
 	RAISE DEBUG '[SHARDMAN] primary_moved trigger called for part %, owner %->%',
 		NEW.part_name, OLD.owner, NEW.owner;
 	ASSERT NEW.owner != OLD.owner, 'primary_moved handles only moved parts';
+	ASSERT NEW.nxt = OLD.nxt OR (NEW.nxt IS NULL AND OLD.nxt IS NULL),
+		'both primary and replica must not be moved in one update';
 	IF my_id = OLD.owner THEN -- src node
 		-- Drop publication & repslot used for copy
 		PERFORM shardman.drop_repslot_and_pub(cp_logname);
@@ -141,7 +185,7 @@ CREATE TRIGGER primary_moved AFTER UPDATE ON shardman.partitions
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER primary_moved;
 
 -- Executed on newtail node, see cr_rebuild_lr
-CREATE FUNCTION replica_created_rebuild_drop_cp_sub(
+CREATE FUNCTION replica_created_drop_cp_sub(
 	part_name name, oldtail int, newtail int) RETURNS void AS $$
 DECLARE
 	cp_logname text := shardman.get_cp_logname(part_name, oldtail, newtail);
@@ -152,7 +196,7 @@ BEGIN
 END $$ LANGUAGE plpgsql;
 
 -- Executed on oldtail node, see cr_rebuild_lr
-CREATE FUNCTION replica_created_rebuild_lr_create_data_pub(
+CREATE FUNCTION replica_created_create_data_pub(
 	part_name name, oldtail int, newtail int) RETURNS void AS $$
 DECLARE
 	cp_logname text := shardman.get_cp_logname(part_name, oldtail, newtail);
@@ -175,7 +219,7 @@ BEGIN
 END $$ LANGUAGE plpgsql;
 
 -- Executed on oldtail node, see cr_rebuild_lr
-CREATE FUNCTION replica_created_rebuild_lr_create_data_sub(
+CREATE FUNCTION replica_created_create_data_sub(
 	part_name name, oldtail int, newtail int) RETURNS void AS $$
 DECLARE
 	oldtail_pubname name := shardman.get_data_pubname(part_name, oldtail);

src/include/pg_shardman.h

@@ -48,6 +48,7 @@ extern void cmd_canceled(Cmd *cmd);
 extern char *get_worker_node_connstr(int32 node_id);
 extern int32 get_primary_owner(const char *part_name);
 extern int32 get_reptail_owner(const char *part_name);
+extern int32 get_next_node(const char *part_name, int32 node_id);
 extern char *get_partition_relation(const char *part_name);
 
 #endif							/* PG_SHARDMAN_H */

src/pg_shardman.c

@@ -792,7 +792,7 @@ get_reptail_owner(const char *part_name)
 		shmn_elog(FATAL, "Stmt failed : %s", sql);
 
 	if (SPI_processed == 0)
-		owner = -1;
+		owner = SHMN_INVALID_NODE_ID;
 	else
 	{
 		owner = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
@@ -804,6 +804,41 @@ get_reptail_owner(const char *part_name)
 	return owner;
 }
 
+/*
+ * Get node on which replica next to 'node_id' node in the 'part_name' replica
+ * chain resides. SHMN_INVALID_NODE_ID is returned if such partition doesn't
+ * exist at all or there is no next replica.
+ */
+int32
+get_next_node(const char *part_name, int32 node_id)
+{
+	char *sql;
+	bool isnull;
+	int32 next;
+
+	SPI_PROLOG;
+	sql = psprintf( /* allocated in SPI ctxt, freed with ctxt release */
+		"select nxt from shardman.partitions where part_name = '%s'"
+		" and owner = %d;", part_name, node_id);
+
+	if (SPI_execute(sql, true, 0) < 0)
+		shmn_elog(FATAL, "Stmt failed : %s", sql);
+
+	if (SPI_processed == 0)
+		next = SHMN_INVALID_NODE_ID;
+	else
+	{
+		next = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
+										   SPI_tuptable->tupdesc,
+										   1, &isnull));
+		if (isnull)
+			next = SHMN_INVALID_NODE_ID;
+	}
+
+	SPI_EPILOG;
+	return next;
+}
+
 /*
  * Get relation name of partition part_name. Memory is palloc'ed.
  * NULL is returned, if there is no such partition.