JOIN cascades support.

danolivo · danolivo · commit 72ba98698f5c · 2019-04-03T11:39:58.000+05:00
I tested this on the simple queries:
1. select * from rt, pt, st WHERE rt.id=pt.id and pt.id=st.id;
2. select * from rt, pt, st WHERE rt.id=pt.id and pt.id=st.payload;
diff --git a/contrib/pg_exchange/exchange.c b/contrib/pg_exchange/exchange.c
@@ -210,9 +210,7 @@ add_exchange_paths(PlannerInfo *root, RelOptInfo *rel, Index rti, RangeTblEntry
 	ListCell   *lc;
 
 	if (!rte->inh)
-		/*
-		 * Relation is not contain any partitions.
-		 */
+		/* Relation is not contain any partitions. */
 		return;
 
 	/* Traverse all possible paths and search for APPEND */
@@ -223,16 +221,31 @@ add_exchange_paths(PlannerInfo *root, RelOptInfo *rel, Index rti, RangeTblEntry
 		AppendPath	*appendPath = NULL;
 		ListCell	*lc1;
 		Bitmapset	*servers = NULL;
-		List	*subpaths = NIL;
+		List		*subpaths = NIL;
+		List		*append_paths;
 
-		if (path->pathtype != T_Append)
-			continue;
+		/*
+		 * In the case of partitioned relation all paths will be ended by Append
+		 * or MergeAppend path node.
+		 */
+		switch (path->pathtype)
+		{
+		case T_Append:
+			append_paths = ((AppendPath *) path)->subpaths;
+			break;
+		case T_MergeAppend:
+			append_paths = ((MergeAppendPath *) path)->subpaths;
+			break;
+		default:
+			elog(FATAL, "Unexpected node type %d, pathtype %d", path->type,
+																path->pathtype);
+		}
 
 		/*
 		 * Traverse all APPEND subpaths, check for scan-type and search for
 		 * foreign scans
 		 */
-		foreach(lc1, ((AppendPath *) path)->subpaths)
+		foreach(lc1, append_paths)
 		{
 			Path	*subpath = (Path *) lfirst(lc1);
 			Path	*tmpPath;
@@ -279,6 +292,7 @@ add_exchange_paths(PlannerInfo *root, RelOptInfo *rel, Index rti, RangeTblEntry
 								((AppendPath *) path)->partitioned_rels, -1);
 		path = (Path *) create_exchange_path(root, rel, (Path *) appendPath, true);
 		path = create_distexec_path(root, rel, path, servers);
+//		elog(LOG, "Path added");
 		add_path(rel, path);
 	}
 }
@@ -740,6 +754,8 @@ EXCHANGE_Execute(CustomScanState *node)
 				return slot;
 			case 1:
 				state->activeRemotes--;
+				elog(LOG, "[%s] GOT NULL. activeRemotes: %d", state->stream,
+						state->activeRemotes);
 				break;
 			case 2: /* Close EXCHANGE channel */
 				break;
diff --git a/contrib/pg_exchange/hooks.c b/contrib/pg_exchange/hooks.c
@@ -23,7 +23,7 @@
 static set_rel_pathlist_hook_type	prev_set_rel_pathlist_hook = NULL;
 static shmem_startup_hook_type PreviousShmemStartupHook = NULL;
 static set_join_pathlist_hook_type prev_set_join_pathlist_hook = NULL;
-static void second_stage_paths(PlannerInfo *root, List *firstStagePaths, RelOptInfo *joinrel,
+static void second_stage_paths(PlannerInfo *root, RelOptInfo *joinrel,
 							   RelOptInfo *outerrel, RelOptInfo *innerrel,
 							   JoinType jointype, JoinPathExtraData *extra);
 
@@ -95,15 +95,56 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 				   JoinPathExtraData *extra)
 {
 	ListCell *lc;
-	List *firstStagePaths = NIL; /* Trivial paths, made with exchange */
+	int i=0;
+	List *newpaths = NIL;
+	List *delpaths = NIL;
+	List *rpaths = NIL;
+	int distpaths = 0;
 
 	if (prev_set_join_pathlist_hook)
 		prev_set_join_pathlist_hook(root, joinrel, outerrel, innerrel,
 									jointype, extra);
 
 	/*
-	 * At first, traverse all paths and search for the case with Exchanges at
-	 * the left or right subtree. We need to delete DistPlanExec nodes and
+	 * Search for distributed paths for children relations.
+	 */
+	foreach(lc, innerrel->pathlist)
+	{
+		Path *path = lfirst(lc);
+
+		if (IsDistExecNode(path))
+			distpaths++;
+	}
+
+	if (distpaths == 0)
+		/* Distributed query execution does not needed. */
+		return;
+
+	/*
+	 * Force try to add hash joins into the pathlist. Collect any paths that
+	 * do not satisfy the exchange rules and delete it.
+	 */
+	foreach(lc, joinrel->pathlist)
+	{
+		Path *path = lfirst(lc);
+		if ((path->pathtype != T_HashJoin) && !IsDistExecNode(path))
+			rpaths = lappend(rpaths, path);
+	}
+	foreach(lc, rpaths)
+		joinrel->pathlist = list_delete_ptr(joinrel->pathlist, lfirst(lc));
+
+	/*
+	 * Try to create hash join path.
+	 * XXX: Here we have a problem of path additions: hash join may be not added
+	 * if it is not cheap. But cost of the this path includes costs of child paths.
+	 * Child paths will be tuned below and costs will be changed too.
+	 */
+	hash_inner_and_outer(root, joinrel, outerrel, innerrel, jointype, extra);
+	Assert(list_length(joinrel->pathlist) > 0);
+
+	/*
+	 * Traverse all paths and search for the case with EXCHANGE nodes
+	 * at the left or right subtree. We need to delete DistPlanExec nodes and
 	 * insert only one at the head of join.
 	 */
 	foreach(lc, joinrel->pathlist)
@@ -115,9 +156,15 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 		CustomPath *sub;
 		Path *path = lfirst(lc);
 
-		if ((path->pathtype != T_NestLoop) &&
-			(path->pathtype != T_MergeJoin) &&
-			(path->pathtype != T_HashJoin))
+		/*
+		 * NestLoop and MergeJoin need to change EXCHANGE node logic and
+		 * disabled for now.
+		 * For this we need to introduce remote PUSH or PULL operation for
+		 * force transfer tuples from instance to instance.
+		 */
+		Assert(path->pathtype != T_NestLoop && path->pathtype != T_MergeJoin);
+
+		if (path->pathtype != T_HashJoin)
 			continue;
 
 		jp = (JoinPath *) path;
@@ -128,9 +175,7 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 		 * If inner path contains DistExec node - save its servers list and
 		 * delete it from the path.
 		 */
-		if ((inner->pathtype == T_CustomScan) &&
-			(strcmp(((CustomPath *) inner)->methods->CustomName,
-														DISTEXECPATHNAME) == 0))
+		if (IsDistExecNode(inner))
 		{
 			ListCell *lc;
 
@@ -145,15 +190,19 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 			}
 			Assert(list_length(sub->custom_paths) == 1);
 			jp->innerjoinpath = (Path *) linitial(sub->custom_paths);
+		} else
+		{
+			elog(LOG, "inner Path can't contain any FE nodes. JT=%d (innt=%d patht=%d), (outt=%d patht=%d) %d",
+					jp->jointype, inner->type, inner->pathtype,
+					outer->type, outer->pathtype, i++);
+
 		}
 
 		/*
 		 * If outer path contains DistExec node - save its servers list and
 		 * delete it from the path.
 		 */
-		if ((outer->pathtype == T_CustomScan) &&
-			(strcmp(((CustomPath *) outer)->methods->CustomName,
-														DISTEXECPATHNAME) == 0))
+		if (IsDistExecNode(outer))
 		{
 			ListCell *lc;
 
@@ -177,25 +226,29 @@ HOOK_Join_pathlist(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 		path = create_distexec_path(root, joinrel,
 									(Path *) copy_join_pathnode(jp),
 									servers);
-		add_path(joinrel, path);
+		newpaths = lappend(newpaths, path);
+		delpaths = lappend(delpaths, jp);
+	}
 
-		/*
-		 * We need guarantee, that previous JOIN path was deleted. It was
-		 * incorrect.
-		 */
-		list_delete_ptr(joinrel->pathlist, jp);
+	/*
+	 * We need to guarantee, that previous JOIN path was deleted from the path
+	 * list. It was incorrect.
+	 */
+	foreach(lc, delpaths)
+	{
+		Path *path = lfirst(lc);
+		joinrel->pathlist = list_delete_ptr(joinrel->pathlist, path);
+	}
 
-		/* Save link to the path for future works. */
-		firstStagePaths = lappend(firstStagePaths, path);
+	foreach(lc, newpaths)
+	{
+		Path *path = lfirst(lc);
+		add_path(joinrel, path);
 	}
 
-	second_stage_paths(root, firstStagePaths, joinrel, outerrel, innerrel, jointype,
-			   extra);
+	second_stage_paths(root, joinrel, outerrel, innerrel, jointype, extra);
 }
 
-#define IsDistExecNode(pathnode) ((pathnode->path.pathtype == T_CustomScan) && \
-	(strcmp(((CustomPath *)pathnode)->methods->CustomName, DISTEXECPATHNAME) == 0))
-
 static CustomPath *
 duplicate_join_path(CustomPath	*distExecPath)
 {
@@ -321,10 +374,7 @@ arrange_partitioning_attrs(RelOptInfo *rel1,
 		part_scheme->partnatts++;
 		ReleaseSysCache(opclasstup);
 	}
-//elog(INFO, "arrange_partitioning_attrs: ");
-//elog(INFO, "->1: %s ", nodeToString(rel1->partexprs[0]));
-//elog(INFO, "->2: %s ", nodeToString(rel2->partexprs[0]));
-//elog(INFO, "restrictlist: %s", nodeToString(restrictlist));
+
 	/* Now we use hash partition only */
 	Assert((rel1->part_scheme->strategy == PARTITION_STRATEGY_HASH) &&
 			(rel1->part_scheme->strategy == rel2->part_scheme->strategy));
@@ -394,24 +444,38 @@ arrange_partitions(RelOptInfo *rel1,
  * Add Paths same as the case of partitionwise join.
  */
 static void
-second_stage_paths(PlannerInfo *root, List *firstStagePaths, RelOptInfo *joinrel, RelOptInfo *outerrel,
+second_stage_paths(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel,
 	 	   RelOptInfo *innerrel, JoinType jointype, JoinPathExtraData *extra)
 {
 	ListCell *lc;
 
-	if (list_length(firstStagePaths) == 0)
-		return;
+	elog(LOG, "List length=%d", list_length(joinrel->pathlist));
+	foreach(lc, joinrel->pathlist)
+	{
+		Path *path = lfirst(lc);
+		elog(LOG, "[%d] BEFORE SECOND STAGE: type=%d pathtype=%d",
+				list_length(joinrel->pathlist), path->type, path->pathtype);
+		if (path->type > 300)
+			Assert(0);
+	}
 
-	foreach(lc, firstStagePaths)
+	foreach(lc, joinrel->pathlist)
 	{
-		CustomPath *path = (CustomPath *) lfirst(lc);
+		Path *pathhead = (Path *) lfirst(lc);
+		CustomPath *path;
 		JoinPath *jp;
 		ExchangePath *innerex;
 		ExchangePath *outerex;
 		ExchangePath *expath;
 		int i;
 
-		Assert(IsDistExecNode(path));
+		if (!IsDistExecNode(pathhead))
+		{
+			elog(LOG, "NO second_stage_paths. type=%d, pathtype=%d", pathhead->type, pathhead->pathtype);
+			continue;
+		}
+
+		path = (CustomPath *) pathhead;
 
 		/*
 		 * Add gather-type EXCHANGE node into the head of the path.
@@ -420,7 +484,9 @@ second_stage_paths(PlannerInfo *root, List *firstStagePaths, RelOptInfo *joinrel
 		if (!IsA(((Path *) linitial(path->custom_paths)), CustomScan))
 		{
 			jp = (JoinPath *) linitial(path->custom_paths);
-			Assert(jp->path.pathtype == T_HashJoin);
+
+			if (jp->path.pathtype != T_HashJoin)
+				continue;
 			expath = create_exchange_path(root, joinrel, (Path *) jp, GATHER_MODE);
 			path->custom_paths = list_delete(path->custom_paths, jp);
 			path->custom_paths = lappend(path->custom_paths, expath);
@@ -433,7 +499,6 @@ second_stage_paths(PlannerInfo *root, List *firstStagePaths, RelOptInfo *joinrel
 				&innerex->altrel, extra->restrictlist, jointype))
 		{
 			/* Simple case like foreign-push-join case. */
-//			elog(INFO, "--- MAKE SIMPLE PATH ---");
 			innerex->mode = STEALTH_MODE;
 			outerex->mode = STEALTH_MODE;
 		}
@@ -442,7 +507,6 @@ second_stage_paths(PlannerInfo *root, List *firstStagePaths, RelOptInfo *joinrel
 			CustomPath *newpath;
 			bool res;
 
-//			elog(INFO, "--- MAKE SHUFFLE PATH ---");
 			/* Get a copy of the simple path */
 			newpath = duplicate_join_path(path);
 			set_path_pointers(newpath, &jp, &expath, &outerex, &innerex);
diff --git a/contrib/pg_exchange/nodeDistPlanExec.c b/contrib/pg_exchange/nodeDistPlanExec.c
@@ -238,7 +238,7 @@ EstablishDMQConnections(const lcontext *context, const char *serverName, PlanSta
 			sprintf(connstr, "host=%s port=%d "
 							 "fallback_application_name=%s",
 							 host, port, senderName);
-			elog(LOG, "Add destination: senderName=%s, receiverName=%s, connstr=%s", senderName, receiverName, connstr);
+//			elog(LOG, "Add destination: senderName=%s, receiverName=%s, connstr=%s", senderName, receiverName, connstr);
 			sub->dest_id = dmq_destination_add(connstr, senderName, receiverName, 10);
 			memcpy(sub->node, receiverName, strlen(receiverName) + 1);
 		}
@@ -337,7 +337,7 @@ ExecEndDistPlanExec(CustomScanState *node)
 		PGresult	*result;
 
 		while ((result = PQgetResult(dpe->conn[i])) != NULL);
-		elog(LOG, "ExecEndDistPlanExec: %d", PQresultStatus(result));
+//		elog(LOG, "ExecEndDistPlanExec: %d", PQresultStatus(result));
 	}
 	if (dpe->conn)
 		pfree(dpe->conn);
diff --git a/contrib/pg_exchange/nodeDistPlanExec.h b/contrib/pg_exchange/nodeDistPlanExec.h
@@ -28,6 +28,9 @@ typedef struct
 extern char destsName[10];
 #define DISTEXECPATHNAME	"DistExecPath"
 
+#define IsDistExecNode(pathnode) ((((Path *) pathnode)->pathtype == T_CustomScan) && \
+	(strcmp(((CustomPath *)pathnode)->methods->CustomName, DISTEXECPATHNAME) == 0))
+
 extern void DistExec_Init_methods(void);
 extern CustomScan *make_distplanexec(List *custom_plans, List *tlist, List *private_data);
 extern Path *create_distexec_path(PlannerInfo *root, RelOptInfo *rel,
diff --git a/contrib/pg_execplan/tests/init_node0.sql b/contrib/pg_execplan/tests/init_node0.sql
@@ -25,6 +25,17 @@ CREATE TABLE rt_0 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 0);
 CREATE FOREIGN TABLE rt_1 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 1) SERVER remote1;
 CREATE FOREIGN TABLE rt_2 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 2) SERVER remote2;
 
+DROP TABLE IF EXISTS st cascade;
+CREATE TABLE st (
+    id integer not null,
+    payload integer,
+    test integer
+) PARTITION BY hash (id);
+
+CREATE TABLE st_0 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 0);
+CREATE FOREIGN TABLE st_1 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 1) SERVER remote1;
+CREATE FOREIGN TABLE st_2 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 2) SERVER remote2;
+
 
 -- For local tests
 CREATE TABLE a (
diff --git a/contrib/pg_execplan/tests/init_node1.sql b/contrib/pg_execplan/tests/init_node1.sql
@@ -24,3 +24,14 @@ CREATE TABLE rt (
 CREATE FOREIGN TABLE rt_0 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 0) SERVER remote1;
 CREATE TABLE rt_1 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 1);
 CREATE FOREIGN TABLE rt_2 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 2) SERVER remote2;
+
+DROP TABLE IF EXISTS st cascade;
+CREATE TABLE st (
+    id integer not null,
+    payload integer,
+    test integer
+) PARTITION BY hash (id);
+
+CREATE FOREIGN TABLE st_0 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 0) SERVER remote1;
+CREATE TABLE st_1 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 1);
+CREATE FOREIGN TABLE st_2 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 2) SERVER remote2;
diff --git a/contrib/pg_execplan/tests/init_node2.sql b/contrib/pg_execplan/tests/init_node2.sql
@@ -24,3 +24,14 @@ CREATE TABLE rt (
 CREATE FOREIGN TABLE rt_0 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 0) SERVER remote1;
 CREATE FOREIGN TABLE rt_1 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 1) SERVER remote2;
 CREATE TABLE rt_2 PARTITION OF rt FOR VALUES WITH (modulus 3, remainder 2);
+
+DROP TABLE IF EXISTS st cascade;
+CREATE TABLE st (
+    id integer not null,
+    payload integer,
+    test integer
+) PARTITION BY hash (id);
+
+CREATE FOREIGN TABLE st_0 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 0) SERVER remote1;
+CREATE FOREIGN TABLE st_1 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 1) SERVER remote2;
+CREATE TABLE st_2 PARTITION OF st FOR VALUES WITH (modulus 3, remainder 2);
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
diff --git a/src/include/optimizer/paths.h b/src/include/optimizer/paths.h
