Remove partitions in rm_node with force option

Author: knizhnik

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,81 @@ 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;
+
+	IF OLD.prv IS NOT NULL THEN
+		prev_src_lname := shardman.get_data_lname(OLD.part_name, OLD.prv, OLD.owner);
+	ELSE
+	    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
+		-- ???? Can we really do it now? We will have FDW pointing to removed table... 
+		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);
+		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);
+		PERFORM update shardman.partitions set prv=OLD.prv where owner=me and part_name=OLD.part_name;
+	END IF;
+
+	-- If primary was moved 
+	IF OLD.prv IS NULL THEN
+	   -- And 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

@@ -760,6 +760,16 @@ node_in_cluster(int id)
 	return res;
 }
 
+static void
+rm_partition(int node_id, char const* part_name)
+{
+	char* sql;
+	sql = psprintf("delete from shardman.partitions where owner=%d and part_name='%s'",
+				   node_id, part_name);
+	void_spi(sql);
+	pfree(sql);
+}
+
 /*
  * Remove node, losing all data on it. We
  * - ensure that there is active node with given id in the cluster
@@ -774,6 +784,43 @@ rm_node(Cmd *cmd)
 {
 	int32 node_id = atoi(cmd->opts[0]);
 	char *sql;
+	char **opts;
+	bool force = false;
+	int i, e;
+
+	for (opts = cmd->opts; *opts; opts++)
+	{
+		if (strcmp(*opts, "force") == 0)
+		{
+			force = true;
+			break;
+		}
+	}
+
+	SPI_PROLOG;
+	sql = psprintf("select part_name from shardman.partitions where owner=%d", node_id);
+	e = SPI_execute(sql, true, 0);
+	if (e < 0)
+		shmn_elog(FATAL, "Stmt failed: %s", sql);
+	pfree(sql);
+	if (SPI_processed > 0)
+	{
+		TupleDesc rowdesc = SPI_tuptable->tupdesc;
+		if (!force)
+		{
+
+			ereport(ERROR, (errmsg("Can not remove node with existed partitions"),
+							errhint("Add \"force\" option to remove node with existed partitions.")));
+		}
+		/* Remove partitions belonging to this node */
+		for (i = 0; i < SPI_processed; i++)
+		{
+			HeapTuple tuple = SPI_tuptable->vals[i];
+			char const* partition = SPI_getvalue(tuple, rowdesc, 1);
+			rm_partition(node_id, partition);
+		}
+	}
+	SPI_EPILOG;
 
 	elog(INFO, "Removing node %d ", node_id);
 	if (!node_in_cluster(node_id))