Cost model for EXCHANGE and DistExec nodes.

Author: danolivo

contrib/pg_exchange/exchange.c

@@ -37,7 +37,13 @@
 #include "partutils.h"
 #include "stream.h"
 
-#define DEFAULT_EXCHANGE_STARTUP_COST	100.0
+/*
+ * Startup cost of EXCHANGE node is smaller, than DistExec node, because cost
+ * of DistExec node contains cost of query transfer and localization at each
+ * instance. Startup cost of EXCHANGE node includes only connection channel
+ * establishing between instances.
+ */
+#define DEFAULT_EXCHANGE_STARTUP_COST	10.0
 #define DEFAULT_TRANSFER_TUPLE_COST		0.01
 
 
@@ -111,6 +117,7 @@ static void create_gather_dfn(EPPNode *epp, RelOptInfo *rel);
 static void create_stealth_dfn(EPPNode *epp, RelOptInfo *rel, PlannerInfo *root);
 static void create_shuffle_dfn(EPPNode *epp, RelOptInfo *rel, PlannerInfo *root);
 static void create_broadcast_dfn(EPPNode *epp, RelOptInfo *rel, PlannerInfo *root);
+static void force_add_path(RelOptInfo *rel, Path *path);
 
 
 /*
@@ -400,42 +407,52 @@ add_exchange_paths(PlannerInfo *root, RelOptInfo *rel, Index rti, RangeTblEntry
 		set_exchange_altrel(EXCH_GATHER, (ExchangePath *) path, rel, NULL, NULL,
 																	servers);
 		path = (Path *) create_distexec_path(root, rel, path, servers);
-		add_path(rel, path);
+
+		force_add_path(rel, path);
 	}
 }
 
+static void
+force_add_path(RelOptInfo *rel, Path *path)
+{
+	List *pathlist = rel->pathlist;
+
+	rel->pathlist = NIL;
+	rel->cheapest_parameterized_paths = NIL;
+	rel->cheapest_startup_path = rel->cheapest_total_path =
+											rel->cheapest_unique_path = NULL;
+	add_path(rel, path);
+	rel->pathlist = list_concat(rel->pathlist, pathlist);
+	set_cheapest(rel);
+}
+
 #include "optimizer/cost.h"
 
-static void
+void
 cost_exchange(PlannerInfo *root, RelOptInfo *baserel, ExchangePath *expath)
 {
 	Path *subpath;
 
-	if (baserel->pages == 0 && baserel->tuples == 0)
-	{
-		baserel->pages = 10;
-		baserel->tuples =
-				(10 * BLCKSZ) / (baserel->reltarget->width +
-									 MAXALIGN(SizeofHeapTupleHeader));
-	}
-
 	/* Estimate baserel size as best we can with local statistics. */
-//	set_baserel_size_estimates(root, baserel);
 	subpath = cstmSubPath1(expath);
-	expath->cp.path.rows = baserel->tuples;
-	expath->cp.path.startup_cost = subpath->startup_cost;
+	expath->cp.path.rows = subpath->rows;
+	expath->cp.path.startup_cost = 0.;
+	expath->cp.path.total_cost = subpath->total_cost;
+
 	switch (expath->mode)
 	{
 	case EXCH_GATHER:
 		expath->cp.path.startup_cost += DEFAULT_EXCHANGE_STARTUP_COST;
-		expath->cp.path.rows = subpath->rows * expath->altrel.nparts;
+		expath->cp.path.total_cost += cpu_tuple_cost * expath->cp.path.rows;
 		break;
 	case EXCH_STEALTH:
-		expath->cp.path.rows = subpath->rows;
+		expath->cp.path.startup_cost = 0.;
+		expath->cp.path.total_cost += cpu_tuple_cost * expath->cp.path.rows /
+														expath->altrel.nparts;
 		break;
 	case EXCH_BROADCAST:
 		expath->cp.path.startup_cost += DEFAULT_EXCHANGE_STARTUP_COST;
-		expath->cp.path.rows = subpath->rows * expath->altrel.nparts;
+		expath->cp.path.total_cost += cpu_tuple_cost * expath->cp.path.rows;
 		break;
 	case EXCH_SHUFFLE:
 	{
@@ -446,15 +463,14 @@ cost_exchange(PlannerInfo *root, RelOptInfo *baserel, ExchangePath *expath)
 		Path *path = &expath->cp.path;
 
 		path->startup_cost += DEFAULT_EXCHANGE_STARTUP_COST;
-		path->total_cost += path->startup_cost;
 
 		/*
-		 * We count on perfect balance of tuple distribution:
-		 * If we have N instances, M tuples from subtree, than we send up to
-		 * local subtree M/N tuples, send to network [M-M/N] tuples and same to
+		 * We assume perfect balance of tuple distribution:
+		 * If we have N instances, M tuples from subtree, than we send up by
+		 * subtree M/N local tuples, send to network [M-M/N] tuples and same to
 		 * receive.
 		 */
-		path->rows = subpath->rows;
+		path->rows /= expath->altrel.nparts;
 		instances = expath->altrel.nparts > 0 ? expath->altrel.nparts : 2;
 		send_rows = path->rows - (path->rows/instances);
 		received_rows = send_rows;
@@ -466,8 +482,7 @@ cost_exchange(PlannerInfo *root, RelOptInfo *baserel, ExchangePath *expath)
 	default:
 		elog(FATAL, "Unknown EXCHANGE mode.");
 	}
-
-	expath->cp.path.total_cost = 0.1;
+	expath->cp.path.total_cost += expath->cp.path.startup_cost;
 }
 
 /*
@@ -493,13 +508,6 @@ ExchangePlanCustomPath(PlannerInfo *root,
 	exchange = make_exchange(custom_plans, tlist);
 	private->node.extnodename = EXCHANGE_PRIVATE_NAME;
 
-	exchange->scan.plan.startup_cost = best_path->path.startup_cost;
-	exchange->scan.plan.total_cost = best_path->path.total_cost;
-	exchange->scan.plan.plan_rows = best_path->path.rows;
-	exchange->scan.plan.plan_width = best_path->path.pathtarget->width;
-	exchange->scan.plan.parallel_aware = best_path->path.parallel_aware;
-	exchange->scan.plan.parallel_safe = best_path->path.parallel_safe;
-
 	/* Add stream name into private field*/
 	GetMyServerName(&host, &port);
 	sprintf(streamName, "%s-%d-%d", host, port, exchange_counter++);
@@ -791,15 +799,9 @@ make_exchange(List *custom_plans, List *tlist)
 	Plan		*plan = &node->scan.plan;
 	List *child_tlist;
 
-	plan->startup_cost = 1;
-	plan->total_cost = 1;
-	plan->plan_rows = 1;
-	plan->plan_width =1;
 	plan->qual = NIL;
 	plan->lefttree = NULL;
 	plan->righttree = NULL;
-	plan->parallel_aware = false; /* Use Shared Memory in parallel worker */
-	plan->parallel_safe = false;
 	plan->targetlist = tlist;
 
 	/* Setup methods and child plan */
@@ -1026,7 +1028,6 @@ EXCHANGE_Execute(CustomScanState *node)
 		else
 		{
 			state->stuples++;
-//			elog(LOG, "SEND TUPLE to stream [%s]", state->stream);
 			SendTuple(dest, state->stream, slot, false);
 		}
 	}
@@ -1043,8 +1044,6 @@ EXCHANGE_End(CustomScanState *node)
 
 	if (state->mode != EXCH_STEALTH)
 		Stream_unsubscribe(state->stream);
-
-//	elog(INFO, "EXCHANGE_END");
 }
 
 static void

contrib/pg_exchange/exchange.h

@@ -106,5 +106,7 @@ extern CustomScan *make_exchange(List *custom_plans, List *tlist);
 extern ExchangePath *create_exchange_path(PlannerInfo *root, RelOptInfo *rel,
 		  	  	  	  	  	  	  	  	  	 Path *children, ExchangeMode mode);
 extern void createNodeName(char *nodeName, const char *hostname, int port);
+extern void cost_exchange(PlannerInfo *root, RelOptInfo *baserel,
+														ExchangePath *expath);
 
 #endif /* EXCHANGE_H_ */

contrib/pg_exchange/hooks.c

@@ -179,6 +179,7 @@ create_distributed_join_paths(PlannerInfo *root, RelOptInfo *joinrel,
 			reset_cheapest(innerrel);
 			add_paths_to_joinrel(root, joinrel, outerrel, innerrel, jointype,
 											extra->sjinfo, extra->restrictlist);
+
 			set_cheapest(joinrel);
 			Assert(!join_path_contains_distexec(joinrel->cheapest_total_path));
 			Assert(joinrel->cheapest_total_path);
@@ -195,6 +196,8 @@ create_distributed_join_paths(PlannerInfo *root, RelOptInfo *joinrel,
 									bms_union(inner_servers, outer_servers));
 			set_exchange_altrel(inner_mode, inn_child, rel2, NULL, NULL,
 									bms_union(inner_servers, outer_servers));
+			cost_exchange(root, outerrel, out_child);
+			cost_exchange(root, innerrel, inn_child);
 
 			/* Check for special case */
 			if (path->pathtype == T_NestLoop)
@@ -209,7 +212,7 @@ create_distributed_join_paths(PlannerInfo *root, RelOptInfo *joinrel,
 				set_exchange_altrel(EXCH_BROADCAST, out_child, NULL,
 						&((ExchangePath *)cstmSubPath1(out_child))->altrel,
 						NIL, bms_union(inner_servers, outer_servers));
-
+				cost_exchange(root, joinrel, out_child);
 				inn_child->mode = EXCH_STEALTH;
 			}
 			else if (inner_mode == EXCH_SHUFFLE)
@@ -225,6 +228,7 @@ create_distributed_join_paths(PlannerInfo *root, RelOptInfo *joinrel,
 			set_exchange_altrel(EXCH_GATHER, gather, &out_child->altrel,
 							&inn_child->altrel, extra->restrictlist,
 							bms_union(inner_servers, outer_servers));
+			cost_exchange(root, joinrel, gather);
 			Assert(gather->altrel.part_scheme != NULL);
 			path = (Path *) create_distexec_path(root, joinrel, (Path *) gather,
 									bms_union(inner_servers, outer_servers));
@@ -259,7 +263,6 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 {
 	ListCell *lc;
 	List *delpaths = NIL;
-	List *dist_paths = NIL;
 	List *pathlist = NIL;
 
 	if (recursion)
@@ -289,35 +292,21 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 	foreach(lc, delpaths)
 		joinrel->pathlist = list_delete_ptr(joinrel->pathlist, lfirst(lc));
 
-	/* Add trivial paths */
-	dist_paths = create_distributed_join_paths(root, joinrel, outerrel, innerrel,
-									jointype, extra, EXCH_GATHER, EXCH_GATHER);
-	foreach(lc, dist_paths)
-	{
-		Path *path = lfirst(lc);
-		add_path(joinrel, path);
-	}
-	pathlist = create_distributed_join_paths(root, joinrel, outerrel, innerrel,
-								jointype, extra, EXCH_BROADCAST, EXCH_STEALTH);
-	foreach(lc, pathlist)
-	{
-		Path *path = lfirst(lc);
-		path->total_cost -= 0.1;
-		add_path(joinrel, path);
-	}
-	list_free(pathlist);
-	pathlist = NIL;
+	/* Add distributed paths */
+	pathlist = list_concat(pathlist, create_distributed_join_paths(root, joinrel,
+					   outerrel, innerrel, jointype, extra, EXCH_GATHER,
+					   EXCH_GATHER));
+
+	pathlist = list_concat(pathlist, create_distributed_join_paths(root, joinrel,
+					   outerrel, innerrel, jointype, extra, EXCH_BROADCAST,
+					   EXCH_STEALTH));
+
+	pathlist = list_concat(pathlist, create_distributed_join_paths(root, joinrel,
+					   outerrel, innerrel, jointype, extra, EXCH_SHUFFLE,
+					   EXCH_SHUFFLE));
 
-	pathlist = create_distributed_join_paths(root, joinrel, outerrel, innerrel,
-								jointype, extra, EXCH_SHUFFLE, EXCH_SHUFFLE);
 	foreach(lc, pathlist)
-	{
-		Path *path = lfirst(lc);
-		path->total_cost -= 0.2;
-		add_path(joinrel, path);
-	}
-	list_free(pathlist);
-	pathlist = NIL;
+		add_path(joinrel, lfirst(lc));
 
 	recursion = false;
 }
@@ -499,4 +488,3 @@ EXEC_Hooks_init(void)
 	PreviousShmemStartupHook = shmem_startup_hook;
 	shmem_startup_hook = HOOK_shmem_startup;
 }
-

contrib/pg_exchange/nodeDistPlanExec.c

@@ -395,14 +395,6 @@ CreateDistExecPlan(PlannerInfo *root,
 	CustomScan *distExecNode;
 
 	distExecNode = make_distplanexec(custom_plans, tlist, best_path->custom_private);
-
-	distExecNode->scan.plan.startup_cost = best_path->path.startup_cost;
-	distExecNode->scan.plan.total_cost = best_path->path.total_cost;
-	distExecNode->scan.plan.plan_rows = best_path->path.rows;
-	distExecNode->scan.plan.plan_width = best_path->path.pathtarget->width;
-	distExecNode->scan.plan.parallel_aware = best_path->path.parallel_aware;
-	distExecNode->scan.plan.parallel_safe = best_path->path.parallel_safe;
-
 	return &distExecNode->scan.plan;
 }
 
@@ -442,15 +434,9 @@ make_distplanexec(List *custom_plans, List *tlist, List *private_data)
 	ListCell	*lc;
 	List *child_tlist;
 
-	plan->startup_cost = 10;
-	plan->total_cost = 10;
-	plan->plan_rows = 1000;
-	plan->plan_width =10;
 	plan->qual = NIL;
 	plan->lefttree = NULL;
 	plan->righttree = NULL;
-	plan->parallel_aware = false;
-	plan->parallel_safe = false;
 	plan->targetlist = tlist;
 
 	/* Setup methods and child plan */
@@ -502,8 +488,8 @@ create_distexec_path(PlannerInfo *root, RelOptInfo *rel, Path *children,
 	path->methods = &distplanexec_path_methods;
 
 	pathnode->rows = children->rows;
-	pathnode->startup_cost = 100.;
-	pathnode->total_cost = pathnode->startup_cost;
+	pathnode->startup_cost = children->startup_cost + 100.;
+	pathnode->total_cost = children->total_cost + pathnode->startup_cost;
 
 	return path;
 }