Add OIDs list processing + portable plan sign

Author: danolivo

.gitignore

@@ -42,3 +42,6 @@ lib*.pc
 /Debug/
 /Release/
 /tmp_install/
+/.cproject
+/.project
+/.settings/

contrib/Makefile

@@ -30,6 +30,7 @@ SUBDIRS = \
 		pageinspect	\
 		passwordcheck	\
 		pg_buffercache	\
+		pg_execplan	\
 		pg_freespacemap \
 		pg_prewarm	\
 		pg_standby	\
@@ -38,6 +39,7 @@ SUBDIRS = \
 		pgcrypto	\
 		pgrowlocks	\
 		pgstattuple	\
+		pg_repeater	\
 		pg_visibility	\
 		postgres_fdw	\
 		seg		\

contrib/pg_execplan/init.sql

@@ -4,14 +4,15 @@
 -- query - query string which will be parsed and planned.
 -- filename - path to the file on a disk.
 CREATE OR REPLACE FUNCTION @extschema@.pg_store_query_plan(
-															filename TEXT,
-															query TEXT
-														  )
+															filename	TEXT,
+															query		TEXT
+															)
 RETURNS VOID AS 'pg_execplan'
 LANGUAGE C;
 
-CREATE OR REPLACE FUNCTION @extschema@.pg_exec_plan(query TEXT,
-													plan TEXT
+CREATE OR REPLACE FUNCTION @extschema@.pg_exec_plan(query	TEXT,
+													plan	TEXT,
+													params	TEXT
 													)
 RETURNS BOOL AS 'pg_execplan'
 LANGUAGE C;

contrib/pg_execplan/pg_execplan.c

@@ -11,6 +11,7 @@
 #include "common/base64.h"
 #include "executor/executor.h"
 #include "nodes/nodes.h"
+#include "nodes/params.h"
 #include "nodes/plannodes.h"
 #include "tcop/pquery.h"
 #include "tcop/utility.h"
@@ -42,10 +43,13 @@ _PG_init(void)
 Datum
 pg_store_query_plan(PG_FUNCTION_ARGS)
 {
-	char			*query_string = TextDatumGetCString(PG_GETARG_DATUM(1)),
-					*filename = TextDatumGetCString(PG_GETARG_DATUM(0)),
-					*plan_string;
-	int				nstmts;
+	char			*filename = TextDatumGetCString(PG_GETARG_DATUM(0)),
+					*squery = TextDatumGetCString(PG_GETARG_DATUM(1)),
+					*sparams = NULL,
+					*start_address,
+					*splan;
+	int				nstmts,
+					sparams_len;
 	FILE			*fout;
 	MemoryContext	oldcontext;
 	List	  		*parsetree_list;
@@ -56,21 +60,21 @@ pg_store_query_plan(PG_FUNCTION_ARGS)
 	size_t			string_len;
 
 	if (EXPLAN_DEBUG_LEVEL > 0)
-		elog(LOG, "Store into %s plan of the query %s.", filename, query_string);
+		elog(LOG, "Store into %s plan of the query %s.", filename, squery);
 
 	oldcontext = MemoryContextSwitchTo(MessageContext);
 
-	parsetree_list = pg_parse_query(query_string);
+	parsetree_list = pg_parse_query(squery);
 	nstmts = list_length(parsetree_list);
 	if (nstmts != 1)
 		elog(ERROR, "Query contains %d elements, but must contain only one.", nstmts);
 
 	parsetree = (RawStmt *) linitial(parsetree_list);
-	querytree_list = pg_analyze_and_rewrite(parsetree, query_string, NULL, 0, NULL);
+	querytree_list = pg_analyze_and_rewrite(parsetree, squery, NULL, 0, NULL);
 	plantree_list = pg_plan_queries(querytree_list, CURSOR_OPT_PARALLEL_OK, NULL);
 
 	queryDesc = CreateQueryDesc((PlannedStmt *) linitial(plantree_list),
-												query_string,
+												squery,
 												InvalidSnapshot,
 												InvalidSnapshot,
 												None_Receiver,
@@ -82,24 +86,35 @@ pg_store_query_plan(PG_FUNCTION_ARGS)
 
 	fout = fopen(filename, "wb");
 	Assert(fout != NULL);
-	string_len = strlen(query_string);
+	string_len = strlen(squery);
 	fwrite(&string_len, sizeof(size_t), 1, fout);
-	fwrite(query_string, sizeof(char), string_len, fout);
+	fwrite(squery, sizeof(char), string_len, fout);
 
 	set_portable_output(true);
-	plan_string = nodeToString(queryDesc->plannedstmt);
+	splan = nodeToString(queryDesc->plannedstmt);
 	set_portable_output(false);
-	string_len = strlen(plan_string);
+	string_len = strlen(splan);
 	fwrite(&string_len, sizeof(size_t), 1, fout);
-	fwrite(plan_string, sizeof(char), string_len, fout);
-
+	fwrite(splan, sizeof(char), string_len, fout);
+
+	/*
+	 * Serialize parameters list. In this case we have no parameters and will
+	 * serialize NULL list.
+	 */
+	sparams_len = EstimateParamListSpace(NULL);
+	sparams = palloc0(sparams_len);
+	start_address = sparams;
+	SerializeParamList(NULL, &start_address);
+	string_len = sparams_len;
+	fwrite(&string_len, sizeof(size_t), 1, fout);
+	fwrite(sparams, sizeof(char), string_len, fout);
 	fclose(fout);
 	MemoryContextSwitchTo(oldcontext);
 	PG_RETURN_VOID();
 }
 
 static void
-exec_plan(char *query_string, char *plan_string)
+exec_plan(char *squery, char *splan, char *sparams)
 {
 	PlannedStmt			*pstmt;
 	ParamListInfo 		paramLI = NULL;
@@ -108,28 +123,42 @@ exec_plan(char *query_string, char *plan_string)
 	QueryDesc			*queryDesc;
 	DestReceiver		*receiver;
 	int					eflags = 0;
+	Oid					*param_types = NULL;
+	char				*start_address = sparams;
+
+	Assert(squery && splan && sparams);
 
 	PG_TRY();
 	{
-		set_portable_input(true);
-		pstmt = (PlannedStmt *) stringToNode(plan_string);
-		set_portable_input(false);
+		pstmt = (PlannedStmt *) stringToNode(splan);
+
+		/* Deserialize parameters of the query */
+		paramLI = RestoreParamList(&start_address);
 	}
 	PG_CATCH();
 	{
-		elog(INFO, "BAD PLAN: %s. Query: %s", plan_string, query_string);
+		elog(INFO, "BAD PLAN: %s. Query: %s", splan, squery);
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
 
 	if (EXPLAN_DEBUG_LEVEL > 0)
-		elog(INFO, "query: %s\n", query_string);
+		elog(INFO, "query: %s\n", squery);
 	if (EXPLAN_DEBUG_LEVEL > 1)
-		elog(INFO, "\nplan: %s\n", plan_string);
+		elog(INFO, "\nplan: %s\n", splan);
+
+	psrc = CreateCachedPlan(NULL, squery, NULL);
+
+	if (paramLI->numParams > 0)
+	{
+		int i;
 
-	psrc = CreateCachedPlan(NULL, query_string, NULL);
-	CompleteCachedPlan(psrc, NIL, NULL, NULL, 0, NULL, NULL,
-							   CURSOR_OPT_GENERIC_PLAN, false);
+		param_types = palloc(sizeof(Oid) * paramLI->numParams);
+		for (i = 0; i < paramLI->numParams; i++)
+			param_types[i] = paramLI->params[i].ptype;
+	}
+	CompleteCachedPlan(psrc, NIL, NULL, param_types, paramLI->numParams, NULL,
+								NULL, CURSOR_OPT_GENERIC_PLAN, false);
 
 	SetRemoteSubplan(psrc, pstmt);
 	cplan = GetCachedPlan(psrc, paramLI, false, NULL);
@@ -139,7 +168,7 @@ exec_plan(char *query_string, char *plan_string)
 	PG_TRY();
 	{
 		queryDesc = CreateQueryDesc(pstmt,
-									query_string,
+									squery,
 									GetActiveSnapshot(),
 									InvalidSnapshot,
 									receiver,
@@ -155,7 +184,7 @@ exec_plan(char *query_string, char *plan_string)
 	}
 	PG_CATCH();
 	{
-		elog(INFO, "BAD QUERY: '%s'.", query_string);
+		elog(INFO, "BAD QUERY: '%s'.", squery);
 		ReleaseCachedPlan(cplan, false);
 		PG_RE_THROW();
 	}
@@ -171,37 +200,49 @@ exec_plan(char *query_string, char *plan_string)
 Datum
 pg_exec_plan(PG_FUNCTION_ARGS)
 {
-	char	*query_string = TextDatumGetCString(PG_GETARG_DATUM(0));
-	char	*plan_string = TextDatumGetCString(PG_GETARG_DATUM(1));
+	char	*squery = TextDatumGetCString(PG_GETARG_DATUM(0));
+	char	*splan = TextDatumGetCString(PG_GETARG_DATUM(1));
+	char	*sparams = TextDatumGetCString(PG_GETARG_DATUM(2));
 
 	char	*dec_query,
-			*dec_plan;
+			*dec_plan,
+			*dec_params;
 	int		dec_query_len,
 			dec_query_len1,
 			dec_plan_len,
-			dec_plan_len1;
+			dec_plan_len1,
+			dec_params_len,
+			dec_params_len1;
 
-	Assert(query_string != NULL);
-	Assert(plan_string != NULL);
+	Assert(squery != NULL);
+	Assert(splan != NULL);
+	Assert(sparams != NULL);
 
-	dec_query_len = pg_b64_dec_len(strlen(query_string) + 1) + 1;
+	dec_query_len = pg_b64_dec_len(strlen(squery));
 	dec_query = palloc0(dec_query_len + 1);
-	dec_query_len1 = pg_b64_decode(query_string, strlen(query_string), dec_query);
-	Assert(dec_query_len > dec_query_len1);
+	dec_query_len1 = pg_b64_decode(squery, strlen(squery), dec_query);
+	Assert(dec_query_len >= dec_query_len1);
 
-	dec_plan_len = pg_b64_dec_len(strlen(plan_string) + 1);
+	dec_plan_len = pg_b64_dec_len(strlen(splan));
 	dec_plan = palloc0(dec_plan_len + 1);
-	dec_plan_len1 = pg_b64_decode(plan_string, strlen(plan_string), dec_plan);
-	Assert(dec_plan_len > dec_plan_len1);
+	dec_plan_len1 = pg_b64_decode(splan, strlen(splan), dec_plan);
+	Assert(dec_plan_len >= dec_plan_len1);
+
+	dec_params_len = pg_b64_dec_len(strlen(sparams));
+	dec_params = palloc0(dec_params_len + 1);
+	dec_params_len1 = pg_b64_decode(sparams, strlen(sparams), dec_params);
+	Assert(dec_params_len >= dec_params_len1);
 
-	exec_plan(dec_query, dec_plan);
+	exec_plan(dec_query, dec_plan, dec_params);
 	pfree(dec_query);
 	pfree(dec_plan);
+	pfree(dec_params);
 	PG_RETURN_BOOL(true);
 }
 
 static void
-LoadPlanFromFile(const char *filename, char **query_string, char **plan_string)
+LoadPlanFromFile(const char *filename, char **squery, char **splan,
+				 char **sparams)
 {
 	FILE	*fin;
 	size_t	string_len;
@@ -210,16 +251,28 @@ LoadPlanFromFile(const char *filename, char **query_string, char **plan_string)
 	fin = fopen(filename, "rb");
 	Assert(fin != NULL);
 
+	/* Read query string size, allocate memory and read query from the file. */
 	nelems = fread(&string_len, sizeof(size_t), 1, fin);
 	Assert(nelems == 1);
-	*query_string = palloc0(string_len + 1);
-	nelems = fread(*query_string, sizeof(char), string_len, fin);
+	*squery = palloc0(string_len + 1);
+	nelems = fread(*squery, sizeof(char), string_len, fin);
 	Assert(nelems == string_len);
 
+	/* Read plan size, allocate memory and read plan from the file. */
 	nelems = fread(&string_len, sizeof(size_t), 1, fin);
 	Assert(nelems == 1);
-	*plan_string = palloc0(string_len + 1);
-	nelems = fread(*plan_string, sizeof(char), string_len, fin);
+	*splan = palloc0(string_len + 1);
+	nelems = fread(*splan, sizeof(char), string_len, fin);
+	Assert(nelems == string_len);
+
+	/*
+	 * Read serialized query parameters string length, allocate memory and
+	 * read it from the file.
+	 */
+	nelems = fread(&string_len, sizeof(size_t), 1, fin);
+	Assert(nelems == 1);
+	*sparams = palloc0(string_len + 1);
+	nelems = fread(*sparams, sizeof(char), string_len, fin);
 	Assert(nelems == string_len);
 
 	fclose(fin);
@@ -229,11 +282,17 @@ LoadPlanFromFile(const char *filename, char **query_string, char **plan_string)
 Datum
 pg_exec_stored_plan(PG_FUNCTION_ARGS)
 {
-	char				*filename = TextDatumGetCString(PG_GETARG_DATUM(0)),
-						*query_string = NULL,
-						*plan_string = NULL;
-
-	LoadPlanFromFile(filename, &query_string, &plan_string);
-	exec_plan(query_string, plan_string);
+	char	*filename = TextDatumGetCString(PG_GETARG_DATUM(0)),
+			*squery = NULL,
+			*splan = NULL,
+			*sparams = NULL;
+
+	LoadPlanFromFile(filename, &squery, &splan, &sparams);
+
+	Assert(squery && splan && sparams);
+	exec_plan(squery, splan, sparams);
+	pfree(squery);
+	pfree(splan);
+	pfree(sparams);
 	PG_RETURN_BOOL(true);
 }

contrib/pg_execplan/tests/rpl.sh

@@ -91,19 +91,3 @@ psql -p 5432 -c "SELECT pg_store_query_plan('../test.txt', '
 				');"
 psql -p 5433 -c "SELECT pg_exec_stored_plan('../test.txt');"
 
-# Prepared operator
-psql -p 5432 -c "PREPARE abc (TEXT, tests.bug_status) AS
-									INSERT INTO bug (description, status)
-									VALUES (\$1,\$2);
-				SELECT pg_store_query_plan('../test.txt', '
-									EXECUTE abc(''test1'', ''closed'')
-						');"
-
-psql -p 5433 -c "PREPARE abc (TEXT, tests.bug_status) AS
-									INSERT INTO bug (description, status)
-									VALUES (\$1,\$2);
-				SELECT pg_exec_stored_plan('../test.txt');"
-									 
-psql -p 5433 -c "SELECT * FROM bug;"
-psql -p 5433 -c "SELECT * FROM bug1;"
-