Add overseer node (not working yet

Author: ildus

Makefile

@@ -8,7 +8,7 @@ OBJS = src/init.o src/relation_info.o src/utils.o src/partition_filter.o \
 	src/hooks.o src/nodes_common.o src/xact_handling.o src/utility_stmt_hooking.o \
 	src/planner_tree_modification.o src/debug_print.o src/partition_creation.o \
 	src/compat/pg_compat.o src/compat/rowmarks_fix.o src/partition_router.o \
-	$(WIN32RES)
+	src/partition_overseer.o $(WIN32RES)
 
 ifdef USE_PGXS
 override PG_CPPFLAGS += -I$(CURDIR)/src/include

src/hooks.c

@@ -68,7 +68,6 @@ planner_hook_type				pathman_planner_hook_next				= NULL;
 post_parse_analyze_hook_type	pathman_post_parse_analyze_hook_next	= NULL;
 shmem_startup_hook_type			pathman_shmem_startup_hook_next			= NULL;
 ProcessUtility_hook_type		pathman_process_utility_hook_next		= NULL;
-ExecutorRun_hook_type			pathman_executor_run_hook_next			= NULL;
 
 
 /* Take care of joins */
@@ -616,6 +615,29 @@ pathman_enable_assign_hook(bool newval, void *extra)
 		 newval ? "enabled" : "disabled");
 }
 
+static void
+execute_for_plantree(PlannedStmt *planned_stmt,
+					 Plan *(*proc) (List *rtable, Plan *plan))
+{
+	List		*subplans = NIL;
+	ListCell	*lc;
+	Plan		*resplan = proc(planned_stmt->rtable, planned_stmt->planTree);
+
+	if (resplan)
+		planned_stmt->planTree = resplan;
+
+	foreach (lc, planned_stmt->subplans)
+	{
+		Plan	*subplan = lfirst(lc);
+		resplan = proc(planned_stmt->rtable, (Plan *) lfirst(lc));
+		if (resplan)
+			subplans = lappend(subplans, resplan);
+		else
+			subplans = lappend(subplans, subplan);
+	}
+	planned_stmt->subplans = subplans;
+}
+
 /*
  * Planner hook. It disables inheritance for tables that have been partitioned
  * by pathman to prevent standart PostgreSQL partitioning mechanism from
@@ -624,14 +646,6 @@ pathman_enable_assign_hook(bool newval, void *extra)
 PlannedStmt *
 pathman_planner_hook(Query *parse, int cursorOptions, ParamListInfo boundParams)
 {
-#define ExecuteForPlanTree(planned_stmt, proc) \
-	do { \
-		ListCell *lc; \
-		proc((planned_stmt)->rtable, (planned_stmt)->planTree); \
-		foreach (lc, (planned_stmt)->subplans) \
-			proc((planned_stmt)->rtable, (Plan *) lfirst(lc)); \
-	} while (0)
-
 	PlannedStmt	   *result;
 	uint32			query_id = parse->queryId;
 
@@ -658,10 +672,10 @@ pathman_planner_hook(Query *parse, int cursorOptions, ParamListInfo boundParams)
 		if (pathman_ready)
 		{
 			/* Add PartitionFilter node for INSERT queries */
-			ExecuteForPlanTree(result, add_partition_filters);
+			execute_for_plantree(result, add_partition_filters);
 
 			/* Add PartitionRouter node for UPDATE queries */
-			ExecuteForPlanTree(result, add_partition_routers);
+			execute_for_plantree(result, add_partition_routers);
 
 			/* Decrement planner() calls count */
 			decr_planner_calls_count();
@@ -686,7 +700,6 @@ pathman_planner_hook(Query *parse, int cursorOptions, ParamListInfo boundParams)
 
 	/* Finally return the Plan */
 	return result;
-#undef ExecuteForPlanTree
 }
 
 /*
@@ -950,40 +963,3 @@ pathman_process_utility_hook(Node *first_arg,
 								context, params, queryEnv,
 								dest, completionTag);
 }
-
-/*
- * Executor hook (for PartitionRouter).
- */
-#if PG_VERSION_NUM >= 100000
-void
-pathman_executor_hook(QueryDesc *queryDesc,
-					  ScanDirection direction,
-					  ExecutorRun_CountArgType count,
-					  bool execute_once)
-#else
-void
-pathman_executor_hook(QueryDesc *queryDesc,
-					  ScanDirection direction,
-					  ExecutorRun_CountArgType count)
-#endif
-{
-#define EXECUTOR_HOOK				pathman_executor_run_hook_next
-#if PG_VERSION_NUM >= 100000
-#define EXECUTOR_HOOK_NEXT(q,d,c)	EXECUTOR_HOOK((q),(d),(c), execute_once)
-#define EXECUTOR_RUN(q,d,c)			standard_ExecutorRun((q),(d),(c), execute_once)
-#else
-#define EXECUTOR_HOOK_NEXT(q,d,c)	EXECUTOR_HOOK((q),(d),(c))
-#define EXECUTOR_RUN(q,d,c)			standard_ExecutorRun((q),(d),(c))
-#endif
-
-	/* Prepare ModifyTable nodes for PartitionRouter hackery */
-	state_tree_visitor((PlanState *) queryDesc->planstate,
-					   prepare_modify_table_for_partition_router,
-					   NULL);
-
-	/* Call hooks set by other extensions if needed */
-	if (EXECUTOR_HOOK)
-		EXECUTOR_HOOK_NEXT(queryDesc, direction, count);
-	/* Else call internal implementation */
-	else EXECUTOR_RUN(queryDesc, direction, count);
-}

src/include/partition_overseer.h

@@ -0,0 +1,54 @@
+/* ------------------------------------------------------------------------
+ *
+ * partition_overseer.h
+ *		Restart ModifyTable for unobvious reasons
+ *
+ * Copyright (c) 2018, Postgres Professional
+ *
+ * ------------------------------------------------------------------------
+ */
+
+#ifndef PARTITION_OVERSEER_H
+#define PARTITION_OVERSEER_H
+
+#include "relation_info.h"
+#include "utils.h"
+
+#include "postgres.h"
+#include "access/tupconvert.h"
+#include "commands/explain.h"
+#include "optimizer/planner.h"
+
+#if PG_VERSION_NUM >= 90600
+#include "nodes/extensible.h"
+#endif
+
+
+#define OVERSEER_NODE_NAME "PartitionOverseer"
+
+
+extern CustomScanMethods	partition_overseer_plan_methods;
+extern CustomExecMethods	partition_overseer_exec_methods;
+
+
+void init_partition_overseer_static_data(void);
+Plan *make_partition_overseer(Plan *subplan);
+
+Node *partition_overseer_create_scan_state(CustomScan *node);
+
+void partition_overseer_begin(CustomScanState *node,
+							  EState *estate,
+							  int eflags);
+
+TupleTableSlot *partition_overseer_exec(CustomScanState *node);
+
+void partition_overseer_end(CustomScanState *node);
+
+void partition_overseer_rescan(CustomScanState *node);
+
+void partition_overseer_explain(CustomScanState *node,
+								List *ancestors,
+								ExplainState *es);
+
+
+#endif /* PARTITION_OVERSEER_H */

src/include/partition_router.h

@@ -74,12 +74,7 @@ extern CustomExecMethods	partition_router_exec_methods;
 
 void init_partition_router_static_data(void);
 
-
-Plan *make_partition_router(Plan *subplan,
-							Oid parent_relid,
-							Index parent_rti,
-							int epq_param,
-							List *returning_list);
+Plan *make_partition_router(Plan *subplan, int epq_param);
 
 void prepare_modify_table_for_partition_router(PlanState *state, void *context);
 
@@ -98,5 +93,6 @@ void partition_router_explain(CustomScanState *node,
 							  List *ancestors,
 							  ExplainState *es);
 
+TupleTableSlot *partition_router_run_modify_table(PlanState *state);
 
 #endif /* PARTITION_UPDATE_H */

src/include/planner_tree_modification.h

@@ -25,8 +25,8 @@ void assign_query_id(Query *query);
 void reset_query_id_generator(void);
 
 /* Plan tree rewriting utility */
-void plan_tree_visitor(Plan *plan,
-					   void (*visitor) (Plan *plan, void *context),
+Plan * plan_tree_visitor(Plan *plan,
+					   Plan *(*visitor) (Plan *plan, void *context),
 					   void *context);
 
 /* PlanState tree rewriting utility */
@@ -38,8 +38,8 @@ void state_tree_visitor(PlanState *state,
 void pathman_transform_query(Query *parse, ParamListInfo params);
 
 /* These functions scribble on Plan tree */
-void add_partition_filters(List *rtable, Plan *plan);
-void add_partition_routers(List *rtable, Plan *plan);
+Plan *add_partition_filters(List *rtable, Plan *plan);
+Plan *add_partition_routers(List *rtable, Plan *plan);
 
 
 /* used by assign_rel_parenthood_status() etc */

src/partition_overseer.c

@@ -0,0 +1,105 @@
+#include "postgres.h"
+
+#include "partition_overseer.h"
+#include "partition_filter.h"
+#include "partition_router.h"
+
+CustomScanMethods	partition_overseer_plan_methods;
+CustomExecMethods	partition_overseer_exec_methods;
+
+void
+init_partition_overseer_static_data(void)
+{
+	partition_overseer_plan_methods.CustomName 			= OVERSEER_NODE_NAME;
+	partition_overseer_plan_methods.CreateCustomScanState = partition_overseer_create_scan_state;
+
+	partition_overseer_exec_methods.CustomName			= OVERSEER_NODE_NAME;
+	partition_overseer_exec_methods.BeginCustomScan		= partition_overseer_begin;
+	partition_overseer_exec_methods.ExecCustomScan		= partition_overseer_exec;
+	partition_overseer_exec_methods.EndCustomScan		= partition_overseer_end;
+	partition_overseer_exec_methods.ReScanCustomScan	= partition_overseer_rescan;
+	partition_overseer_exec_methods.MarkPosCustomScan	= NULL;
+	partition_overseer_exec_methods.RestrPosCustomScan	= NULL;
+	partition_overseer_exec_methods.ExplainCustomScan	= partition_overseer_explain;
+
+	RegisterCustomScanMethods(&partition_overseer_plan_methods);
+}
+
+Plan *
+make_partition_overseer(Plan *subplan)
+{
+	CustomScan *cscan = makeNode(CustomScan);
+
+	/* Copy costs etc */
+	cscan->scan.plan.startup_cost	= subplan->startup_cost;
+	cscan->scan.plan.total_cost		= subplan->total_cost;
+	cscan->scan.plan.plan_rows		= subplan->plan_rows;
+	cscan->scan.plan.plan_width		= subplan->plan_width;
+
+	/* Setup methods, child plan and param number for EPQ */
+	cscan->methods = &partition_overseer_plan_methods;
+	cscan->custom_plans = list_make1(subplan);
+	cscan->custom_private = NIL;
+
+	/* No physical relation will be scanned */
+	cscan->scan.scanrelid = 0;
+
+	/* Build an appropriate target list */
+	cscan->scan.plan.targetlist = pfilter_build_tlist(subplan);
+	cscan->custom_scan_tlist = subplan->targetlist;
+
+	return &cscan->scan.plan;
+}
+
+
+Node *
+partition_overseer_create_scan_state(CustomScan *node)
+{
+	CustomScanState *state = palloc0(sizeof(CustomScanState));
+	NodeSetTag(state, T_CustomScanState);
+
+	state->flags = node->flags;
+	state->methods = &partition_overseer_exec_methods;
+
+	return (Node *) state;
+}
+
+void
+partition_overseer_begin(CustomScanState *node,
+						 EState *estate,
+						 int eflags)
+{
+	CustomScan *css = (CustomScan *) node->ss.ps.plan;
+	Plan	   *plan = linitial(css->custom_plans);
+
+	/* It's convenient to store PlanState in 'custom_ps' */
+	node->custom_ps = list_make1(ExecInitNode(plan, estate, eflags));
+}
+
+TupleTableSlot *
+partition_overseer_exec(CustomScanState *node)
+{
+	PlanState *state = linitial(node->custom_ps);
+	return partition_router_run_modify_table(state);
+}
+
+void
+partition_overseer_end(CustomScanState *node)
+{
+	Assert(list_length(node->custom_ps) == 1);
+	ExecEndNode((PlanState *) linitial(node->custom_ps));
+}
+
+void
+partition_overseer_rescan(CustomScanState *node)
+{
+	elog(ERROR, "partition_overseer_rescan is not implemented");
+}
+
+void
+partition_overseer_explain(CustomScanState *node,
+						   List *ancestors,
+						   ExplainState *es)
+{
+	/* nothing to do */
+}