create_replica almost done in plpgsql, now it needs to be rewritten.

Replication slots can't be created from transactions that wrote something, and
this is another reason to manage LR channels entirely from libpq. At least
repslots creation must be moved out of trigger.

Also I have removed partition numbering as there is apparently no reason to keep
it: partitions can refer to nodes keeping its previous and next replica.

Author: arssher

init.sql

@@ -213,6 +213,14 @@ BEGIN
 				   pg_replication_slots WHERE slot_name = %L', slot_name);
 END $$ LANGUAGE plpgsql STRICT;
 
+-- Drop with fire repslot and publication with the same name. Useful for 1-to-1
+-- pub-sub.
+CREATE FUNCTION drop_repslot_and_pub(pub name) RETURNS void AS $$
+BEGIN
+	EXECUTE format('DROP PUBLICATION IF EXISTS %I', pub);
+	PERFORM shardman.drop_repslot(pub, true);
+END $$ LANGUAGE plpgsql STRICT;
+
 -- If sub exists, disable it, detach repslot from it and possibly drop. We
 -- manage repslots ourselves, so it is essential to detach rs before dropping
 -- sub, and repslots can't be detached while subscription is active.

membership.sql

@@ -94,3 +94,15 @@ BEGIN
 	RETURN n_id;
 END
 $$ LANGUAGE plpgsql;
+
+-- Get connstr of worker node with id node_id. ERROR is raised if there isn't
+-- one.
+CREATE OR REPLACE FUNCTION get_worker_node_connstr(node_id int) RETURNS text as $$
+DECLARE
+	connstr text := connstring FROM shardman.nodes WHERE id = node_id AND worker;
+BEGIN
+	IF connstr IS NULL THEN
+		RAISE EXCEPTION 'Worker node with id % not found', node_id;
+	END IF;
+	RETURN connstr;
+END $$ LANGUAGE plpgsql STRICT;

postgresql.conf.master.template

@@ -1,4 +1,6 @@
 # master-specific part
 shardman.master = on
 shardman.master_dbname = ars
-shardman.master_connstring = 'port=5432'
+shardman.master_connstring = 'port=5432'
+shardman.cmd_retry_naptime = 3000
+shardman.poll_interval = 3000

shard.sql

@@ -54,7 +54,7 @@ ALTER TABLE shardman.tables ENABLE REPLICA TRIGGER new_table_worker_side;
 CREATE FUNCTION new_table_master_side() RETURNS TRIGGER AS $$
 BEGIN
 	INSERT INTO shardman.partitions
-	SELECT part_name, 0, NULL, NULL, NEW.relation AS relation, NEW.initial_node AS owner
+	SELECT part_name, NEW.initial_node AS owner, NULL, NULL, NEW.relation AS relation
 	  FROM (SELECT part_name FROM shardman.gen_part_names(
 		  NEW.relation, NEW.partitions_count))
 			   AS partnames;
@@ -68,21 +68,18 @@ CREATE TRIGGER new_table_master_side AFTER INSERT ON shardman.tables
 -- Partitions
 ------------------------------------------------------------
 
--- Primary shard and its replicas compose a doubly-linked list with 0 shard in
--- the beginning.
+-- Primary shard and its replicas compose a doubly-linked list: nxt refers to
+-- the node containing next replica, prv to node with previous replica (or
+-- primary, if we are the first replica). If prv is NULL, this is primary
+-- replica. We don't number parts separately since we are not ever going to
+-- allow several copies of the same partition on one node.
 CREATE TABLE partitions (
 	part_name text,
-	-- Shard number. 0 means primary shard.
-	num serial,
-	nxt int,
-	prv int,
+	owner int NOT NULL REFERENCES nodes(id), -- node on which partition lies
+	prv int REFERENCES nodes(id),
+	nxt int REFERENCES nodes(id),
 	relation text NOT NULL REFERENCES tables(relation),
-	owner int REFERENCES nodes(id), -- node on which partition lies
-	PRIMARY KEY (part_name, num),
-	FOREIGN KEY (part_name, nxt) REFERENCES shardman.partitions(part_name, num),
-	FOREIGN KEY (part_name, prv) REFERENCES shardman.partitions(part_name, num),
-	-- primary has no prv, replica must have prv
-	CONSTRAINT prv_existence CHECK (num = 0 OR prv IS NOT NULL)
+	PRIMARY KEY (part_name, owner)
 );
 
 ------------------------------------------------------------
@@ -96,43 +93,31 @@ CREATE TABLE partitions (
 -- it.
 CREATE FUNCTION new_primary() RETURNS TRIGGER AS $$
 BEGIN
-	IF NEW.owner != (SELECT shardman.get_node_id()) THEN
+	RAISE DEBUG '[SHARDMAN] new_primary trigger called for part %, owner %',
+		NEW.part_name, NEW.owner;
+	IF NEW.owner != shardman.get_node_id() THEN
 		PERFORM shardman.replace_usual_part_with_foreign(NEW);
 	END IF;
 	RETURN NULL;
 END
 $$ LANGUAGE plpgsql;
 CREATE TRIGGER new_primary AFTER INSERT ON shardman.partitions
-	FOR EACH ROW WHEN (NEW.num = 0) EXECUTE PROCEDURE new_primary();
+	FOR EACH ROW WHEN (NEW.prv IS NULL) EXECUTE PROCEDURE new_primary();
 -- fire trigger only on worker nodes
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER new_primary;
 
--- Replace foreign table-partition with local. The latter must exist!
--- Foreign table will be dropped.
-CREATE FUNCTION replace_foreign_part_with_usual(part partitions)
-	RETURNS void AS $$
-DECLARE
-	fdw_part_name name;
-BEGIN
-	ASSERT to_regclass(part.part_name) IS NOT NULL;
-	SELECT shardman.get_fdw_part_name(part.part_name) INTO fdw_part_name;
-	EXECUTE format('SELECT replace_hash_partition(%L, %L);',
-				   fdw_part_name, part.part_name);
-	EXECUTE format('DROP FOREIGN TABLE %I;', fdw_part_name);
-END $$ LANGUAGE plpgsql;
-
 -- Update metadata according to primary move
 CREATE FUNCTION primary_moved() RETURNS TRIGGER AS $$
 DECLARE
-	cp_logname text := format('shardman_copy_%s_%s_%s',
-							   OLD.part_name, OLD.owner, NEW.owner);
+	cp_logname text := shardman.get_cp_logname(OLD.part_name, OLD.owner, NEW.owner);
 	my_id int := shardman.get_node_id();
 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';
 	IF my_id = OLD.owner THEN -- src node
 		-- Drop publication & repslot used for copy
-		EXECUTE format('DROP PUBLICATION IF EXISTS %I', cp_logname);
-		PERFORM shardman.drop_repslot(cp_logname, true);
+		PERFORM shardman.drop_repslot_and_pub(cp_logname);
 		-- On src node, replace its partition with foreign one
 		PERFORM shardman.replace_usual_part_with_foreign(NEW);
 	ELSEIF my_id = NEW.owner THEN -- dst node
@@ -148,44 +133,75 @@ BEGIN
 END
 $$ LANGUAGE plpgsql;
 CREATE TRIGGER primary_moved AFTER UPDATE ON shardman.partitions
-	FOR EACH ROW EXECUTE PROCEDURE primary_moved();
+	FOR EACH ROW WHEN (OLD.prv is NULL AND NEW.prv IS NULL -- it is primary
+					   AND OLD.owner != NEW.owner -- and it is really moved
+					   AND OLD.part_name = NEW.part_name) -- sanity check
+	EXECUTE PROCEDURE primary_moved();
 -- fire trigger only on worker nodes
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER primary_moved;
 
 -- Update metadata according to new replica creation.
+-- Old tail part is still read-only when this called. There are two main jobs
+-- to do: set up LR sync channel between old tail and new replica and update fdw
+-- everywhere. For the former we could configure already existing channel used
+-- for partition copy, but we will not do that, because
+-- * It is not easier than creating new pub & sub: we have to rename pub, drop
+--   and create repslot (there is no way to rename it), rename sub, alter sub's
+--   slot_name, alter sub's publication, probably rename sub application name,
+--   probably run REFRESH (which requires alive pub just as CREATE SUBSCRIPTION)
+--   and hope that everything will be ok. Not sure about refreshing, though -- I
+--   don't know is it ok not doing it if tables didn't change. Doc says it
+--   should be executed.
+-- * Since it is not possible to rename repslot and and it is not possible to
+--   specify since which lsn start replication, tables must be synced anyway
+--   during these operations, so what the point of reusing old sub? And copypart
+--   in shard.c really cares that tables are synced at this moment and src is
+--   read-only.
 CREATE FUNCTION replica_created() RETURNS TRIGGER AS $$
 DECLARE
-	cp_logname text := format('shardman_copy_%s_%s_%s',
-							   NEW.part_name, NEW.prv, NEW.owner);
+	cp_logname text := shardman.get_cp_logname(NEW.part_name, NEW.prv, NEW.owner);
+	oldtail_pubname name := shardman.get_data_pubname(NEW.part_name, NEW.prv);
+	oldtail_connstr text := shardman.get_worker_node_connstr(NEW.prv);
+	newtail_subname name := shardman.get_data_subname(NEW.part_name, NEW.prv, NEW.owner);
 	my_id int := shardman.get_node_id();
 BEGIN
-	-- ASSERT NEW.owner != OLD.owner, 'partition_moved handles only moved parts';
-	-- cp_logname := format('shardman_copy_%s_%s_%s',
-	-- 						   OLD.part_name, OLD.owner, NEW.owner);
-	-- my_id := (SELECT shardman.get_node_id());
-	-- IF my_id = OLD.owner THEN -- src node
-	-- 	-- Drop publication & repslot used for copy
-	-- 	EXECUTE format('DROP PUBLICATION IF EXISTS %I', cp_logname);
-	-- 	PERFORM shardman.drop_repslot(cp_logname, true);
-	-- 	-- On src node, replace its partition with foreign one
-	-- 	PERFORM shardman.replace_usual_part_with_foreign(NEW);
-	-- ELSEIF my_id = NEW.owner THEN -- dst node
-	-- 	-- Drop subscription used for copy
-	-- 	PERFORM shardman.eliminate_sub(cp_logname);
-	-- 	PERFORM shardman.replace_foreign_part_with_usual(NEW);
-	-- ELSE -- other nodes
-	-- 	-- just update foreign server
-	-- 	PERFORM shardman.update_fdw_server(NEW);
-	-- END IF;
-	-- RETURN NULL;
+	RAISE DEBUG '[SHARDMAN] replica_created trigger called';
+	IF my_id = NEW.prv THEN -- old tail node
+		-- Drop publication & repslot used for copy
+		PERFORM shardman.drop_repslot_and_pub(cp_logname);
+		-- Create publication & repslot for new data channel
+		PERFORM shardman.create_repslot(oldtail_pubname);
+		EXECUTE format('DROP PUBLICATION IF EXISTS %I', oldtail_pubname);
+		EXECUTE format('CREATE PUBLICATION %I FOR TABLE %I',
+					   oldtail_pubname, NEW.part_name);
+		-- Make this channel sync
+		PERFORM shardman.ensure_sync_standby(newtail_subname);
+		-- Now it is safe to make old tail writable again
+		PERFORM shardman.readonly_table_off(relation);
+	ELSEIF my_id = NEW.owner THEN -- created replica, i.e. new tail node
+		-- Drop subscription used for copy
+		PERFORM shardman.eliminate_sub(cp_logname);
+		-- And create subscription for new data channel
+		-- It should never exist at this moment, but just in case...
+		PERFORM shardman.eliminate_sub(newtail_subname);
+		EXECUTE format(
+			'CREATE SUBSCRIPTION %I connection %L
+			PUBLICATION %I with (create_slot = false, slot_name = %L);',
+			newtail_subname, oldtail_connstr, oldtail_pubname, oldtail_pubname);
+		-- Now fdw connstring to this part should include only primary and myself
+		PERFORM shardman.update_fdw_server(NEW);
+	ELSE -- other nodes
+		-- just update fdw connstr to add new replica
+		PERFORM shardman.update_fdw_server(NEW);
+	END IF;
+	RETURN NULL;
 END
 $$ LANGUAGE plpgsql;
 CREATE TRIGGER replica_created AFTER INSERT ON shardman.partitions
-	FOR EACH ROW WHEN (NEW.num != 0) EXECUTE PROCEDURE replica_created();
+	FOR EACH ROW WHEN (NEW.prv IS NOT NULL) EXECUTE PROCEDURE replica_created();
 -- fire trigger only on worker nodes
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER replica_created;
 
-
 -- Otherwise partitioned tables on worker nodes not will be dropped properly,
 -- see pathman's docs.
 ALTER EVENT TRIGGER pathman_ddl_trigger ENABLE ALWAYS;
@@ -194,8 +210,8 @@ ALTER EVENT TRIGGER pathman_ddl_trigger ENABLE ALWAYS;
 -- Funcs related to fdw
 ------------------------------------------------------------
 
--- We use _fdw suffix for foreign tables to avoid interleaving with real
--- ones.
+-- Convention: we use _fdw suffix for foreign tables to avoid interleaving with
+-- real ones.
 CREATE FUNCTION get_fdw_part_name(part_name name) RETURNS name AS $$
 BEGIN
 	RETURN format('%s_fdw', part_name);
@@ -324,6 +340,20 @@ BEGIN
 	EXECUTE format('DROP TABLE %I', part.part_name);
 END $$ LANGUAGE plpgsql;
 
+-- Replace foreign table-partition with local. The latter must exist!
+-- Foreign table will be dropped.
+CREATE FUNCTION replace_foreign_part_with_usual(part partitions)
+	RETURNS void AS $$
+DECLARE
+	fdw_part_name name;
+BEGIN
+	ASSERT to_regclass(part.part_name) IS NOT NULL;
+	SELECT shardman.get_fdw_part_name(part.part_name) INTO fdw_part_name;
+	EXECUTE format('SELECT replace_hash_partition(%L, %L);',
+				   fdw_part_name, part.part_name);
+	EXECUTE format('DROP FOREIGN TABLE %I;', fdw_part_name);
+END $$ LANGUAGE plpgsql;
+
 -- Options to postgres_fdw are specified in two places: user & password in user
 -- mapping and everything else in create server. The problem is that we use
 -- single connstring, however user mapping and server doesn't understand this
@@ -493,3 +523,36 @@ BEGIN
 							 AS range(num)) AS range;
 END
 $$ LANGUAGE plpgsql;
+
+-- Convention about pub, sub and repslot name used for copying part part_name
+-- from src node to dst node.
+CREATE FUNCTION get_cp_logname(part_name text, src int, dst int)
+	RETURNS name AS $$
+BEGIN
+	RETURN format('shardman_copy_%s_%s_%s', part_name, src, dst);
+END $$ LANGUAGE plpgsql STRICT;
+
+-- Convention about pub and repslot name used for data replication from part
+-- on pub_node node to any part. We don't change pub and repslot while
+-- switching subs, so sub node is not included here.
+CREATE FUNCTION get_data_pubname(part_name text, pub_node int)
+	RETURNS name AS $$
+BEGIN
+	RETURN format('shardman_data_%s_%s', part_name, pub_node);
+END $$ LANGUAGE plpgsql STRICT;
+
+-- Convention about sub and application_name used for data replication. We do
+-- recreate sub while switching pub, so pub node is included here.
+-- See comment to replica_created on why we don't reuse subs.
+CREATE FUNCTION get_data_subname(part_name text, pub_node int, sub_node int)
+	RETURNS name AS $$
+BEGIN
+	RETURN format('shardman_data_%s_%s_%s', part_name, pub_node, sub_node);
+END $$ LANGUAGE plpgsql STRICT;
+
+-- Make sure that standby_name is present in synchronous_standby_names. If not,
+-- add it via ALTER SYSTEM and SIGHUP postmaster to reread conf.
+CREATE FUNCTION ensure_sync_standby(newtail_subname text) RETURNS void as $$
+BEGIN
+	RAISE DEBUG '[SHARDMAN] imagine standby updated';
+END $$ LANGUAGE plpgsql STRICT;

src/include/pg_shardman.h

@@ -45,10 +45,9 @@ extern void check_for_sigterm(void);
 extern uint64 void_spi(char *sql);
 extern void update_cmd_status(int64 id, const char *new_status);
 extern void cmd_canceled(Cmd *cmd);
-extern char *get_worker_node_connstr(int node_id);
+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, int32 *owner,
-							   int32 *partnum);
+extern int32 get_reptail_owner(const char *part_name);
 extern char *get_partition_relation(const char *part_name);
 
 #endif							/* PG_SHARDMAN_H */

src/pg_shardman.c

@@ -707,7 +707,7 @@ rm_node(Cmd *cmd)
  * NULL is returned, if there is no such node.
  */
 char*
-get_worker_node_connstr(int node_id)
+get_worker_node_connstr(int32 node_id)
 {
 	MemoryContext oldcxt = CurrentMemoryContext;
 	char *sql = psprintf("select connstring from shardman.nodes where id = %d"
@@ -753,7 +753,7 @@ get_primary_owner(const char *part_name)
 
 	SPI_PROLOG;
 	sql = psprintf( /* allocated in SPI ctxt, freed with ctxt release */
-		"select owner from shardman.partitions where part_name = '%s' and num = 0;",
+		"select owner from shardman.partitions where part_name = '%s' and prv IS NULL;",
 		part_name);
 
 	if (SPI_execute(sql, true, 0) < 0)
@@ -774,15 +774,14 @@ get_primary_owner(const char *part_name)
 
 /*
  * Get node id on which the last replica in the 'part_name' replica chain
- * resides, and its partnum.  -1 is returned if such partition doesn't exist
- * at all.
+ * resides. -1 is returned if such partition doesn't exist at all.
  */
 int32
-get_reptail_owner(const char *part_name, int32 *owner, int32 *partnum)
+get_reptail_owner(const char *part_name)
 {
 	char *sql;
 	bool isnull;
-	int result = 0;
+	int owner;
 
 	SPI_PROLOG;
 	sql = psprintf( /* allocated in SPI ctxt, freed with ctxt release */
@@ -793,19 +792,16 @@ get_reptail_owner(const char *part_name, int32 *owner, int32 *partnum)
 		shmn_elog(FATAL, "Stmt failed : %s", sql);
 
 	if (SPI_processed == 0)
-		result = -1;
+		owner = -1;
 	else
 	{
-		*owner = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
-											 SPI_tuptable->tupdesc,
-											 1, &isnull));
-		*partnum = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
-											   SPI_tuptable->tupdesc,
-											   2, &isnull));
+		owner = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],
+											SPI_tuptable->tupdesc,
+											1, &isnull));
 	}
 
 	SPI_EPILOG;
-	return result;
+	return owner;
 }
 
 /*