be slightly more clever about guc variables

Author: kelvich

multimaster--1.0.sql

@@ -13,9 +13,10 @@ CREATE FUNCTION mtm.drop_node(node integer, drop_slot bool default false) RETURN
 AS 'MODULE_PATHNAME','mtm_drop_node'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.add_node(conn_str cstring) RETURNS void
-AS 'MODULE_PATHNAME','mtm_add_node'
-LANGUAGE C;
+-- -- XXX: cstring as an argument breaks sanity check
+-- CREATE FUNCTION mtm.add_node(conn_str cstring) RETURNS void
+-- AS 'MODULE_PATHNAME','mtm_add_node'
+-- LANGUAGE C;
 
 -- Create replication slot for the node which was previously dropped together with it's slot 
 CREATE FUNCTION mtm.recover_node(node integer) RETURNS void
@@ -69,7 +70,7 @@ CREATE FUNCTION mtm.inject_2pc_error(stage integer) RETURNS void
 AS 'MODULE_PATHNAME','mtm_inject_2pc_error'
 LANGUAGE C;
 
-CREATE TABLE IF NOT EXISTS public.ddl_log (issued timestamp with time zone not null, query text);
+-- CREATE TABLE IF NOT EXISTS public.ddl_log (issued timestamp with time zone not null, query text);
 
-CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema text, rel_name text, primary key(rel_schema, rel_name));
+-- CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema text, rel_name text, primary key(rel_schema, rel_name));
 

multimaster.c

@@ -234,8 +234,8 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag);
 
-static StringInfo	MtmGUCBuffer;
-static bool			MtmGUCBufferAllocated = false;
+// static StringInfo	MtmGUCBuffer;
+// static bool			MtmGUCBufferAllocated = false;
 
 /*
  * -------------------------------------------
@@ -3171,63 +3171,37 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 	}
 }
 
-static void MtmGUCBufferAppend(const char *gucQueryString){
+// static void MtmGUCBufferAppend(const char *gucQueryString){
 
-	if (!MtmGUCBufferAllocated)
-	{
-		MemoryContext oldcontext;
-		oldcontext = MemoryContextSwitchTo(TopMemoryContext);
-		MtmGUCBuffer = makeStringInfo();
-		MemoryContextSwitchTo(oldcontext);
-		MtmGUCBufferAllocated = true;
-		appendStringInfoString(MtmGUCBuffer, "RESET SESSION AUTHORIZATION; reset all;");
-	}
+// 	if (!MtmGUCBufferAllocated)
+// 	{
+// 		MemoryContext oldcontext;
+// 		oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+// 		MtmGUCBuffer = makeStringInfo();
+// 		MemoryContextSwitchTo(oldcontext);
+// 		MtmGUCBufferAllocated = true;
+// 		appendStringInfoString(MtmGUCBuffer, "RESET SESSION AUTHORIZATION; reset all;");
+// 	}
 
-	appendStringInfoString(MtmGUCBuffer, gucQueryString);
-	/* sometimes there is no ';' char at the end. */
-	// appendStringInfoString(MtmGUCBuffer, ";");
-}
-
-static char * MtmGUCBufferGet(void){
-	if (!MtmGUCBufferAllocated)
-		MtmGUCBufferAppend("");
-	return MtmGUCBuffer->data;
-}
+// 	appendStringInfoString(MtmGUCBuffer, gucQueryString);
+// 	/* sometimes there is no ';' char at the end. */
+// 	// appendStringInfoString(MtmGUCBuffer, ";");
+// }
 
-static void MtmGUCBufferClear(void)
-{
-	if (MtmGUCBufferAllocated)
-	{
-		resetStringInfo(MtmGUCBuffer);
-		MtmGUCBufferAppend("");
-	}
-}
-
-static bool MtmProcessDDLCommand(char const* queryString)
-{
-	char	   *queryWithContext;
-	char	   *gucContext;
-
-	/* Append global GUC to utility stmt. */
-	gucContext = MtmGUCBufferGet();
-	if (gucContext)
-	{
-		queryWithContext = palloc(strlen(gucContext) + strlen(queryString) +  1);
-		strcpy(queryWithContext, gucContext);
-		strcat(queryWithContext, queryString);
-	}
-	else
-	{
-		queryWithContext = (char *) queryString;
-	}
-
-	MTM_LOG1("Sending utility: %s", queryWithContext);
-	LogLogicalMessage("MTM:GUC", queryWithContext, strlen(queryWithContext), true);
-
-	MtmTx.containsDML = true;
-	return false;
-}
+// static char * MtmGUCBufferGet(void){
+// 	if (!MtmGUCBufferAllocated)
+// 		MtmGUCBufferAppend("");
+// 	return MtmGUCBuffer->data;
+// }
 
+// static void MtmGUCBufferClear(void)
+// {
+// 	if (MtmGUCBufferAllocated)
+// 	{
+// 		resetStringInfo(MtmGUCBuffer);
+// 		MtmGUCBufferAppend("");
+// 	}
+// }
 
 /*
  * Genenerate global transaction identifier for two-pahse commit.
@@ -3280,6 +3254,167 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	return false;
 }
 
+
+/*
+ * -------------------------------------------
+ * GUC Context Handling
+ * -------------------------------------------
+ */
+
+// XXX: is it defined somewhere?
+#define GUC_KEY_MAXLEN 255
+
+#define MTM_GUC_HASHSIZE 20
+
+typedef struct MtmGucHashEntry
+{
+	char	key[GUC_KEY_MAXLEN];
+	char   *value;
+} MtmGucHashEntry;
+
+static HTAB *MtmGucHash = NULL;
+
+static void MtmGucHashInit(void)
+{
+	HASHCTL		hash_ctl;
+
+	MemSet(&hash_ctl, 0, sizeof(hash_ctl));
+	hash_ctl.keysize = GUC_KEY_MAXLEN;
+	hash_ctl.entrysize = sizeof(MtmGucHashEntry);
+	hash_ctl.hcxt = TopMemoryContext;
+	MtmGucHash = hash_create("MtmGucHash",
+						MTM_GUC_HASHSIZE,
+						&hash_ctl,
+						HASH_ELEM | HASH_CONTEXT);
+}
+
+static void MtmGucSet(VariableSetStmt *stmt, const char *queryStr)
+{
+	MemoryContext oldcontext;
+	MtmGucHashEntry *hentry;
+	bool found;
+	char *key;
+
+	if (!MtmGucHash)
+		MtmGucHashInit();
+
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+
+	switch (stmt->kind)
+	{
+		case VAR_SET_VALUE:
+		case VAR_SET_DEFAULT:
+		case VAR_SET_CURRENT:
+			{
+				char *value;
+
+				key = pstrdup(stmt->name);
+				hash_search(MtmGucHash, key, HASH_FIND,  &found);
+				value = ExtractSetVariableArgs(stmt);
+
+				fprintf(stderr, ":MtmGucSet: %s -> %s\n", key, value);
+
+				if (value)
+				{
+					hentry = (MtmGucHashEntry *) hash_search(MtmGucHash, key,
+															 HASH_ENTER,  &found);
+
+					// if (found)
+						// pfree(hentry->value);
+
+					hentry->value = palloc(strlen(value) + 1);
+					strcpy(hentry->value, value);
+				}
+				else if (found)
+				{
+					/* That was SET TO DEFAULT and we already had some value */
+					hash_search(MtmGucHash, key, HASH_REMOVE, NULL);
+				}
+			}
+			break;
+
+		case VAR_RESET:
+			{
+				key = pstrdup(stmt->name);
+				hash_search(MtmGucHash, key, HASH_REMOVE, NULL);
+			}
+			break;
+		case VAR_RESET_ALL:
+			break;
+
+		case VAR_SET_MULTI:
+			break;
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+}
+
+static void MtmGucDiscard(DiscardStmt *stmt)
+{
+
+}
+
+static void MtmGucClear(void)
+{
+
+}
+
+static char * MtmGucSerialize(void)
+{
+	HASH_SEQ_STATUS status;
+	MtmGucHashEntry *hentry;
+	StringInfo serialized_gucs;
+
+	serialized_gucs = makeStringInfo();
+	appendStringInfoString(serialized_gucs, "RESET SESSION AUTHORIZATION; reset all; ");
+
+	if (MtmGucHash)
+	{
+		hash_seq_init(&status, MtmGucHash);
+		while ((hentry = (MtmGucHashEntry *) hash_seq_search(&status)) != NULL)
+		{
+			appendStringInfoString(serialized_gucs, "SET ");
+			appendStringInfoString(serialized_gucs, hentry->key);
+			appendStringInfoString(serialized_gucs, " TO ");
+			appendStringInfoString(serialized_gucs, hentry->value);
+			appendStringInfoString(serialized_gucs, "; ");
+		}
+	}
+
+	return serialized_gucs->data;
+}
+
+/*
+ * -------------------------------------------
+ * DDL Handling
+ * -------------------------------------------
+ */
+
+static bool MtmProcessDDLCommand(char const* queryString)
+{
+	char	   *queryWithContext;
+	char	   *gucContext;
+
+	/* Append global GUC to utility stmt. */
+	gucContext = MtmGucSerialize();
+	if (gucContext)
+	{
+		queryWithContext = palloc(strlen(gucContext) + strlen(queryString) +  1);
+		strcpy(queryWithContext, gucContext);
+		strcat(queryWithContext, queryString);
+	}
+	else
+	{
+		queryWithContext = (char *) queryString;
+	}
+
+	MTM_LOG1("Sending utility: %s", queryWithContext);
+	LogLogicalMessage("MTM:GUC", queryWithContext, strlen(queryWithContext), true);
+
+	MtmTx.containsDML = true;
+	return false;
+}
+
 static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag)
@@ -3339,30 +3474,24 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		    skipCommand = true;
 			break;
 
-		/* Do not skip following unless temp object was accessed */
-		case T_CreateTableAsStmt:
-		case T_CreateStmt:
-		case T_ViewStmt:
-		case T_IndexStmt:
-		case T_DropStmt:
-			break;
+		// /* Do not skip following unless temp object was accessed */
+		// case T_CreateTableAsStmt:
+		// case T_CreateStmt:
+		// case T_ViewStmt:
+		// case T_IndexStmt:
+		// case T_DropStmt:
+		// 	break;
 
 		/* Save GUC context for consequent DDL execution */
 		case T_DiscardStmt:
 			{
-				/*
-				 * DiscardStmt *stmt = (DiscardStmt *) parsetree;
-				 * skipCommand = stmt->target == DISCARD_TEMP;
-				 */
+				DiscardStmt *stmt = (DiscardStmt *) parsetree;
+				skipCommand = stmt->target == DISCARD_TEMP; // XXX
 
 				if (!IsTransactionBlock())
 				{
-					/*
-					 * XXX: move allocation somewhere to backend startup and check
-					 * where buffer is empty in send routines.
-					 */
-					MtmGUCBufferAllocated = false;
-					pfree(MtmGUCBuffer);
+					skipCommand = true;
+					MtmGucDiscard(stmt);
 				}
 			}
 			break;
@@ -3406,12 +3535,12 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 				bool		isTopLevel = (context == PROCESS_UTILITY_TOPLEVEL);
 
 				if (stmt->kind == VAR_RESET && strcmp(stmt->name, "session_authorization") == 0)
-					MtmGUCBufferClear();
+					MtmGucClear();
 
 				if (!IsTransactionBlock())
 				{
 					skipCommand = true;
-					MtmGUCBufferAppend(queryString);
+					MtmGucSet(stmt, queryString);
 				}
 			}
 			break;
@@ -3444,7 +3573,8 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			skipCommand = false;
 			break;
 	}
-	if (context == PROCESS_UTILITY_TOPLEVEL)
+
+	if (context == PROCESS_UTILITY_TOPLEVEL) // || context == PROCESS_UTILITY_QUERY)
 	{
 		if (!skipCommand && !MtmTx.isReplicated) {
 			if (MtmProcessDDLCommand(queryString)) {