ExecPlan is released under the PostgreSQL License, a liberal Open Source license, similar to the BSD or MIT licenses.

Copyright (c) 2018-2019, Postgres Professional

Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group

Portions Copyright (c) 1994, The Regents of the University of California

Permission to use, copy, modify, and distribute this software and its documentation for any purpose, without fee, and without a written agreement is hereby granted, provided that the above copyright notice and this paragraph and the following two paragraphs appear in all copies.

IN NO EVENT SHALL POSTGRES PROFESSIONAL BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF POSTGRES PROFESSIONAL HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

POSTGRES PROFESSIONAL SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND POSTGRES PROFESSIONAL HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.

MODULE_big = pg_execplan

EXTENSION = pg_execplan

EXTVERSION = 0.1

PGFILEDESC = "pg_execplan"

MODULES = pg_execplan

OBJS = pg_execplan.o $(WIN32RES)

PG_CPPFLAGS = -I$(libpq_srcdir)

SHLIB_LINK_INTERNAL = $(libpq)

DATA_built = $(EXTENSION)--$(EXTVERSION).sql

ifdef USE_PGXS

PG_CONFIG = pg_config

PGXS := $(shell $(PG_CONFIG) --pgxs)

include $(PGXS)

else

SHLIB_PREREQS = submake-libpq

subdir = contrib/pg_execplan

top_builddir = ../..

include $(top_builddir)/src/Makefile.global

include $(top_srcdir)/contrib/contrib-global.mk

$(EXTENSION)--$(EXTVERSION).sql: init.sql

cat $^ > $@

PostgreSQL patch & extension for raw query plan execution

This project dedicated to query execution problem in DBMS for computing systems with cluster architecture.

The DBMS may need to execute an identical query plan at each computing node.

Today PostgreSQL can process only SQL statements. But it is not guaranteed, that the planner at each node will construct same query plan, because different statistics, relation sizes e.t.c.

This solution based on postgres-xl approach: plan tree is serialized by the nodeToString() routine.

During serialization we transform all database object identifiers (oid) in each node field to portable representation.

Further, the serialized plan transfer by new libpq routine called `PQsendPlan`.

In this project we use postgres_fdw connections for management of sessions and remote transactions.

Some `repeater` extension used for the demonstration of plan transfer machinery.

The `pg12_devel.patch` patch contains all core changes.

The `scripts` directory contains some simplistic demo tests.

\echo Use "CREATE EXTENSION pg_execplan" to load this file. \quit

CREATE OR REPLACE FUNCTION @extschema@.pg_store_query_plan(

filename TEXT,

query TEXT

)

RETURNS VOID AS 'pg_execplan'

LANGUAGE C;

CREATE OR REPLACE FUNCTION @extschema@.pg_exec_plan(query TEXT,

plan TEXT

)

RETURNS BOOL AS 'pg_execplan'

LANGUAGE C;

CREATE OR REPLACE FUNCTION @extschema@.pg_exec_stored_plan(filename TEXT)

RETURNS BOOL AS 'pg_execplan'

LANGUAGE C;

#include "postgres.h"

#include "access/printtup.h"

#include "commands/extension.h"

#include "commands/prepare.h"

#include "executor/executor.h"

#include "nodes/nodes.h"

#include "nodes/plannodes.h"

#include "tcop/pquery.h"

#include "tcop/utility.h"

#include "utils/builtins.h"

#include "utils/memutils.h"

#include "utils/plancache.h"

#include "utils/snapmgr.h"

#define EXPLAN_DEBUG_LEVEL 0

PG_MODULE_MAGIC;

PG_FUNCTION_INFO_V1(pg_store_query_plan);

PG_FUNCTION_INFO_V1(pg_exec_plan);

PG_FUNCTION_INFO_V1(pg_exec_stored_plan);

void _PG_init(void);

_PG_init(void)

{

return;

}

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;

FILE *fout;

MemoryContext oldcontext;

List *parsetree_list;

RawStmt *parsetree;

List *querytree_list,

*plantree_list;

QueryDesc *queryDesc;

size_t string_len;

if (EXPLAN_DEBUG_LEVEL > 0)

elog(LOG, "Store into %s plan of the query %s.", filename, query_string);

oldcontext = MemoryContextSwitchTo(MessageContext);

parsetree_list = pg_parse_query(query_string);

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);

plantree_list = pg_plan_queries(querytree_list, CURSOR_OPT_PARALLEL_OK, NULL);

queryDesc = CreateQueryDesc((PlannedStmt *) linitial(plantree_list),

query_string,

InvalidSnapshot,

InvalidSnapshot,

None_Receiver,

0,

0);

if (EXPLAN_DEBUG_LEVEL > 0)

elog(INFO, "BEFORE writing %s ...", filename);

fout = fopen(filename, "wb");

Assert(fout != NULL);

string_len = strlen(query_string);

fwrite(&string_len, sizeof(size_t), 1, fout);

fwrite(query_string, sizeof(char), string_len, fout);

set_portable_output(true);

plan_string = nodeToString(queryDesc->plannedstmt);

set_portable_output(false);

string_len = strlen(plan_string);

fwrite(&string_len, sizeof(size_t), 1, fout);

fwrite(plan_string, sizeof(char), string_len, fout);

fclose(fout);

MemoryContextSwitchTo(oldcontext);

PG_RETURN_VOID();

}

static void

exec_plan(char *query_string, char *plan_string)

{

PlannedStmt *pstmt;

ParamListInfo paramLI = NULL;

CachedPlanSource *psrc;

CachedPlan *cplan;

QueryDesc *queryDesc;

DestReceiver *receiver;

int eflags = 0;

PG_TRY();

{

set_portable_input(true);

pstmt = (PlannedStmt *) stringToNode(plan_string);

set_portable_input(false);

}

PG_CATCH();

{

elog(INFO, "BAD PLAN: %s. Query: %s", plan_string, query_string);

PG_RE_THROW();

}

PG_END_TRY();

if (EXPLAN_DEBUG_LEVEL > 0)

elog(INFO, "query: %s\n", query_string);

if (EXPLAN_DEBUG_LEVEL > 1)

elog(INFO, "\nplan: %s\n", plan_string);

psrc = CreateCachedPlan(NULL, query_string, NULL);

CompleteCachedPlan(psrc, NIL, NULL, NULL, 0, NULL, NULL,

CURSOR_OPT_GENERIC_PLAN, false);

SetRemoteSubplan(psrc, pstmt);

cplan = GetCachedPlan(psrc, paramLI, false);

receiver = CreateDestReceiver(DestLog);

PG_TRY();

{

queryDesc = CreateQueryDesc(pstmt,

query_string,

GetActiveSnapshot(),

InvalidSnapshot,

receiver,

paramLI,

0);

ExecutorStart(queryDesc, eflags);

PushActiveSnapshot(queryDesc->snapshot);

ExecutorRun(queryDesc, ForwardScanDirection, 0);

PopActiveSnapshot();

ExecutorFinish(queryDesc);

ExecutorEnd(queryDesc);

FreeQueryDesc(queryDesc);

}

PG_CATCH();

{

elog(INFO, "BAD QUERY: '%s'.", query_string);

ReleaseCachedPlan(cplan, false);

PG_RE_THROW();

}

PG_END_TRY();

receiver->rDestroy(receiver);

ReleaseCachedPlan(cplan, false);

if (EXPLAN_DEBUG_LEVEL > 0)

elog(INFO, "query execution finished.\n");

}

pg_exec_plan(PG_FUNCTION_ARGS)

{

char *query_string = TextDatumGetCString(PG_GETARG_DATUM(0));

char *plan_string = TextDatumGetCString(PG_GETARG_DATUM(1));

char *dec_query,

*dec_plan;

int dec_query_len,

dec_query_len1,

dec_plan_len,

dec_plan_len1;

Assert(query_string != NULL);

Assert(plan_string != NULL);

dec_query_len = b64_dec_len(query_string, strlen(query_string) + 1)+1;

dec_query = palloc0(dec_query_len + 1);

dec_query_len1 = b64_decode(query_string, strlen(query_string), dec_query);

Assert(dec_query_len > dec_query_len1);

dec_plan_len = b64_dec_len(plan_string, strlen(plan_string) + 1);

dec_plan = palloc0(dec_plan_len + 1);

dec_plan_len1 = b64_decode(plan_string, strlen(plan_string), dec_plan);

Assert(dec_plan_len > dec_plan_len1);

exec_plan(dec_query, dec_plan);

pfree(dec_query);

pfree(dec_plan);

PG_RETURN_BOOL(true);

}

static void

LoadPlanFromFile(const char *filename, char **query_string, char **plan_string)

{

FILE *fin;

size_t string_len;

int nelems;

fin = fopen(filename, "rb");

Assert(fin != NULL);

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);

Assert(nelems == string_len);

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);

Assert(nelems == string_len);

fclose(fin);

}

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);

PG_RETURN_BOOL(true);

}

# pg_execplan extension

comment = 'Execute raw query plan on remote node'

default_version = '0.1'

module_pathname = '$libdir/pg_execplan'

relocatable = false

CREATE SCHEMA tests;

SET search_path = 'tests';

CREATE TYPE int42;

CREATE FUNCTION int42_in(cstring)

RETURNS int42

AS 'int4in'

LANGUAGE internal STRICT IMMUTABLE;

CREATE FUNCTION int42_out(int42)

RETURNS cstring

AS 'int4out'

LANGUAGE internal STRICT IMMUTABLE;

CREATE TYPE int42 (

internallength = 4,

input = int42_in,

output = int42_out,

alignment = int4,

default = 42,

passedbyvalue

);

CREATE TABLE tests.t1 (id int42);

CREATE TABLE t2 (id int, payload TEXT, par1 INT);

CREATE FUNCTION select1(tid INT) RETURNS VOID AS $$

INSERT INTO tests.t2 (id, payload, par1) VALUES (1, 'qwe', 2);

END;

$$ LANGUAGE plpgsql;

CREATE COLLATION test1 (locale = 'en_US.utf8');

CREATE TABLE ttest1 (

id serial,

a text COLLATE test1,

b text COLLATE test1

);

INSERT INTO ttest1 (a, b) VALUES ('one', 'one');

INSERT INTO ttest1 (a, b) VALUES ('one', 'two');

CREATE OPERATOR public.### (

leftarg = numeric,

rightarg = numeric,

procedure = numeric_add

);

CREATE TYPE bug_status AS ENUM ('new', 'open', 'closed');

CREATE TABLE public.bug (

id serial,

description TEXT,

status bug_status

);

INSERT INTO public.bug (description, status) VALUES ('abc', 'open');

INSERT INTO public.bug (description, status) VALUES ('abc1', 'closed');

CREATE TABLE public.bug1 (

id serial,

status bug_status

);

INSERT INTO public.bug1 (status) VALUES ('new');

INSERT INTO public.bug1 (status) VALUES ('new');

INSERT INTO public.bug1 (status) VALUES ('closed');