Merge remote-tracking branch 'origin/master'

Merging partition removal.

Author: arssher

init.sql

@@ -72,13 +72,16 @@ END
 $$ LANGUAGE plpgsql;
 
 -- Remove node. Its state will be reset, all shardman data lost.
-CREATE FUNCTION rm_node(node_id int) RETURNS int AS $$
+CREATE FUNCTION rm_node(node_id int, force bool default false) RETURNS int AS $$
 DECLARE
 	c_id int;
 BEGIN
 	INSERT INTO @extschema@.cmd_log VALUES (DEFAULT, 'rm_node')
 										   RETURNING id INTO c_id;
 	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, node_id);
+	IF force THEN
+	    INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, 'force');
+	END IF;
 	RETURN c_id;
 END
 $$ LANGUAGE plpgsql;

shard.sql

@@ -251,6 +251,89 @@ CREATE TRIGGER part_moved AFTER UPDATE ON shardman.partitions
 -- fire trigger only on worker nodes
 ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER part_moved;
 
+
+-- Partition removed: drop old LR channels.
+CREATE FUNCTION part_removed() RETURNS TRIGGER AS $$
+DECLARE
+	me int := shardman.my_id();
+	prev_src_lname text;
+	src_next_lname text;
+	new_primary partitions;
+	drop_slot_delay int := 2000; -- two seconds
+BEGIN
+	RAISE DEBUG '[SHMN] part_removed trigger called for part %, owner %',
+		OLD.part_name, OLD.owner;
+
+	ASSERT (OLD.prv IS NULL OR OLD.nxt IS NULL), 'We currently do not support redundancy level > 2';
+
+	IF OLD.prv IS NOT NULL THEN
+		prev_src_lname := shardman.get_data_lname(OLD.part_name, OLD.prv, OLD.owner);
+	ELSE
+	    -- Primaty is moved
+	    select * from shardman.partitions where owner=OLD.nxt and part_name=OLD.part_name into new_primary;
+	END IF;
+	IF OLD.nxt IS NOT NULL THEN
+		src_next_lname := shardman.get_data_lname(OLD.part_name, OLD.owner, OLD.nxt);
+	END IF;
+
+
+	IF me = OLD.owner THEN -- src node
+		-- If primary part was moved, replace on src node its partition with
+		-- foreign one
+		IF OLD.prv IS NULL THEN
+			PERFORM shardman.replace_usual_part_with_foreign(new_primary);
+		ELSE
+			-- On the other hand, if prev replica existed, drop sub for old
+			-- channel prev -> src
+			PERFORM shardman.eliminate_sub(prev_src_lname);
+		END IF;
+		IF OLD.nxt IS NOT NULL THEN
+			-- If next replica existed, drop pub for old channel src -> next
+ 		    -- Wait sometime to let other node first remove subscription
+		    PERFORM pg_sleep(drop_slot_delay);
+			PERFORM shardman.drop_repslot_and_pub(src_next_lname);
+			PERFORM shardman.remove_sync_standby(src_next_lname);
+		END IF;
+		-- Drop old table anyway
+		EXECUTE format('DROP TABLE IF EXISTS %I', OLD.part_name);
+	ELSEIF me = OLD.prv THEN -- node with prev replica
+		-- Wait sometime to let other node first remove subscription
+		PERFORM pg_sleep(drop_slot_delay);
+		-- Drop pub for old channel prev -> src
+		PERFORM shardman.drop_repslot_and_pub(prev_src_lname);
+		PERFORM shardman.remove_sync_standby(prev_src_lname);
+		-- Update L2-list (TODO: change replication model from chain to star)
+		PERFORM update shardman.partitions set nxt=OLD.nxt where owner=me and part_name=OLD.part_name;
+	ELSEIF me = OLD.nxt THEN -- node with next replica
+		-- Drop sub for old channel src -> next
+		PERFORM shardman.eliminate_sub(src_next_lname);
+		-- Update L2-list (TODO: change replication model from chain to star)
+		PERFORM update shardman.partitions set prv=OLD.prv where owner=me and part_name=OLD.part_name;
+	    -- This replica is promoted to primary node, so drop trigger disabling writes to the table
+		PERFORM readonly_replica_off(part_name);
+		-- Replace FDW with local partition
+	    PERFORM shardman.replace_foreign_part_with_usual(new_primary);
+ 	END IF;
+
+	-- If primary was moved 
+	IF OLD.prv IS NULL THEN
+	   -- then update fdw almost everywhere
+	   PERFORM shardman.update_fdw_server(new_primary);
+	END IF;
+
+	RETURN NULL;
+END
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER part_removed AFTER REMOVE ON shardman.partitions
+	FOR EACH ROW
+	EXECUTE PROCEDURE part_removed();
+-- fire trigger only on worker nodes
+ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER part_removed;
+
+
+
+
 -- Executed on newtail node, see cr_rebuild_lr
 CREATE FUNCTION replica_created_drop_cp_sub(
 	part_name name, oldtail int, newtail int) RETURNS void AS $$

src/pg_shardman.c

@@ -768,6 +768,41 @@ rm_node(Cmd *cmd)
 {
 	int32 node_id = atoi(cmd->opts[0]);
 	char *sql;
+	char **opts;
+	bool force = false;
+	int e;
+
+	for (opts = cmd->opts; *opts; opts++)
+	{
+		if (strcmp(*opts, "force") == 0)
+		{
+			force = true;
+			break;
+		}
+	}
+
+	SPI_PROLOG;
+	if (force)
+	{
+		sql = psprintf("delete from shardman.partitions where owner=%d", node_id);
+		void_spi(sql);
+	}
+	else
+	{
+		bool isnull;
+		sql = psprintf("select count(*) from shardman.partitions where owner=%d", node_id);
+		e = SPI_execute(sql, true, 0);
+		if (e < 0)
+			shmn_elog(FATAL, "Stmt failed: %s", sql);
+		Assert(SPI_processed == 1);
+		if (DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &isnull)) != 0)
+		{
+			ereport(ERROR, (errmsg("Can not remove node with existed partitions"),
+							errhint("Add \"force\" option to remove node with existed partitions.")));
+		}
+	}
+	pfree(sql);
+	SPI_EPILOG;
 
 	elog(INFO, "Removing node %d ", node_id);
 	if (!node_in_cluster(node_id))