Fix bugs in create_shared_table

Author: knizhnik

pg_shardman--1.0.sql

@@ -26,7 +26,7 @@ CREATE TABLE tables (
 	relation text PRIMARY KEY,     -- table name
 	sharding_key text,             -- expression by which table is sharded
 	master_node integer REFERENCES nodes(id) ON DELETE CASCADE,
-	partitions_count int NOT NULL, -- number of partitions
+	partitions_count int,          -- number of partitions
 	create_sql text NOT NULL,      -- sql to create the table
 	create_rules_sql text          -- sql to create rules for shared table
 );
@@ -85,6 +85,7 @@ DECLARE
 	table_attrs text;
 	srv_name text;
 	rules text = '';
+	master_node_id int;
 conn_string_effective text = COALESCE(conn_string, super_conn_string);
 BEGIN
 	IF shardman.redirect_to_shardlord(
@@ -194,8 +195,7 @@ BEGIN
 		END LOOP;
 	END LOOP;
 
-	-- Create subscriptions for all shared tables
-	subs := '';
+	-- Create at new node FDWs for all shared tables
 	FOR t IN SELECT * from shardman.tables WHERE master_node IS NOT NULL
 	LOOP
 		SELECT connection_string INTO conn_string from shardman.nodes WHERE id=t.master_node;
@@ -206,24 +206,30 @@ BEGIN
 		fdw_part_name := format('%s_fdw', t.relation);
 		create_fdws := format('%s%s:CREATE FOREIGN TABLE %I %s SERVER %s OPTIONS (table_name %L);',
 			create_fdws, new_node_id, fdw_part_name, table_attrs, srv_name, t.relation);
-		subs := format('%s%s:CREATE SUBSCRIPTION share_%s CONNECTION %L PUBLICATION share_%s with (create_slot=false, copy_data=false, slot_name=''share_%s'', synchronous_commit=local);',
-			 subs, new_node_id, t.relation, conn_string, t.relation, t.relation);
 		create_rules :=  format('%s{%s:%s}',
 			 create_rules, new_node_id, t.create_rules_sql);
 	END LOOP;
 
-	-- Broadcast create table commands
+	-- Create subscriptions for all shared tables
+	subs := '';
+	FOR master_node_id IN SELECT DISTINCT master_node FROM shardman.tables WHERE master_node IS NOT NULL
+	LOOP
+		subs := format('%s%s:CREATE SUBSCRIPTION share_%s_%s CONNECTION %L PUBLICATION shared_tables with (synchronous_commit=local);',
+			 subs, new_node_id, new_node_id, t.master_node, conn_string);
+	END LOOP;
+
+    -- Broadcast create table commands
 	PERFORM shardman.broadcast(create_tables);
 	-- Broadcast create hash partitions command
 	PERFORM shardman.broadcast(create_partitions);
 	-- Broadcast create foreign table commands
 	PERFORM shardman.broadcast(create_fdws);
 	-- Broadcast replace hash partition commands
 	PERFORM shardman.broadcast(replace_parts);
-	-- Broadcast create subscriptions for shared tables
-	PERFORM shardman.broadcast(subs);
 	-- Broadcast create rules for shared tables
 	PERFORM shardman.broadcast(create_rules);
+	-- Broadcast create subscriptions for shared tables
+	PERFORM shardman.broadcast(subs, super_connstr => true);
 END
 $$ LANGUAGE plpgsql;
 
@@ -236,7 +242,6 @@ DECLARE
 	node shardman.nodes;
 	part shardman.partitions;
 	repl shardman.replicas;
-	t shardman.tables;
 	pubs text = '';
 	subs text = '';
 	fdws text = '';
@@ -246,6 +251,7 @@ DECLARE
     new_master_id int;
 	sync_standbys text[];
 	repl_group text;
+	master_node_id int;
 BEGIN
 	IF shardman.redirect_to_shardlord(format('rm_node(%L, %L)', rm_node_id, force))
 	THEN
@@ -293,10 +299,10 @@ BEGIN
 	END LOOP;
 
 	-- Drop shared tables subscriptions
-	FOR t IN SELECT * from shardman.tables WHERE master_node IS NOT NULL
+	FOR master_node_id IN SELECT DISTINCT master_node from shardman.tables WHERE master_node IS NOT NULL
 	LOOP
-		subs := format('%s%s:DROP SUBSCRIPTION share_%s;',
-			 subs, rm_node_id, t.relation);
+		subs := format('%s%s:DROP SUBSCRIPTION share_%s_%s;',
+			 subs, rm_node_id, rm_node_id, master_node_id);
 	END LOOP;
 
 	-- Broadcast drop subscription commands, ignore errors because removed node may be not available
@@ -798,9 +804,10 @@ $$ LANGUAGE plpgsql;
 
 -- Share table between all nodes. This function should be executed at shardlord. The empty table should be present at shardlord,
 -- but not at nodes.
-CREATE FUNCTION create_shared_table(rel regclass, master_node_id int) RETURNS void AS $$
+CREATE FUNCTION create_shared_table(rel regclass, master_node_id int = 1) RETURNS void AS $$
 DECLARE
 	node shardman.nodes;
+	pubs text = '';
 	subs text = '';
 	fdws text = '';
 	rules text = '';
@@ -815,40 +822,14 @@ DECLARE
 	pk text;
 	dst text;
 	src text;
+	new_master bool;
 BEGIN
-	-- Check if valid node ID is passed and get connectoin string of this node
+	-- Check if valid node ID is passed and get connectoin string for this node
 	SELECT connection_string INTO conn_string FROM shardman.nodes WHERE id=master_node_id;
 	IF conn_string IS NULL THEN
-	    RAISE EXCEPTION 'There is no node with ID % in the cluster',master_node_id;
+	    RAISE EXCEPTION 'There is no node with ID % in the cluster', master_node_id;
 	END IF;
 
-	-- Generate SQL statement creating this table
-	SELECT shardman.gen_create_table_sql(rel_name) INTO create_table;
-
-	create_rules := format('CREATE RULE on_update_to_%s AS ON UPDATE TO %I DO INSTEAD UPDATE %I SET (%s) = (%s) WHERE %s;
-		 	                CREATE RULE on_insert_to_%s AS ON INSERT TO %I DO INSTEAD INSERT INTO %I (%s) VALUES (%s);
-		 	                CREATE RULE on_delete_from_%s AS ON DELETE TO %I DO INSTEAD DELETE FROM %I WHERE %s;',
-        rel_name, rel_name, fdw_rel_name, dst, src, pk,
-		rel_name, rel_name, fdw_rel_name, dst, src,
-		rel_name, rel_name, fdw_rel_name, pk);
-
-	INSERT INTO shardman.tables (relation,master_node,create_sql,create_rules_sql) values (rel_name,master_node_id,create_table,create_rules);
-
-	FOR node IN SELECT * FROM shardman.nodes
-	LOOP
-		create_tables := format('%s{%s:%s}',
-			create_tables, node.id, create_table);
-	END LOOP;
-
-	-- Broadcast create table command
-	PERFORM shardman.broadcast(create_tables);
-
-	-- Create publication at master node
-	PERFORM shardman.broadcast(format('%s:CREATE PUBLICATION share_%s FOR TABLE %I;
-		 							   %s:SELECT pg_create_logical_replication_slot(''share_%s'', ''pgoutput'');',
-		master_node_id, rel_name, rel_name,
-		master_node_id, rel_name));
-
 	-- Construct list of attributes of the table for update/insert
 	SELECT INTO dst, src
           string_agg(quote_ident(attname), ', '),
@@ -860,29 +841,71 @@ BEGIN
 
 	-- Construct primary key condition for update
 	SELECT INTO pk
-          string_agg(quote_ident(a.attname) || '=NEW.'|| quote_ident(a.attname), ' AND ')
+          string_agg(quote_ident(a.attname) || '=OLD.'|| quote_ident(a.attname), ' AND ')
 	FROM   pg_index i
 	JOIN   pg_attribute a ON a.attrelid = i.indrelid
                      AND a.attnum = ANY(i.indkey)
     WHERE  i.indrelid = rel
     AND    i.indisprimary;
 
+	-- Generate SQL statement creating this table
+	SELECT shardman.gen_create_table_sql(rel_name) INTO create_table;
+
 	-- Construct table attributes for create foreign table
 	SELECT shardman.reconstruct_table_attrs(rel) INTO table_attrs;
 
+	-- Create rules for redirecting updates
+	create_rules := format('CREATE RULE on_update_to_%s AS ON UPDATE TO %I DO INSTEAD UPDATE %I SET (%s) = (%s) WHERE %s;
+		 	                CREATE RULE on_insert_to_%s AS ON INSERT TO %I DO INSTEAD INSERT INTO %I (%s) VALUES (%s);
+		 	                CREATE RULE on_delete_from_%s AS ON DELETE TO %I DO INSTEAD DELETE FROM %I WHERE %s;',
+        rel_name, rel_name, fdw_name, dst, src, pk,
+		rel_name, rel_name, fdw_name, dst, src,
+		rel_name, rel_name, fdw_name, pk);
+
+	-- Create table at all nodes
+	FOR node IN SELECT * FROM shardman.nodes
+	LOOP
+		create_tables := format('%s{%s:%s}',
+			create_tables, node.id, create_table);
+	END LOOP;
+
+	-- Create publication at master node
+	IF EXISTS(SELECT * from shardman.tables WHERE master_node=master_node_id)
+	THEN
+		new_master := false;
+		pubs := format('%s:ALTER PUBLICATION shared_tables ADD TABLE %I;',
+			 master_node_id, rel_name);
+	ELSE
+		new_master := true;
+		pubs := format('%s:CREATE PUBLICATION shared_tables FOR TABLE %I;',
+	         master_node_id, rel_name);
+	END IF;
+
+	-- Insert information about new table in shardman.tables
+	INSERT INTO shardman.tables (relation,master_node,create_sql,create_rules_sql) values (rel_name,master_node_id,create_table,create_rules);
+
 	-- Create subscriptions, foreign tables and rules at all nodes
 	FOR node IN SELECT * FROM shardman.nodes WHERE id<>master_node_id
 	LOOP
-		subs := format('%s%s:CREATE SUBSCRIPTION share_%s CONNECTION %L PUBLICATION share_%s with (create_slot=false, copy_data=false, slot_name=''share_%s'', synchronous_commit=local);',
-			 subs, node.id, rel_name, conn_string, rel_name, rel_name);
+		IF new_master THEN
+		    subs := format('%s%s:CREATE SUBSCRIPTION share_%s_%s CONNECTION %L PUBLICATION shared_tables WITH (copy_data=false, synchronous_commit=local);',
+				 subs, node.id, node.id, master_node_id, conn_string);
+		ELSE
+			subs := format('%s%s:ALTER SUBSCRIPTION share_%s_%s REFRESH PUBLICATIONS WITH (copy_data=false);',
+				 subs, node.id, node.id, master_node_id);
+		END IF;
 		fdws := format('%s%s:CREATE FOREIGN TABLE %I %s SERVER %s OPTIONS (table_name %L);',
-			 fdws, node.id, fdw_rel_name, table_attrs, srv_name, rel_name);
+			 fdws, node.id, fdw_name, table_attrs, srv_name, rel_name);
 		rules := format('%s{%s:%s}',
 			 rules, node.id, create_rules);
 	END LOOP;
 
+	-- Broadcast create table command
+	PERFORM shardman.broadcast(create_tables);
+	-- Create or alter publication at master node
+	PERFORM shardman.broadcast(pubs);
 	-- Create subscriptions at all nodes
-	PERFORM shardman.broadcast(subs);
+	PERFORM shardman.broadcast(subs, sync_commit_on => true, super_connstr => true);
 	-- Create foreign tables at all nodes
 	PERFORM shardman.broadcast(fdws);
 	-- Create redirect rules at all nodes