Encoding problem

Author: danolivo

exec_plan.c

@@ -11,6 +11,7 @@
 #include "nodes/params.h"
 #include "optimizer/planner.h"
 #include "storage/lmgr.h"
+#include "tcop/utility.h"
 #include "utils/builtins.h"
 #include "utils/guc.h"
 #include "utils/plancache.h"
@@ -22,25 +23,33 @@ PG_FUNCTION_INFO_V1(pg_execute_plan);
 
 void _PG_init(void);
 
+static ProcessUtility_hook_type 	next_ProcessUtility_hook = NULL;
 static planner_hook_type			prev_planner_hook = NULL;
+static ExecutorStart_hook_type		prev_ExecutorStart = NULL;
 static ExecutorEnd_hook_type		prev_ExecutorEnd = NULL;
 
+static void HOOK_Utility_injection(PlannedStmt *pstmt, const char *queryString,
+						ProcessUtilityContext context, ParamListInfo params,
+						QueryEnvironment *queryEnv, DestReceiver *dest,
+						char *completionTag);
 static PlannedStmt *HOOK_Planner_injection(Query *parse, int cursorOptions,
 											ParamListInfo boundParams);
+static void HOOK_ExecStart_injection(QueryDesc *queryDesc, int eflags);
 static void HOOK_ExecEnd_injection(QueryDesc *queryDesc);
 static char * serialize_plan(PlannedStmt *pstmt, ParamListInfo boundParams,
-																	int *size);
+							 const char *querySourceText);
+static void execute_query(char *planString);
+
 static PGconn	*conn = NULL;
 
 int node_number1 = 0;
-//#include "utils/guc.h"
+
 /*
  * Module load/unload callback
  */
 void
 _PG_init(void)
 {
-	elog(LOG, "_PG_Init");
 	DefineCustomIntVariable("pargres.node",
 							"Node number in instances collaboration",
 							NULL,
@@ -54,28 +63,90 @@ _PG_init(void)
 							NULL,
 							NULL);
 
+	/* ProcessUtility hook */
+	next_ProcessUtility_hook = ProcessUtility_hook;
+	ProcessUtility_hook = HOOK_Utility_injection;
+
 	/* Planner hook */
 	prev_planner_hook	= planner_hook;
 	planner_hook		= HOOK_Planner_injection;
 
+	prev_ExecutorStart = ExecutorStart_hook;
+	ExecutorStart_hook = HOOK_ExecStart_injection;
+
 	prev_ExecutorEnd = ExecutorEnd_hook;
 	ExecutorEnd_hook = HOOK_ExecEnd_injection;
 }
 
 static void
-HOOK_ExecEnd_injection(QueryDesc *queryDesc)
+HOOK_Utility_injection(PlannedStmt *pstmt,
+						const char *queryString,
+						ProcessUtilityContext context,
+						ParamListInfo params,
+						QueryEnvironment *queryEnv,
+						DestReceiver *dest,
+						char *completionTag)
 {
-	PGresult *result;
+	Node	*parsetree = pstmt->utilityStmt;
+
+	if ((OidIsValid(get_extension_oid("execplan", true))) &&
+		(node_number1 == 0) &&
+		(nodeTag(parsetree) == T_CreateStmt))
+	{
+		char	conninfo[1024];
+		int		status;
+
+		elog(LOG, "Send UTILITY query: %s", queryString);
+
+		/* Connect to slave and send it a query plan */
+		sprintf(conninfo, "host=localhost port=5433%c", '\0');
+		conn = PQconnectdb(conninfo);
+		if (PQstatus(conn) == CONNECTION_BAD)
+			elog(LOG, "Connection error. conninfo: %s", conninfo);
+
+		status = PQsendQuery(conn, queryString);
+		if (status == 0)
+			elog(ERROR, "Query sending error: %s", PQerrorMessage(conn));
+	}
+
+	if (next_ProcessUtility_hook)
+		(*next_ProcessUtility_hook) (pstmt, queryString, context, params,
+									 queryEnv, dest, completionTag);
+	else
+		standard_ProcessUtility(pstmt, queryString,
+											context, params, queryEnv,
+											dest, completionTag);
 
-	/* Execute before hook because it destruct memory context of exchange list */
 	if (conn)
+	{
+		PGresult *result;
+
 		while ((result = PQgetResult(conn)) != NULL)
 			Assert(PQresultStatus(result) != PGRES_FATAL_ERROR);
+		PQfinish(conn);
+		conn = NULL;
+	}
+}
 
-	if (prev_ExecutorEnd)
-		prev_ExecutorEnd(queryDesc);
-	else
-		standard_ExecutorEnd(queryDesc);
+static void
+execute_query(char *planString)
+{
+	char	conninfo[1024];
+	char	*SQLCommand;
+	int		status;
+
+	/* Connect to slave and send it a query plan */
+	sprintf(conninfo, "host=localhost port=5433%c", '\0');
+	conn = PQconnectdb(conninfo);
+	if (PQstatus(conn) == CONNECTION_BAD)
+		elog(LOG, "Connection error. conninfo: %s", conninfo);
+
+	SQLCommand = (char *) palloc0(strlen(planString)+100);
+	sprintf(SQLCommand, "SELECT pg_execute_plan('%s')", planString);
+//elog(LOG, "query: %s", SQLCommand);
+	status = PQsendQuery(conn, SQLCommand);
+	if (status == 0)
+		elog(ERROR, "Query sending error: %s", PQerrorMessage(conn));
 }
 
 static PlannedStmt *
@@ -91,72 +162,103 @@ HOOK_Planner_injection(Query *parse, int cursorOptions,
 	else
 		pstmt = standard_planner(parse, cursorOptions, boundParams);
 
-	if (node_number1 > 0)
+	if ((node_number1 > 0) || (parse->utilityStmt != NULL))
 		return pstmt;
-	else
-		printf("SEND Query\n");
 
 	/* Extension is not initialized. */
 	if (OidIsValid(get_extension_oid("execplan", true)))
 	{
-		char	conninfo[1024];
-		char	*data,
-				*SQLCommand;
-		int		status,
-				data_size;
 
-		/* Connect to slave and send it a query plan */
-		sprintf(conninfo, "host=localhost port=5433%c", '\0');
-		conn = PQconnectdb(conninfo);
-		if (PQstatus(conn) == CONNECTION_BAD)
-			elog(LOG, "Connection error. conninfo: %s", conninfo);
-
-		data = serialize_plan(pstmt, boundParams, &data_size);
-		SQLCommand = (char *) palloc0(strlen(data)+100);
-		sprintf(SQLCommand, "SELECT pg_execute_plan('%s')", data);
-elog(LOG, "query: %s", SQLCommand);
-		status = PQsendQuery(conn, SQLCommand);
-		if (status == 0)
-			elog(ERROR, "Query sending error: %s", PQerrorMessage(conn));
 	}
 	return pstmt;
 }
 
+static void
+HOOK_ExecStart_injection(QueryDesc *queryDesc, int eflags)
+{
+	Node	*parsetree = queryDesc->plannedstmt->utilityStmt;
+
+	if ((OidIsValid(get_extension_oid("execplan", true))) &&
+		(node_number1 == 0) &&
+		((parsetree == NULL) || (nodeTag(parsetree) != T_CreatedbStmt)))
+	{
+		elog(LOG, "Send query: %s", queryDesc->sourceText);
+		execute_query(serialize_plan(queryDesc->plannedstmt, queryDesc->params,
+									 queryDesc->sourceText));
+	}
+	else
+	{
+//		elog(LOG, "EXECUTOR Process query without sending. IsParsetree=%hhu node_number1=%d IsExt=%hhu", parsetree != NULL, node_number1, OidIsValid(get_extension_oid("execplan", true)));
+	}
+
+	if (prev_ExecutorStart)
+		prev_ExecutorStart(queryDesc, eflags);
+	else
+		standard_ExecutorStart(queryDesc, eflags);
+}
+
+static void
+HOOK_ExecEnd_injection(QueryDesc *queryDesc)
+{
+	/* Execute before hook because it destruct memory context of exchange list */
+	if (conn)
+	{
+		PGresult *result;
+
+		while ((result = PQgetResult(conn)) != NULL)
+			Assert(PQresultStatus(result) != PGRES_FATAL_ERROR);
+		PQfinish(conn);
+		conn = NULL;
+	}
+
+	if (prev_ExecutorEnd)
+		prev_ExecutorEnd(queryDesc);
+	else
+		standard_ExecutorEnd(queryDesc);
+}
+
 #include "utils/fmgrprotos.h"
 
 static char *
-serialize_plan(PlannedStmt *pstmt, ParamListInfo boundParams, int *size)
+serialize_plan(PlannedStmt *pstmt, ParamListInfo boundParams,
+			   const char *querySourceText)
 {
 	int		splan_len,
 			sparams_len,
+			qtext_len,
 			econtainer_len;
 	char	*serialized_plan,
 			*container,
 			*start_address,
 			*econtainer;
 
-	Assert(size != NULL);
-
 	serialized_plan = nodeToString(pstmt);
 
-	/* We use splan_len+1 bytes for include end-of-string symbol. */
+	/* We use len+1 bytes for include end-of-string symbol. */
 	splan_len = strlen(serialized_plan) + 1;
+	qtext_len = strlen(querySourceText) + 1;
 
 	sparams_len = EstimateParamListSpace(boundParams);
 
-	container = (char *) palloc0(splan_len+sparams_len);
+	container = (char *) palloc0(splan_len + sparams_len + qtext_len);
 //elog(LOG, "Serialize sizes: plan: %d params: %d, numParams: %d", splan_len, sparams_len, boundParams->numParams);
 	memcpy(container, serialized_plan, splan_len);
-	start_address = container+splan_len;
+	start_address = container + splan_len;
 	SerializeParamList(boundParams, &start_address);
 
-	econtainer_len = esc_enc_len(container, splan_len+sparams_len);
-	econtainer = (char *) palloc0(econtainer_len + 1);
+	Assert(start_address == container + splan_len + sparams_len);
+	memcpy(start_address, querySourceText, qtext_len);
 
-	Assert(econtainer_len == esc_encode(container, splan_len+sparams_len, econtainer));
+	econtainer_len = pg_base64_enc_len(container, splan_len + sparams_len + qtext_len);
+	econtainer = (char *) palloc0(econtainer_len + 1);
+	if (econtainer_len != pg_base64_encode(container, splan_len + sparams_len +
+			qtext_len, econtainer))
+		elog(LOG, "econtainer_len: %d %d", econtainer_len, pg_base64_encode(container, splan_len + sparams_len +
+				qtext_len, econtainer));
+	Assert(econtainer_len == pg_base64_encode(container, splan_len + sparams_len +
+										qtext_len, econtainer));
 	econtainer[econtainer_len] = '\0';
-	*size = econtainer_len + 1;
-	elog(LOG, "Serialize sizes: econtainer: %d", *size);
+
 	return econtainer;
 }
 
@@ -209,30 +311,22 @@ pg_execute_plan(PG_FUNCTION_ARGS)
 	char			*data = TextDatumGetCString(PG_GETARG_DATUM(0));
 	PlannedStmt		*pstmt;
 	QueryDesc		*queryDesc;
-	char			queryString[5] = "NONE";
+	char			*queryString;
 	ParamListInfo 	paramLI = NULL;
 	int				dec_tot_len;
 	char			*dcontainer,
 					*start_addr;
 
-	elog(LOG, "datalen=%lu\n", strlen(data));
 	/* Compute decoded size of bytea data */
-	dec_tot_len = esc_dec_len(data, strlen(data));
-	elog(LOG, "dec_tot_len=%d datalen=%lu\n", dec_tot_len, strlen(data));
+	dec_tot_len = pg_base64_dec_len(data, strlen(data));
 	dcontainer = (char *) palloc0(dec_tot_len);
-	Assert(dec_tot_len == esc_decode(data, strlen(data), dcontainer));
+	Assert(dec_tot_len == pg_base64_decode(data, strlen(data), dcontainer));
 
 	pstmt = (PlannedStmt *) stringToNode((char *) dcontainer);
-	elog(LOG, "Serialize Plan Size=%lu\n", strlen(dcontainer));
 	start_addr = dcontainer + strlen(dcontainer) + 1;
 	paramLI = RestoreParamList((char **) &start_addr);
-	elog(LOG, "Decoded params. numParams: %d\n", paramLI->numParams);
-//	printf("INCOMING: %s\n", data);
-//	PG_RETURN_BOOL(true);
-	/* Execute query plan. Based on execParallel.c ParallelQueryMain() */
-//
-//	ptr += strlen((const char *) ptr);
-//
+	queryString = start_addr;
+//	elog(LOG, "Decoded query: %s\n", start_addr);
 
 	queryDesc = CreateQueryDesc(pstmt, queryString, GetActiveSnapshot(),
 								InvalidSnapshot, None_Receiver, paramLI, NULL,

execplan.control

@@ -1,4 +1,4 @@
-# Pargres extension
+# Execplan extension
 comment = 'Execute raw query plan'
 default_version = '0.1'
 module_pathname = '$libdir/execplan'