Enabling/disabling synchronous replication.

But now we need to rewrite things, because enabling has to be performed in
separate transaction to avoid hangup. Also, get_node_id renamed to my_id.
Also, I have found func for text -> cstring conversion, yeah.

Author: arssher

init.sql

@@ -141,7 +141,7 @@ BEGIN
 	raise INFO 'Booting master';
 	PERFORM shardman.create_meta_pub();
 
-	master_id := shardman.get_node_id();
+	master_id := shardman.my_id();
 	IF master_id IS NULL THEN
 		SELECT pg_settings.setting into master_connstring from pg_settings
 			WHERE NAME = 'shardman.master_connstring';

membership.sql

@@ -31,7 +31,7 @@ CREATE TABLE nodes (
 -- of transaction. Later we should delete subscriptions fully.
 CREATE FUNCTION rm_node_worker_side() RETURNS TRIGGER AS $$
 BEGIN
-	IF OLD.id = (SELECT shardman.get_node_id()) THEN
+	IF OLD.id = (SELECT shardman.my_id()) THEN
 		RAISE DEBUG '[SHARDMAN] rm_node_worker_side called';
 		PERFORM shardman.pg_shardman_cleanup(false);
 	END IF;
@@ -53,7 +53,7 @@ CREATE TABLE local_meta (
 INSERT INTO shardman.local_meta VALUES ('node_id', NULL);
 
 -- Get local node id. NULL means node is not in the cluster yet.
-CREATE FUNCTION get_node_id() RETURNS int AS $$
+CREATE FUNCTION my_id() RETURNS int AS $$
 	SELECT v::int FROM shardman.local_meta WHERE k = 'node_id';
 $$ LANGUAGE sql;
 
@@ -95,9 +95,20 @@ BEGIN
 END
 $$ LANGUAGE plpgsql;
 
+-- Get local node connstr. Throws an error, if node is not in cluster
+CREATE FUNCTION my_connstr() RETURNS text AS $$
+DECLARE
+	connstr text := connstring FROM shardman.nodes WHERE id = shardman.my_id();
+BEGIN
+	IF connstr IS NULL THEN
+		RAISE EXCEPTION 'Node not in cluster, can''t get its connstring';
+	END IF;
+	RETURN connstr;
+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 $$
+CREATE 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

shard.sql

@@ -29,7 +29,7 @@ DECLARE
 	partition_names text[];
 	pname text;
 BEGIN
-	IF NEW.initial_node != (SELECT shardman.get_node_id()) THEN
+	IF NEW.initial_node != (SELECT shardman.my_id()) THEN
 		EXECUTE format('DROP TABLE IF EXISTS %I CASCADE;', NEW.relation);
 		partition_names :=
 			(SELECT ARRAY(SELECT part_name FROM shardman.gen_part_names(
@@ -94,8 +94,8 @@ CREATE TABLE partitions (
 CREATE FUNCTION new_primary() RETURNS TRIGGER AS $$
 BEGIN
 	RAISE DEBUG '[SHARDMAN] new_primary trigger called on node % for part %, owner %',
-		shardman.get_node_id(), NEW.part_name, NEW.owner;
-	IF NEW.owner != shardman.get_node_id() THEN
+		shardman.my_id(), NEW.part_name, NEW.owner;
+	IF NEW.owner != shardman.my_id() THEN
 		PERFORM shardman.replace_usual_part_with_foreign(NEW);
 	END IF;
 	RETURN NULL;
@@ -110,7 +110,7 @@ ALTER TABLE shardman.partitions ENABLE REPLICA TRIGGER new_primary;
 CREATE FUNCTION part_moved_prev(p_name name, src int, dst int)
 	RETURNS void AS $$
 DECLARE
-	me int := shardman.get_node_id();
+	me int := shardman.my_id();
 	lname text := shardman.get_data_lname(p_name, me, dst);
 BEGIN
 	PERFORM shardman.create_repslot(lname);
@@ -132,7 +132,7 @@ DECLARE
 	prev_lname text;
 	prev_connstr text;
 BEGIN
-	ASSERT dst = shardman.get_node_id(), 'part_moved_dst must be called on dst';
+	ASSERT dst = shardman.my_id(), 'part_moved_dst must be called on dst';
 	IF next_rep IS NOT NULL THEN -- we need to setup channel dst -> next replica
 		next_lname := shardman.get_data_lname(p_name, dst, next_rep);
 		-- This must be first write in the transaction!
@@ -159,7 +159,7 @@ END $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION part_moved_next(p_name name, src int, dst int)
 	RETURNS void AS $$
 DECLARE
-	me int := shardman.get_node_id();
+	me int := shardman.my_id();
 	lname text := shardman.get_data_lname(p_name, dst, me);
 	dst_connstr text := shardman.get_worker_node_connstr(dst);
 BEGIN
@@ -176,7 +176,7 @@ END $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION part_moved() RETURNS TRIGGER AS $$
 DECLARE
 	cp_logname text := shardman.get_cp_logname(NEW.part_name, OLD.owner, NEW.owner);
-	me int := shardman.get_node_id();
+	me int := shardman.my_id();
 	prev_src_lname text;
 	src_next_lname text;
 BEGIN
@@ -297,7 +297,7 @@ END $$ LANGUAGE plpgsql;
 -- CREATE FUNCTION replica_created_update_fdw() RETURNS TRIGGER AS $$
 -- BEGIN
 	-- RAISE DEBUG '[SHARDMAN] replica_created_update_fdw trigger called';
-	-- IF shardman.get_node_id() != NEW.prv THEN -- don't update on oldtail node
+	-- IF shardman.my_id() != NEW.prv THEN -- don't update on oldtail node
 		-- PERFORM shardman.update_fdw_server(NEW);
 	-- END IF;
 	-- RETURN NULL;
@@ -372,11 +372,11 @@ DECLARE
 	connstring text;
 	server_opts text;
 	um_opts text;
-	me int := shardman.get_node_id();
+	me int := shardman.my_id();
 	my_part shardman.partitions;
 BEGIN
 	RAISE DEBUG '[SHARDMAN %] update_fdw_server called for part %, owner %',
-		shardman.get_node_id(), part.part_name, part.owner;
+		shardman.my_id(), part.part_name, part.owner;
 
 	SELECT * FROM shardman.partitions WHERE part_name = part.part_name AND
 											owner = me INTO my_part;
@@ -451,7 +451,7 @@ BEGIN
 	-- change reached us? We might also use it with local table (create
 	-- foreign server pointing to it, etc), but that's just ugly.
 	RAISE DEBUG '[SHARDMAN] my id: %, creating ft %',
-		shardman.get_node_id(), part.part_name;
+		shardman.my_id(), part.part_name;
 	EXECUTE format('CREATE FOREIGN TABLE %I %s SERVER %I OPTIONS (table_name %L)',
 				   fdw_part_name,
 				   (SELECT
@@ -628,7 +628,7 @@ BEGIN
 END $$ LANGUAGE plpgsql STRICT;
 
 CREATE FUNCTION gen_create_table_sql(relation text, connstring text) RETURNS text
-    AS 'pg_shardman' LANGUAGE C;
+    AS 'pg_shardman' LANGUAGE C STRICT;
 
 CREATE FUNCTION reconstruct_table_attrs(relation regclass)
 	RETURNS text AS 'pg_shardman' LANGUAGE C STRICT;
@@ -670,13 +670,32 @@ 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(standby text) RETURNS void as $$
+DECLARE
+	newval text := shardman.ensure_sync_standby_c(standby);
 BEGIN
-	RAISE DEBUG '[SHARDMAN] imagine standby % added', standby;
+	IF newval IS NOT NULL THEN
+		RAISE DEBUG '[SHARDMAN] Adding standby %, new value is %', standby, newval;
+		PERFORM shardman.set_sync_standbys(newval);
+		PERFORM pg_reload_conf();
+	END IF;
 END $$ LANGUAGE plpgsql STRICT;
+CREATE FUNCTION ensure_sync_standby_c(standby text) RETURNS text
+    AS 'pg_shardman' LANGUAGE C;
 
 -- Remove 'standby' from in synchronous_standby_names, if it is there, and SIGHUP
 -- postmaster.
 CREATE FUNCTION remove_sync_standby(standby text) RETURNS void as $$
+DECLARE
+	newval text := shardman.remove_sync_standby_c(standby);
 BEGIN
-	RAISE DEBUG '[SHARDMAN] imagine standby % removed', standby;
+	IF newval IS NOT NULL THEN
+		RAISE DEBUG '[SHARDMAN] Removing standby %, new value is %', standby, newval;
+		PERFORM shardman.set_sync_standbys(newval);
+		PERFORM pg_reload_conf();
+	END IF;
 END $$ LANGUAGE plpgsql STRICT;
+CREATE FUNCTION remove_sync_standby_c(standby text) RETURNS text
+	AS 'pg_shardman' LANGUAGE C;
+
+CREATE FUNCTION set_sync_standbys(standby text) RETURNS void
+	AS 'pg_shardman' LANGUAGE C;

src/pg_shardman.c

@@ -504,7 +504,7 @@ add_node(Cmd *cmd)
 		{
 			/* extension is installed, so we have to check whether this node
 			 * is already in the cluster */
-			res = PQexec(conn, "select shardman.get_node_id();");
+			res = PQexec(conn, "select shardman.my_id();");
 			if (PQresultStatus(res) != PGRES_TUPLES_OK)
 			{
 				shmn_elog(NOTICE, "Failed to get node id, %s", PQerrorMessage(conn));

src/udf.c

@@ -18,6 +18,7 @@
 #include "catalog/pg_type.h"
 #include "storage/lmgr.h"
 #include "replication/logicalworker.h"
+#include "replication/syncrep.h"
 #include "libpq-fe.h"
 
 #include "pg_shardman.h"
@@ -138,8 +139,6 @@ reconstruct_table_attrs(PG_FUNCTION_ARGS)
 	Relation local_rel = heap_open(relid, AccessExclusiveLock);
 	TupleDesc local_descr = RelationGetDescr(local_rel);
 	int i;
-	text *text_query;
-	int32 text_query_size;
 
 	initStringInfo(&query);
 	appendStringInfoChar(&query, '(');
@@ -165,15 +164,9 @@ reconstruct_table_attrs(PG_FUNCTION_ARGS)
 
 	appendStringInfoChar(&query, ')');
 
-	/* Now prepare result as 'text' */
-	text_query_size = VARHDRSZ + query.len;
-	text_query = (text *) palloc(text_query_size);
-	SET_VARSIZE(text_query, text_query_size);
-	memcpy(VARDATA(text_query), query.data, query.len);
-
 	/* Let xact unlock this */
 	heap_close(local_rel, NoLock);
-	PG_RETURN_TEXT_P(text_query);
+	PG_RETURN_TEXT_P(cstring_to_text(query.data));
 }
 
 /*
@@ -297,9 +290,152 @@ reset_node_id_c(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/* Are we a logical apply worker? */
 PG_FUNCTION_INFO_V1(inside_apply_worker);
 Datum
 inside_apply_worker(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_BOOL(IsLogicalWorker());
 }
+
+/*
+ * Check whether 'standby' is present in current value of
+ * synchronous_standby_names. If yes, return NULL. Otherwise, form properly
+ * quoted new value of the setting with 'standby' appended. Currently we
+ * support only the case when *all* standbys must agree on commit, so FIRST or
+ * ANY doesn't matter here. '*' wildcard is not supported too.
+ */
+PG_FUNCTION_INFO_V1(ensure_sync_standby_c);
+Datum
+ensure_sync_standby_c(PG_FUNCTION_ARGS)
+{
+	char *standby = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	char *cur_standby_name;
+	StringInfoData standby_list;
+	char *newval;
+	int processed = 0;
+
+	initStringInfo(&standby_list);
+	if (SyncRepConfig != NULL)
+	{
+		Assert(SyncRepConfig->num_sync == SyncRepConfig->nmembers);
+
+		cur_standby_name = SyncRepConfig->member_names;
+		for (; processed < SyncRepConfig->nmembers; processed++)
+		{
+			Assert(strcmp(cur_standby_name, "*") != 0);
+			if (pg_strcasecmp(cur_standby_name, standby) == 0)
+			{
+				PG_RETURN_NULL(); /* already set */
+			}
+
+			if (processed != 0)
+				appendStringInfoString(&standby_list, ", ");
+			appendStringInfoString(&standby_list, quote_identifier(cur_standby_name));
+
+			cur_standby_name += strlen(cur_standby_name) + 1;
+		}
+	}
+
+	if (processed != 0)
+		appendStringInfoString(&standby_list, ", ");
+	appendStringInfoString(&standby_list, quote_identifier(standby));
+
+	newval = psprintf("FIRST %d (%s)", processed + 1, standby_list.data);
+	PG_RETURN_TEXT_P(cstring_to_text(newval));
+}
+
+/*
+ * Check whether 'standby' is present in current value of
+ * synchronous_standby_names. If no, return NULL. Otherwise, form properly
+ * quoted new value of the setting with 'standby' removed. Currently we
+ * support only the case when *all* standbys must agree on commit, so FIRST or
+ * ANY doesn't matter here. '*' wildcard is not supported too.
+ * All entries are removed.
+ */
+PG_FUNCTION_INFO_V1(remove_sync_standby_c);
+Datum
+remove_sync_standby_c(PG_FUNCTION_ARGS)
+{
+	char *standby = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	char *cur_standby_name;
+	StringInfoData standby_list;
+	char *newval;
+	int processed;
+	int num_sync;
+
+	if (SyncRepConfig == NULL)
+		PG_RETURN_NULL();
+
+	initStringInfo(&standby_list);
+	Assert(SyncRepConfig->num_sync == SyncRepConfig->nmembers);
+
+	cur_standby_name = SyncRepConfig->member_names;
+	num_sync = 0;
+	for (processed = 0; processed < SyncRepConfig->nmembers; processed++)
+	{
+		Assert(strcmp(cur_standby_name, "*") != 0);
+		if (pg_strcasecmp(cur_standby_name, standby) != 0)
+		{
+			if (num_sync != 0)
+				appendStringInfoString(&standby_list, ", ");
+			appendStringInfoString(&standby_list, quote_identifier(cur_standby_name));
+			num_sync++;
+		}
+
+		cur_standby_name += strlen(cur_standby_name) + 1;
+	}
+
+	if (SyncRepConfig->num_sync == num_sync)
+		PG_RETURN_NULL(); /* nothing was removed */
+
+	if (num_sync > 0)
+		newval = psprintf("FIRST %d (%s)", num_sync, standby_list.data);
+	else
+		newval = "";
+	PG_RETURN_TEXT_P(cstring_to_text(newval));
+}
+
+/*
+ * Execute ALTER SYSTEM SET synchronous_standby_names TO 'arg'. We can't do
+ * that from usual function, because ALTER SYSTEM cannon be executed within
+ * transaction, so we resort to another exquisite hack: we connect to
+ * ourselves via libpq and perform the job.
+ */
+PG_FUNCTION_INFO_V1(set_sync_standbys);
+Datum
+set_sync_standbys(PG_FUNCTION_ARGS)
+{
+	char *standbys = quote_literal_cstr(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+	char *cmd = psprintf("alter system set synchronous_standby_names to %s",
+						 standbys);
+	char *my_id_sql = "select shardman.my_connstr();";
+	char *connstr;
+	PGconn *conn = NULL;
+	PGresult *res = NULL;
+
+	SPI_connect();
+	if (SPI_execute(my_id_sql, true, 0) < 0 || SPI_processed != 1)
+		elog(FATAL, "Stmt failed: %s", my_id_sql);
+	connstr = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
+
+	conn = PQconnectdb(connstr);
+	if (PQstatus(conn) != CONNECTION_OK)
+	{
+		PQfinish(conn);
+		elog(ERROR, "Connection to myself with connstr %s failed", connstr);
+
+	}
+	res = PQexec(conn, cmd);
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		PQclear(res);
+		PQfinish(conn);
+		elog(ERROR, "setting sync standby namesfailed");
+	}
+
+	PQclear(res);
+	PQfinish(conn);
+	SPI_finish(); /* Can't do earlier since connstr is allocated there */
+	PG_RETURN_VOID();
+}