Merge commit '0421001d2d784ae4e6100a689c516cda8b177209' into PGPROEE10_pg_shardman

Author: arssher

contrib/pg_shardman/conf.add

@@ -1 +1 @@
-shared_preload_libraries='pg_pathman'
+shared_preload_libraries='pg_shardman, pg_pathman'

contrib/pg_shardman/pg_shardman--0.0.2.sql

@@ -7,19 +7,22 @@
  * -------------------------------------------------------------------------
  */
 #include "postgres.h"
-#include "libpq-fe.h"
-#include "miscadmin.h"
+
+#include "access/htup_details.h"
+#include "access/xact.h"
+#include "access/xlog.h"
+#include "catalog/pg_type.h"
+#include "commands/event_trigger.h"
 #include "executor/spi.h"
 #include "funcapi.h"
+#include "libpq-fe.h"
+#include "miscadmin.h"
 #include "pgstat.h"
+#include "storage/latch.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 "access/xlog.h"
-#include "storage/latch.h"
 
 /* ensure that extension won't load against incompatible version of Postgres */
 PG_MODULE_MAGIC;
@@ -31,6 +34,7 @@ PG_FUNCTION_INFO_V1(broadcast);
 PG_FUNCTION_INFO_V1(reconstruct_table_attrs);
 PG_FUNCTION_INFO_V1(pq_conninfo_parse);
 PG_FUNCTION_INFO_V1(get_system_identifier);
+PG_FUNCTION_INFO_V1(reset_synchronous_standby_names_on_commit);
 
 /* GUC variables */
 static bool is_lord;
@@ -39,6 +43,11 @@ static char *shardlord_connstring;
 
 extern void _PG_init(void);
 
+static bool reset_ssn_callback_set = false;
+static bool reset_ssn_requested = false;
+
+static void reset_ssn_xact_callback(XactEvent event, void *arg);
+
 /*
  * Entrypoint of the module. Define GUCs.
  */
@@ -75,6 +84,15 @@ _PG_init()
 		PGC_SUSET,
 		0,
 		NULL, NULL, NULL);
+
+	/*
+	 * Tell pathman that we want it to do shardman-specific COPY FROM: that
+	 * is, support copy to foreign partitions by copying to foreign parent.
+	 * For now we just ask to do it always. Better to turn on this in copy
+	 * hook turn off after, however for that we need metadata on all nodes.
+	 */
+	*find_rendezvous_variable(
+		"shardman_pathman_copy_from_rendezvous") = DatumGetPointer(1);
 }
 
 Datum
@@ -95,6 +113,10 @@ is_shardlord(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(is_lord);
 }
 
+/*
+ * Wait until PQgetResult would certainly be non-blocking. Returns true if
+ * everything is ok, false on error.
+ */
 static bool
 wait_command_completion(PGconn* conn)
 {
@@ -119,7 +141,7 @@ wait_command_completion(PGconn* conn)
 				return false;
 		}
 	}
-	return  true;
+	return true;
 }
 
 typedef struct
@@ -139,22 +161,25 @@ broadcast(PG_FUNCTION_ARGS)
 	bool  sync_commit_on = PG_GETARG_BOOL(3);
 	bool  sequential = PG_GETARG_BOOL(4);
 	bool  super_connstr = PG_GETARG_BOOL(5);
+	char* iso_level = (PG_GETARG_POINTER(6) != NULL) ?
+		text_to_cstring(PG_GETARG_TEXT_PP(6)) : NULL;
 	char* sep;
 	char* sql;
 	PGresult *res;
 	char* fetch_node_connstr;
 	int   rc;
 	int   node_id;
-	int   n;
+	int	  n;
 	char* conn_str;
 	int   n_cmds = 0;
 	int   i;
-	int n_cons = 1024;
+	int n_cons = 1024; /* num of channels allocated currently */
 	Channel* chan;
 	PGconn* con;
 	StringInfoData resp;
+	StringInfoData fin_sql;
 
-	char const* errmsg = "";
+	char const* errstr = "";
 
 	elog(DEBUG1, "Broadcast commmand '%s'",  cmd);
 
@@ -163,6 +188,7 @@ broadcast(PG_FUNCTION_ARGS)
 	SPI_connect();
 	chan = (Channel*) palloc(sizeof(Channel) * n_cons);
 
+	/* Open connections and send all queries */
 	while ((sep = strchr(cmd, *cmd == '{' ? '}' : ';')) != NULL)
 	{
 		*sep = '\0';
@@ -171,9 +197,10 @@ broadcast(PG_FUNCTION_ARGS)
 			cmd += 1;
 		rc = sscanf(cmd, "%d:%n", &node_id, &n);
 		if (rc != 1) {
-			elog(ERROR, "SHARDMAN: Invalid command string: '%s' in '%s'", cmd, sql_full);
+			elog(ERROR, "SHARDMAN: Invalid command string: '%s' in '%s'",
+				 cmd, sql_full);
 		}
-		sql = cmd + n;
+		sql = cmd + n; /* eat node id and colon */
 		cmd = sep + 1;
 		if (node_id != 0)
 		{
@@ -187,7 +214,8 @@ broadcast(PG_FUNCTION_ARGS)
 			}
 			pfree(fetch_node_connstr);
 
-			conn_str = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);
+			conn_str = SPI_getvalue(SPI_tuptable->vals[0],
+									SPI_tuptable->tupdesc, 1);
 		}
 		else
 		{
@@ -212,39 +240,40 @@ broadcast(PG_FUNCTION_ARGS)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s<error>%d:Connection failure: %s</error>",
-								  errmsg, node_id, PQerrorMessage(con));
+				errstr = psprintf("%s<error>%d:Connection failure: %s</error>",
+								  errstr, node_id, PQerrorMessage(con));
 				chan[n_cmds-1].sql = NULL;
 				continue;
 			}
-			errmsg = psprintf("Failed to connect to node %d: %s", node_id,
+			errstr = psprintf("Failed to connect to node %d: %s", node_id,
 							  PQerrorMessage(con));
 			goto cleanup;
 		}
+		/* Build the actual sql to send, mem freed with ctxt */
+		initStringInfo(&fin_sql);
 		if (!sync_commit_on)
-		{
-			/* mem freed with context */
-			if (two_phase)
-			{
-				sql = psprintf("SET SESSION synchronous_commit TO local; BEGIN; %s; PREPARE TRANSACTION 'shardlord';", sql);
-			}
-			else
-			{
-				sql = psprintf("SET SESSION synchronous_commit TO local; %s", sql);
-			}
-		}
-		elog(DEBUG1, "Send command '%s' to node %d", sql, node_id);
-		if (!PQsendQuery(con, sql)
+			appendStringInfoString(&fin_sql, "SET SESSION synchronous_commit TO local; ");
+		if (iso_level)
+			appendStringInfo(&fin_sql, "BEGIN TRANSACTION ISOLATION LEVEL %s; ", iso_level);
+		appendStringInfoString(&fin_sql, sql);
+		appendStringInfoChar(&fin_sql, ';'); /* it was removed after strchr */
+		if (two_phase)
+			appendStringInfoString(&fin_sql, "PREPARE TRANSACTION 'shardlord';");
+		else if (iso_level)
+			appendStringInfoString(&fin_sql, "END;");
+
+		elog(DEBUG1, "Sending command '%s' to node %d", fin_sql.data, node_id);
+		if (!PQsendQuery(con, fin_sql.data)
 			|| (sequential && !wait_command_completion(con)))
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s<error>%d:Failed to send query '%s': %s</error>",
-								  errmsg, node_id, sql, PQerrorMessage(con));
+				errstr = psprintf("%s<error>%d:Failed to send query '%s': %s</error>",
+								  errstr, node_id, fin_sql.data, PQerrorMessage(con));
 				chan[n_cmds-1].sql = NULL;
 				continue;
 			}
-			errmsg = psprintf("Failed to send query '%s' to node %d: %s'", sql,
+			errstr = psprintf("Failed to send query '%s' to node %d: %s'", fin_sql.data,
 							  node_id, PQerrorMessage(con));
 			goto cleanup;
 		}
@@ -255,6 +284,9 @@ broadcast(PG_FUNCTION_ARGS)
 		elog(ERROR, "SHARDMAN: Junk at end of command list: %s", cmd);
 	}
 
+	/*
+	 * Now collect results
+	 */
 	for (i = 0; i < n_cmds; i++)
 	{
 		PGresult* next_res;
@@ -269,6 +301,7 @@ broadcast(PG_FUNCTION_ARGS)
 			continue;
 		}
 
+		/* Skip all but the last result */
 		while ((next_res = PQgetResult(con)) != NULL)
 		{
 			if (res != NULL)
@@ -277,31 +310,32 @@ broadcast(PG_FUNCTION_ARGS)
 			}
 			res = next_res;
 		}
+
 		if (res == NULL)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s<error>%d:Failed to received response for '%s': %s</error>",
-								  errmsg, chan[i].node, chan[i].sql, PQerrorMessage(con));
+				errstr = psprintf("%s<error>%d:Failed to received response for '%s': %s</error>",
+								  errstr, chan[i].node, chan[i].sql, PQerrorMessage(con));
 				continue;
 			}
-			errmsg = psprintf("Failed to receive response for query %s from node %d: %s",
+			errstr = psprintf("Failed to receive response for query %s from node %d: %s",
 							  chan[i].sql, chan[i].node, PQerrorMessage(con));
 			goto cleanup;
 		}
 
-		/* Ok, result was successfully fetched */
+		/* Ok, result was successfully fetched, add it to resp */
 		status = PQresultStatus(res);
 		if (status != PGRES_TUPLES_OK && status != PGRES_COMMAND_OK)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s<error>%d:Command %s failed: %s</error>",
-								  errmsg, chan[i].node, chan[i].sql, PQerrorMessage(con));
+				errstr = psprintf("%s<error>%d:Command %s failed: %s</error>",
+								  errstr, chan[i].node, chan[i].sql, PQerrorMessage(con));
 				PQclear(res);
 				continue;
 			}
-			errmsg = psprintf("Command %s failed at node %d: %s",
+			errstr = psprintf("Command %s failed at node %d: %s",
 							  chan[i].sql, chan[i].node, PQerrorMessage(con));
 			PQclear(res);
 			goto cleanup;
@@ -322,7 +356,7 @@ broadcast(PG_FUNCTION_ARGS)
 				}
 				else
 				{
-					errmsg = psprintf("Query '%s' doesn't return single tuple at node %d",
+					errstr = psprintf("Query '%s' doesn't return single tuple at node %d",
 									  chan[i].sql, chan[i].node);
 					PQclear(res);
 					goto cleanup;
@@ -339,13 +373,14 @@ broadcast(PG_FUNCTION_ARGS)
 		}
 		PQclear(res);
 	}
+
   cleanup:
 	for (i = 0; i < n_cmds; i++)
 	{
 		con = chan[i].con;
 		if (two_phase)
 		{
-			if (*errmsg)
+			if (*errstr)
 			{
 				res = PQexec(con, "ROLLBACK PREPARED 'shardlord'");
 				if (PQresultStatus(res) != PGRES_COMMAND_OK)
@@ -369,17 +404,19 @@ broadcast(PG_FUNCTION_ARGS)
 		PQfinish(con);
 	}
 
-	if (*errmsg)
+	if (*errstr)
 	{
 		if (ignore_errors)
 		{
 			resetStringInfo(&resp);
-			appendStringInfoString(&resp, errmsg);
-			elog(WARNING, "SHARDMAN: %s", errmsg);
+			appendStringInfoString(&resp, errstr);
+			elog(WARNING, "SHARDMAN: %s", errstr);
 		}
 		else
 		{
-			elog(ERROR, "SHARDMAN: %s", errmsg);
+			ereport(ERROR,
+					(errcode(ERRCODE_EXTERNAL_ROUTINE_INVOCATION_EXCEPTION),
+					 errmsg("SHARDMAN: %s", errstr)));
 		}
 	}
 
@@ -605,3 +642,45 @@ get_system_identifier(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_INT64(GetSystemIdentifier());
 }
+
+/*
+ * Execute "ALTER SYSTEM SET synchronous_standby_names = '' on commit"
+ */
+Datum
+reset_synchronous_standby_names_on_commit(PG_FUNCTION_ARGS)
+{
+	if (!reset_ssn_callback_set)
+		RegisterXactCallback(reset_ssn_xact_callback, NULL);
+	reset_ssn_requested = true;
+	PG_RETURN_VOID();
+}
+
+static void
+reset_ssn_xact_callback(XactEvent event, void *arg)
+{
+	if (reset_ssn_requested)
+	{
+		/* I just wanted to practice a bit with PG nodes and lists */
+		A_Const *aconst = makeNode(A_Const);
+		List *set_stmt_args = list_make1(aconst);
+		VariableSetStmt setstmt;
+		AlterSystemStmt altersysstmt;
+
+		aconst->val.type = T_String;
+		aconst->val.val.str = ""; /* set it to empty value */
+		aconst->location = -1;
+
+		setstmt.type = T_VariableSetStmt;
+		setstmt.kind = VAR_SET_VALUE;
+		setstmt.name = "synchronous_standby_names";
+		setstmt.args = set_stmt_args;
+
+		altersysstmt.type = T_AlterSystemStmt;
+		altersysstmt.setstmt = &setstmt;
+		AlterSystemSetConfigFile(&altersysstmt);
+		pg_reload_conf(NULL);
+
+		list_free_deep(setstmt.args);
+		reset_ssn_requested = false;
+	}
+}

contrib/pg_shardman/pg_shardman.c

@@ -1,4 +1,4 @@
-shared_preload_libraries = 'pg_pathman'
+shared_preload_libraries = 'pg_shardman, pg_pathman'
 
 shardman.shardlord_connstring = 'port=5432' # shardlord's connstring
 

contrib/pg_shardman/postgresql.conf.common

@@ -22,7 +22,7 @@ synchronous_commit = on
 # for 2pc
 max_prepared_transactions = 1000
 postgres_fdw.use_2pc = on
-# only for testing performace; setting this to 'on' violates visibility
+# only for testing performance; setting this to 'off' violates visibility
 postgres_fdw.use_repeatable_read = off
 
 # This node is shardlord?