Make it possible to specify redundancy in create_hash_partition

Author: knizhnik

pg_shardman--1.0.sql

@@ -115,14 +115,15 @@ BEGIN
 	THEN
 		-- Take all nodes in replicationg group excluding myself
 		FOR node IN SELECT * FROM shardman.nodes WHERE replication_group = repl_group LOOP
-		sync_standbys :=
-			coalesce(ARRAY(SELECT format('sub_%s_%s', id, node.id) FROM shardman.nodes
+			-- Construct list of synchronous standbyes=subscriptions to this node
+			sync_standbys :=
+				 coalesce(ARRAY(SELECT format('sub_%s_%s', id, node.id) FROM shardman.nodes
 				   WHERE replication_group = repl_group AND id <> node.id), '{}'::text[]);
-		sync := format('%s%s:ALTER SYSTEM SET synchronous_standby_names to ''FIRST %s (%s)'';',
-					   sync, node.id, array_length(sync_standbys, 1),
-					   array_to_string(sync_standbys, ','));
-		conf := format('%s%s:SELECT pg_reload_conf();', conf, node.id);
-	END LOOP;
+			sync := format('%s%s:ALTER SYSTEM SET synchronous_standby_names to ''FIRST %s (%s)'';',
+				 sync, node.id, array_length(sync_standbys, 1),
+				 array_to_string(sync_standbys, ','));
+			conf := format('%s%s:SELECT pg_reload_conf();', conf, node.id);
+		END LOOP;
 
 	    PERFORM shardman.broadcast(sync, sync_commit => true);
 	    PERFORM shardman.broadcast(conf);
@@ -299,7 +300,7 @@ BEGIN
 			    pubs := format('%s%s:ALTER PUBLICATION node_%s ADD TABLE %I;',
 				     pubs, new_master_id, repl.node_id, part.part_name);
 				-- And refresh subscriptions and replicas
-				subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION;',
+				subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION WITH (copy_data=false);',
 					 subs, repl.node_id, repl.node_id, new_master_id);
 			END LOOP;
 
@@ -343,7 +344,7 @@ $$ LANGUAGE plpgsql;
 -- It also scatter partitions through all nodes.
 -- This function expects that empty table is created at shardlord.
 -- So it can be executed only at shardlord and there is no need to redirect this function to shardlord.
-CREATE FUNCTION create_hash_partitions(rel regclass, expr text, part_count int)
+CREATE FUNCTION create_hash_partitions(rel regclass, expr text, part_count int, redundancy int = 0)
 RETURNS void AS $$
 DECLARE
 	create_table text;
@@ -362,7 +363,7 @@ DECLARE
 	i int;
 	n_nodes int;
 BEGIN
-	IF EXISTS(SELECT relation FROM shardman.partitions WHERE relation = rel_name)
+	IF EXISTS(SELECT relation FROM shardman.tables WHERE relation = rel_name)
 	THEN
 		RAISE EXCEPTION 'Table % is already sharded', rel_name;
 	END IF;
@@ -419,13 +420,18 @@ BEGIN
 	PERFORM shardman.broadcast(create_fdws);
 	-- Broadcast replace hash partition commands
 	PERFORM shardman.broadcast(replace_parts);
+
+	IF redundancy <> 0
+	THEN
+		PERFORM shardman.set_redundancy(rel, redundancy, copy_data => false);
+	END IF;
 END
 $$ LANGUAGE plpgsql;
 
 -- Provide requested level of redundancy. 0 means no redundancy.
 -- If existing level of redundancy is greater than specified, then right now this
 -- function does nothing.
-CREATE FUNCTION set_redundancy(rel regclass, redundancy int)
+CREATE FUNCTION set_redundancy(rel regclass, redundancy int, copy_data bool = true)
 RETURNS void AS $$
 DECLARE
 	part shardman.partitions;
@@ -435,12 +441,17 @@ DECLARE
 	pubs text = '';
 	subs text = '';
 	rel_name text = rel::text;
+	sub_options text = '';
 BEGIN
 	IF shardman.redirect_to_shardlord(format('set_redundancy(%L, %L)', rel_name, redundancy))
 	THEN
 		RETURN;
 	END IF;
 
+	IF NOT copy_data THEN
+	    sub_options := ' WITH (copy_data=false)';
+	END IF;
+
 	-- Loop through all partitions of this table
 	FOR part IN SELECT * from shardman.partitions where relation=rel_name
 	LOOP
@@ -460,16 +471,16 @@ BEGIN
 				-- Establish publications and subscriptions for this partition
 				pubs := format('%s%s:ALTER PUBLICATION node_%s ADD TABLE %I;',
 					 pubs, part.node_id, repl_node, part.part_name);
-				subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION;',
-					 subs, repl_node, repl_node, part.node_id);
+				subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION%s;',
+					 subs, repl_node, repl_node, part.node_id, sub_options);
 			END LOOP;
 		END IF;
 	END LOOP;
 
 	-- Broadcast alter publication commands
 	PERFORM shardman.broadcast(pubs);
 	-- Broadcast alter subscription commands
-	PERFORM shardman.broadcast(subs);
+	PERFORM shardman.broadcast(subs, synchronous => copy_data);
 
 	-- This function doesn't wait completion of replication sync
 END
@@ -585,8 +596,8 @@ BEGIN
 			 			  %s:ALTER PUBLICATION node_%s ADD TABLE %I;',
 			 pubs, src_node_id, repl_node_id, mv_part_name,
 			 	   dst_node_id, repl_node_id, mv_part_name);
-		subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION;
-			 			  %s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION;',
+		subs := format('%s%s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION WITH (copy_data=false);
+			 			  %s:ALTER SUBSCRIPTION sub_%s_%s REFRESH PUBLICATION WITH (copy_data=false);',
 			 subs, repl_node_id, repl_node_id, src_node_id,
 			 	   repl_node_id, repl_node_id, dst_node_id);
 	END LOOP;
@@ -691,8 +702,11 @@ RETURNS text AS 'pg_shardman' LANGUAGE C STRICT;
 -- prepared with subsequent commit or rollback of prepared transaction at second
 -- phase of two phase commit.
 -- If sync_commit is false, we do set session synchronous_commit to local;
-CREATE FUNCTION broadcast(cmds text, ignore_errors bool = false,
-						  two_phase bool = false, sync_commit bool = false)
+CREATE FUNCTION broadcast(cmds text,
+						  ignore_errors bool = false,
+						  two_phase bool = false,
+						  sync_commit bool = false,
+						  synchronous bool = false)
 RETURNS text AS 'pg_shardman' LANGUAGE C STRICT;
 
 -- Options to postgres_fdw are specified in two places: user & password in user

pg_shardman.c

@@ -11,12 +11,14 @@
 #include "miscadmin.h"
 #include "executor/spi.h"
 #include "funcapi.h"
+#include "pgstat.h"
 #include "utils/guc.h"
 #include "utils/rel.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "catalog/pg_type.h"
 #include "access/htup_details.h"
+#include "storage/latch.h"
 
 /* ensure that extension won't load against incompatible version of Postgres */
 PG_MODULE_MAGIC;
@@ -28,6 +30,7 @@ PG_FUNCTION_INFO_V1(reconstruct_table_attrs);
 PG_FUNCTION_INFO_V1(pq_conninfo_parse);
 
 /* GUC variables */
+static bool is_shardlord;
 static bool sync_replication;
 static char *shardlord_connstring;
 
@@ -49,6 +52,16 @@ _PG_init()
 		0,
 		NULL, NULL, NULL);
 
+	DefineCustomBoolVariable(
+		"shardman.shardlord",
+		"This node is the shardlord?",
+		NULL,
+		&is_shardlord,
+		false,
+		PGC_SUSET,
+		0,
+		NULL, NULL, NULL);
+
 	DefineCustomStringVariable(
 		"shardman.shardlord_connstring",
 		"Active only if shardman.shardlord is on. Connstring to reach shardlord from"
@@ -73,13 +86,38 @@ synchronous_replication(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(sync_replication);
 }
 
+static bool
+wait_command_completion(PGconn* conn)
+{
+	while (PQisBusy(conn))
+	{
+		/* Sleep until there's something to do */
+		int wc = WaitLatchOrSocket(MyLatch,
+								   WL_LATCH_SET | WL_SOCKET_READABLE,
+								   PQsocket(conn),
+								   -1L, PG_WAIT_EXTENSION);
+		ResetLatch(MyLatch);
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* Data available in socket? */
+		if (wc & WL_SOCKET_READABLE)
+		{
+			if (!PQconsumeInput(conn))
+				return false;
+		}
+	}
+	return  true;
+}
+
 Datum
 broadcast(PG_FUNCTION_ARGS)
 {
 	char* sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	bool  ignore_errors = PG_GETARG_BOOL(1);
 	bool  two_phase = PG_GETARG_BOOL(2);
 	bool  sync_commit = PG_GETARG_BOOL(3);
+	bool  sequential = PG_GETARG_BOOL(4);
 	char* sep;
 	PGresult *res;
 	char* fetch_node_connstr;
@@ -113,16 +151,22 @@ broadcast(PG_FUNCTION_ARGS)
 			elog(ERROR, "SHARDMAN: Invalid command string: %s", sql);
 		}
 		sql += n;
-		fetch_node_connstr = psprintf("select connection_string from shardman.nodes where id=%d", node_id);
-		if (SPI_exec(fetch_node_connstr, 0) < 0 || SPI_processed != 1)
+		if (node_id != 0)
 		{
-			elog(ERROR, "SHARDMAN: Failed to fetch connection string for node %d",
-				 node_id);
-		}
-		pfree(fetch_node_connstr);
-
-		conn_str = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
+			fetch_node_connstr = psprintf("select connection_string from shardman.nodes where id=%d", node_id);
+			if (SPI_exec(fetch_node_connstr, 0) < 0 || SPI_processed != 1)
+			{
+				elog(ERROR, "SHARDMAN: Failed to fetch connection string for node %d",
+					 node_id);
+			}
+			pfree(fetch_node_connstr);
 
+			conn_str = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
+		}
+		else
+		{
+			conn_str = shardlord_connstring;
+		}
 		if (n_cmds >= n_cons)
 		{
 			conn = (PGconn**) repalloc(conn, sizeof(PGconn*) * (n_cons *= 2));
@@ -154,7 +198,8 @@ broadcast(PG_FUNCTION_ARGS)
 			}
 		}
 		elog(DEBUG1, "Send command '%s' to node %d", sql, node_id);
-		if (!PQsendQuery(conn[n_cmds - 1], sql))
+		if (!PQsendQuery(conn[n_cmds - 1], sql)
+			|| (sequential && !wait_command_completion(conn[n_cmds - 1])))
 		{
 			if (ignore_errors)
 			{
@@ -169,6 +214,7 @@ broadcast(PG_FUNCTION_ARGS)
 		}
 		if (!sync_commit)
 			pfree(sql);
+
 		sql = sep + 1;
 	}