Basic implementation of recursive planning of non-colocated relation joins

src/backend/distributed/planner/query_colocation_checker.c

@@ -31,10 +31,10 @@
 #include "parser/parse_relation.h"
 #include "optimizer/planner.h"
 #include "optimizer/prep.h"
+#include "utils/rel.h"
 
 
 static RangeTblEntry * AnchorRte(Query *subquery);
-static Query * WrapRteRelationIntoSubquery(RangeTblEntry *rteRelation);
 static List * UnionRelationRestrictionLists(List *firstRelationList,
 											List *secondRelationList);
 
@@ -238,14 +238,16 @@ SubqueryColocated(Query *subquery, ColocatedJoinChecker *checker)
  * projections. The returned query should be used cautiosly and it is mostly
  * designed for generating a stub query.
  */
-static Query *
+Query *
 WrapRteRelationIntoSubquery(RangeTblEntry *rteRelation)
 {
 	Query *subquery = makeNode(Query);
 	RangeTblRef *newRangeTableRef = makeNode(RangeTblRef);
 	RangeTblEntry *newRangeTableEntry = NULL;
-	Var *targetColumn = NULL;
+
-	TargetEntry *targetEntry = NULL;
+	Relation relation = NULL;
+	int attributeNumber = 0;
+	int numberOfAttributes = 0;
 
 	subquery->commandType = CMD_SELECT;
 
@@ -258,14 +260,25 @@ WrapRteRelationIntoSubquery(RangeTblEntry *rteRelation)
 	newRangeTableRef->rtindex = 1;
 	subquery->jointree = makeFromExpr(list_make1(newRangeTableRef), NULL);
 
-	/* Need the whole row as a junk var */
+	relation = relation_open(rteRelation->relid, AccessShareLock);
-	targetColumn = makeWholeRowVar(newRangeTableEntry, newRangeTableRef->rtindex, 0,
-								   false);
 
-	/* create a dummy target entry */
+	numberOfAttributes = RelationGetNumberOfAttributes(relation);
-	targetEntry = makeTargetEntry((Expr *) targetColumn, 1, "wholerow", true);
 
-	subquery->targetList = lappend(subquery->targetList, targetEntry);
+	for (attributeNumber = 1; attributeNumber <= numberOfAttributes; attributeNumber++)
+	{
+		Form_pg_attribute attributeTuple = TupleDescAttr(relation->rd_att,
+														 attributeNumber - 1);
+		Var *targetColumn =
+			makeVar(1, attributeNumber, attributeTuple->atttypid,
+					attributeTuple->atttypmod, attributeTuple->attcollation, 0);
+		TargetEntry *targetEntry =
+			makeTargetEntry((Expr *) targetColumn, attributeNumber,
+							strdup(attributeTuple->attname.data), false);
+
+		subquery->targetList = lappend(subquery->targetList, targetEntry);
+	}
+
+	relation_close(relation, NoLock);
 
 	return subquery;
 }

src/backend/distributed/planner/recursive_planning.c

@@ -78,6 +78,7 @@
 #include "nodes/relation.h"
 #include "utils/builtins.h"
 #include "utils/guc.h"
+#include "utils/lsyscache.h"
 #include "../../../include/distributed/query_pushdown_planning.h"
 
 
@@ -154,6 +155,8 @@ static bool ShouldRecursivelyPlanSetOperation(Query *query,
 static void RecursivelyPlanSetOperations(Query *query, Node *node,
 										 RecursivePlanningContext *context);
 static bool IsLocalTableRTE(Node *node);
+static void RecursivelyPlanRTERelation(RangeTblEntry *relationRte,
+									   RecursivePlanningContext *planningContext);
 static void RecursivelyPlanSubquery(Query *subquery,
 									RecursivePlanningContext *planningContext);
 static DistributedSubPlan * CreateDistributedSubPlan(uint32 subPlanId,
@@ -474,20 +477,25 @@ RecursivelyPlanNonColocatedJoinWalker(Node *joinNode,
 		RangeTblEntry *rte = rt_fetch(rangeTableIndex, rangeTableList);
 		Query *subquery = NULL;
 
-		/* we're only interested in subqueries for now */
+		if (rte->rtekind == RTE_RELATION)
-		if (rte->rtekind != RTE_SUBQUERY)
 		{
-			return;
+			subquery = WrapRteRelationIntoSubquery(rte);
+			if (!SubqueryColocated(subquery, colocatedJoinChecker))
+			{
+				RecursivelyPlanRTERelation(rte, recursivePlanningContext);
+			}
 		}
-
+		else if (rte->rtekind == RTE_SUBQUERY)
-		/*
-		 * If the subquery is not colocated with the anchor subquery,
-		 * recursively plan it.
-		 */
-		subquery = rte->subquery;
-		if (!SubqueryColocated(subquery, colocatedJoinChecker))
 		{
-			RecursivelyPlanSubquery(subquery, recursivePlanningContext);
+			/*
+			 * If the subquery is not colocated with the anchor subquery,
+			 * recursively plan it.
+			 */
+			subquery = rte->subquery;
+			if (!SubqueryColocated(subquery, colocatedJoinChecker))
+			{
+				RecursivelyPlanSubquery(subquery, recursivePlanningContext);
+			}
 		}
 	}
 	else
@@ -1063,6 +1071,44 @@ IsLocalTableRTE(Node *node)
 }
 
 
+/*
+ * RecursivelyPlanRTERelation wraps the relation with a subquery
+ * (e.g., users_table becomes (SELECT * FROM users_table) as users_table).
+ *
+ * Later, the subquery is recursively planned via RecursivelyPlanSubquery().
+ */
+static void
+RecursivelyPlanRTERelation(RangeTblEntry *relationRte,
+						   RecursivePlanningContext *planningContext)
+{
+	Query *subquery = WrapRteRelationIntoSubquery(relationRte);
+	RangeTblEntry *wrappedSubqueryEntry = makeNode(RangeTblEntry);
+	ListCell *columnListCell = NULL;
+	List *colNames = NIL;
+	char *relationName = get_rel_name(relationRte->relid);
+
+	wrappedSubqueryEntry->rtekind = RTE_SUBQUERY;
+	wrappedSubqueryEntry->subquery = copyObject(subquery);
+
+	foreach(columnListCell, subquery->targetList)
+	{
+		TargetEntry *tle = (TargetEntry *) lfirst(columnListCell);
+
+		colNames =
+			lappend(colNames, makeString(pstrdup(tle->resname)));
+	}
+
+	wrappedSubqueryEntry->eref = makeAlias(pstrdup(relationName), copyObject(colNames));
+	wrappedSubqueryEntry->alias = makeAlias(pstrdup(relationName), copyObject(colNames));
+
+	wrappedSubqueryEntry->inFromCl = true;
+
+	memcpy(relationRte, wrappedSubqueryEntry, sizeof(RangeTblEntry));
+
+	RecursivelyPlanSubquery(wrappedSubqueryEntry->subquery, planningContext);
+}
+
+
 /*
  * RecursivelyPlanQuery recursively plans a query, replaces it with a
  * result query and returns the subplan.

src/include/distributed/query_colocation_checker.h

@@ -34,6 +34,7 @@ extern ColocatedJoinChecker CreateColocatedJoinChecker(Query *subquery,
 													   PlannerRestrictionContext *
 													   restrictionContext);
 extern bool SubqueryColocated(Query *subquery, ColocatedJoinChecker *context);
+extern Query * WrapRteRelationIntoSubquery(RangeTblEntry *rteRelation);
 
 
 #endif /* QUERY_COLOCATION_CHECKER_H */

src/test/regress/expected/multi_subquery_complex_reference_clause.out

@@ -1252,7 +1252,15 @@ WHERE
 GROUP BY 1
 ORDER BY 2 DESC, 1 DESC
 LIMIT 5;
-ERROR:  complex joins are only supported when all distributed tables are joined on their distribution columns with equal operator
+ user_id | count 
+---------+-------
+       5 |   936
+       4 |   644
+       6 |   390
+       3 |   289
+       2 |   144
+(5 rows)
+
 SELECT foo.user_id FROM
 (
   SELECT m.user_id, random() FROM users_table m JOIN events_reference_table r ON int4eq(m.user_id, r.user_id)

src/test/regress/expected/non_colocated_subquery_joins.out

@@ -638,9 +638,14 @@ SELECT true AS valid FROM explain_json_2($$
     FROM
         (users_table u1 JOIN users_table u2 using(value_1)) a JOIN (SELECT value_1, random() FROM users_table) as u3 USING (value_1); 
 $$);
-DEBUG:  generating subplan 66_1 for subquery SELECT value_1, random() AS random FROM public.users_table
+DEBUG:  generating subplan 66_1 for subquery SELECT user_id, "time", value_1, value_2, value_3, value_4 FROM public.users_table u2
-DEBUG:  Plan 66 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((public.users_table u1 JOIN public.users_table u2 USING (value_1)) a(value_1, user_id, "time", value_2, value_3, value_4, user_id_1, time_1, value_2_1, value_3_1, value_4_1) JOIN (SELECT intermediate_result.value_1, intermediate_result.random FROM read_intermediate_result('66_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, random double precision)) u3 USING (value_1))
+DEBUG:  generating subplan 66_2 for subquery SELECT value_1, random() AS random FROM public.users_table
-ERROR:  complex joins are only supported when all distributed tables are joined on their distribution columns with equal operator
+DEBUG:  Plan 66 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((public.users_table u1 JOIN (SELECT intermediate_result.user_id, intermediate_result."time", intermediate_result.value_1, intermediate_result.value_2, intermediate_result.value_3, intermediate_result.value_4 FROM read_intermediate_result('66_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer, "time" timestamp without time zone, value_1 integer, value_2 integer, value_3 double precision, value_4 bigint)) users_table(user_id, "time", value_1, value_2, value_3, value_4) USING (value_1)) a(value_1, user_id, "time", value_2, value_3, value_4, user_id_1, time_1, value_2_1, value_3_1, value_4_1) JOIN (SELECT intermediate_result.value_1, intermediate_result.random FROM read_intermediate_result('66_2'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, random double precision)) u3 USING (value_1))
+ valid 
+-------
+ t
+(1 row)
+
 -- a very similar query to the above
 -- however, this time we users a subquery instead of join alias, and it works
 SELECT true AS valid FROM explain_json_2($$
@@ -651,8 +656,8 @@ SELECT true AS valid FROM explain_json_2($$
         (SELECT * FROM users_table u1 JOIN users_table u2 using(value_1)) a JOIN (SELECT value_1, random() FROM users_table) as u3 USING (value_1); 
 $$);
 DEBUG:  cannot use real time executor with repartition jobs
-DEBUG:  generating subplan 68_1 for subquery SELECT u1.value_1, u1.user_id, u1."time", u1.value_2, u1.value_3, u1.value_4, u2.user_id, u2."time", u2.value_2, u2.value_3, u2.value_4 FROM (public.users_table u1 JOIN public.users_table u2 USING (value_1))
+DEBUG:  generating subplan 69_1 for subquery SELECT u1.value_1, u1.user_id, u1."time", u1.value_2, u1.value_3, u1.value_4, u2.user_id, u2."time", u2.value_2, u2.value_3, u2.value_4 FROM (public.users_table u1 JOIN public.users_table u2 USING (value_1))
-DEBUG:  Plan 68 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT intermediate_result.value_1, intermediate_result.user_id, intermediate_result."time", intermediate_result.value_2, intermediate_result.value_3, intermediate_result.value_4, intermediate_result.user_id_1 AS user_id, intermediate_result.time_1 AS "time", intermediate_result.value_2_1 AS value_2, intermediate_result.value_3_1 AS value_3, intermediate_result.value_4_1 AS value_4 FROM read_intermediate_result('68_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, user_id integer, "time" timestamp without time zone, value_2 integer, value_3 double precision, value_4 bigint, user_id_1 integer, time_1 timestamp without time zone, value_2_1 integer, value_3_1 double precision, value_4_1 bigint)) a(value_1, user_id, "time", value_2, value_3, value_4, user_id_1, time_1, value_2_1, value_3_1, value_4_1) JOIN (SELECT users_table.value_1, random() AS random FROM public.users_table) u3 USING (value_1))
+DEBUG:  Plan 69 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT intermediate_result.value_1, intermediate_result.user_id, intermediate_result."time", intermediate_result.value_2, intermediate_result.value_3, intermediate_result.value_4, intermediate_result.user_id_1 AS user_id, intermediate_result.time_1 AS "time", intermediate_result.value_2_1 AS value_2, intermediate_result.value_3_1 AS value_3, intermediate_result.value_4_1 AS value_4 FROM read_intermediate_result('69_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, user_id integer, "time" timestamp without time zone, value_2 integer, value_3 double precision, value_4 bigint, user_id_1 integer, time_1 timestamp without time zone, value_2_1 integer, value_3_1 double precision, value_4_1 bigint)) a(value_1, user_id, "time", value_2, value_3, value_4, user_id_1, time_1, value_2_1, value_3_1, value_4_1) JOIN (SELECT users_table.value_1, random() AS random FROM public.users_table) u3 USING (value_1))
  valid 
 -------
  t
@@ -670,8 +675,8 @@ SELECT true AS valid FROM explain_json_2($$
         events_table
             using (value_2);
 $$);
-DEBUG:  generating subplan 70_1 for subquery SELECT value_2, random() AS random FROM public.users_table
+DEBUG:  generating subplan 71_1 for subquery SELECT value_2, random() AS random FROM public.users_table
-DEBUG:  Plan 70 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('70_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u1 JOIN public.events_table USING (value_2))
+DEBUG:  Plan 71 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('71_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u1 JOIN public.events_table USING (value_2))
  valid 
 -------
  t
@@ -688,8 +693,8 @@ SELECT true AS valid FROM explain_json_2($$
         (SELECT value_2, random() FROM users_table) as u2
             USING(value_2);
 $$);
-DEBUG:  generating subplan 72_1 for subquery SELECT value_2, random() AS random FROM public.users_table
+DEBUG:  generating subplan 73_1 for subquery SELECT value_2, random() AS random FROM public.users_table
-DEBUG:  Plan 72 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT users_table.value_2, random() AS random FROM public.users_table) u1 LEFT JOIN (SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('72_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u2 USING (value_2))
+DEBUG:  Plan 73 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT users_table.value_2, random() AS random FROM public.users_table) u1 LEFT JOIN (SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('73_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u2 USING (value_2))
  valid 
 -------
  t
@@ -708,8 +713,8 @@ SELECT true AS valid FROM explain_json_2($$
         (SELECT value_2, random() FROM users_table) as u2
             USING(value_2);
 $$);
-DEBUG:  generating subplan 74_1 for subquery SELECT value_2, random() AS random FROM public.users_table
+DEBUG:  generating subplan 75_1 for subquery SELECT value_2, random() AS random FROM public.users_table
-DEBUG:  Plan 74 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT users_table.value_2, random() AS random FROM public.users_table) u1 RIGHT JOIN (SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('74_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u2 USING (value_2))
+DEBUG:  Plan 75 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((SELECT users_table.value_2, random() AS random FROM public.users_table) u1 RIGHT JOIN (SELECT intermediate_result.value_2, intermediate_result.random FROM read_intermediate_result('75_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer, random double precision)) u2 USING (value_2))
 ERROR:  cannot pushdown the subquery
 -- set operations may produce not very efficient plans
 -- although we could have picked a as our anchor subquery,
@@ -727,11 +732,11 @@ SELECT true AS valid FROM explain_json_2($$
         (SELECT value_1 FROM users_table) as foo ON (a.user_id = foo.value_1) 
     );
 $$);
-DEBUG:  generating subplan 77_1 for subquery SELECT user_id FROM public.users_table
+DEBUG:  generating subplan 78_1 for subquery SELECT user_id FROM public.users_table
-DEBUG:  generating subplan 77_2 for subquery SELECT user_id FROM public.users_table
+DEBUG:  generating subplan 78_2 for subquery SELECT user_id FROM public.users_table
-DEBUG:  Plan 77 query after replacing subqueries and CTEs: SELECT intermediate_result.user_id FROM read_intermediate_result('77_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer) UNION SELECT intermediate_result.user_id FROM read_intermediate_result('77_2'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)
+DEBUG:  Plan 78 query after replacing subqueries and CTEs: SELECT intermediate_result.user_id FROM read_intermediate_result('78_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer) UNION SELECT intermediate_result.user_id FROM read_intermediate_result('78_2'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)
-DEBUG:  generating subplan 76_1 for subquery SELECT users_table.user_id FROM public.users_table UNION SELECT users_table.user_id FROM public.users_table
+DEBUG:  generating subplan 77_1 for subquery SELECT users_table.user_id FROM public.users_table UNION SELECT users_table.user_id FROM public.users_table
-DEBUG:  Plan 76 query after replacing subqueries and CTEs: SELECT a.user_id, foo.value_1 FROM ((SELECT intermediate_result.user_id FROM read_intermediate_result('76_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)) a JOIN (SELECT users_table.value_1 FROM public.users_table) foo ON ((a.user_id OPERATOR(pg_catalog.=) foo.value_1)))
+DEBUG:  Plan 77 query after replacing subqueries and CTEs: SELECT a.user_id, foo.value_1 FROM ((SELECT intermediate_result.user_id FROM read_intermediate_result('77_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)) a JOIN (SELECT users_table.value_1 FROM public.users_table) foo ON ((a.user_id OPERATOR(pg_catalog.=) foo.value_1)))
  valid 
 -------
  t
@@ -751,11 +756,11 @@ SELECT true AS valid FROM explain_json_2($$
         users_table as foo ON (a.user_id = foo.value_1) 
     );
 $$);
-DEBUG:  generating subplan 81_1 for subquery SELECT user_id FROM public.users_table
+DEBUG:  generating subplan 82_1 for subquery SELECT user_id FROM public.users_table
-DEBUG:  generating subplan 81_2 for subquery SELECT user_id FROM public.users_table
+DEBUG:  generating subplan 82_2 for subquery SELECT user_id FROM public.users_table
-DEBUG:  Plan 81 query after replacing subqueries and CTEs: SELECT intermediate_result.user_id FROM read_intermediate_result('81_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer) UNION SELECT intermediate_result.user_id FROM read_intermediate_result('81_2'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)
+DEBUG:  Plan 82 query after replacing subqueries and CTEs: SELECT intermediate_result.user_id FROM read_intermediate_result('82_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer) UNION SELECT intermediate_result.user_id FROM read_intermediate_result('82_2'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)
-DEBUG:  generating subplan 80_1 for subquery SELECT users_table.user_id FROM public.users_table UNION SELECT users_table.user_id FROM public.users_table
+DEBUG:  generating subplan 81_1 for subquery SELECT users_table.user_id FROM public.users_table UNION SELECT users_table.user_id FROM public.users_table
-DEBUG:  Plan 80 query after replacing subqueries and CTEs: SELECT a.user_id, foo.user_id, foo."time", foo.value_1, foo.value_2, foo.value_3, foo.value_4 FROM ((SELECT intermediate_result.user_id FROM read_intermediate_result('80_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)) a JOIN public.users_table foo ON ((a.user_id OPERATOR(pg_catalog.=) foo.value_1)))
+DEBUG:  Plan 81 query after replacing subqueries and CTEs: SELECT a.user_id, foo.user_id, foo."time", foo.value_1, foo.value_2, foo.value_3, foo.value_4 FROM ((SELECT intermediate_result.user_id FROM read_intermediate_result('81_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer)) a JOIN public.users_table foo ON ((a.user_id OPERATOR(pg_catalog.=) foo.value_1)))
  valid 
 -------
  t
@@ -784,8 +789,8 @@ SELECT true AS valid FROM explain_json_2($$
          ON(foo.user_id = bar.value_1) 
     );
 $$);
-DEBUG:  generating subplan 84_1 for subquery SELECT value_1 FROM public.users_table
+DEBUG:  generating subplan 85_1 for subquery SELECT value_1 FROM public.users_table
-DEBUG:  Plan 84 query after replacing subqueries and CTEs: SELECT foo.user_id, a.user_id, bar.value_1 FROM (((SELECT users_table.user_id FROM public.users_table) foo JOIN (SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])) UNION SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8]))) a ON ((a.user_id OPERATOR(pg_catalog.=) foo.user_id))) JOIN (SELECT intermediate_result.value_1 FROM read_intermediate_result('84_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer)) bar ON ((foo.user_id OPERATOR(pg_catalog.=) bar.value_1)))
+DEBUG:  Plan 85 query after replacing subqueries and CTEs: SELECT foo.user_id, a.user_id, bar.value_1 FROM (((SELECT users_table.user_id FROM public.users_table) foo JOIN (SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])) UNION SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8]))) a ON ((a.user_id OPERATOR(pg_catalog.=) foo.user_id))) JOIN (SELECT intermediate_result.value_1 FROM read_intermediate_result('85_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer)) bar ON ((foo.user_id OPERATOR(pg_catalog.=) bar.value_1)))
  valid 
 -------
  t
@@ -823,13 +828,13 @@ SELECT true AS valid FROM explain_json_2($$
     WHERE 
         non_colocated_subquery.value_2 != non_colocated_subquery_2.cnt
 $$);
-DEBUG:  generating subplan 86_1 for CTE non_colocated_subquery: SELECT foo.value_2 FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))) bar WHERE (foo.value_2 OPERATOR(pg_catalog.=) bar.value_2)
+DEBUG:  generating subplan 87_1 for CTE non_colocated_subquery: SELECT foo.value_2 FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))) bar WHERE (foo.value_2 OPERATOR(pg_catalog.=) bar.value_2)
-DEBUG:  generating subplan 87_1 for subquery SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))
+DEBUG:  generating subplan 88_1 for subquery SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))
-DEBUG:  Plan 87 query after replacing subqueries and CTEs: SELECT foo.value_2 FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT intermediate_result.value_2 FROM read_intermediate_result('87_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) bar WHERE (foo.value_2 OPERATOR(pg_catalog.=) bar.value_2)
+DEBUG:  Plan 88 query after replacing subqueries and CTEs: SELECT foo.value_2 FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT intermediate_result.value_2 FROM read_intermediate_result('88_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) bar WHERE (foo.value_2 OPERATOR(pg_catalog.=) bar.value_2)
-DEBUG:  generating subplan 86_2 for CTE non_colocated_subquery_2: SELECT count(*) AS cnt FROM public.events_table WHERE (event_type OPERATOR(pg_catalog.=) ANY (SELECT events_table_1.event_type FROM public.events_table events_table_1 WHERE (events_table_1.user_id OPERATOR(pg_catalog.<) 4)))
+DEBUG:  generating subplan 87_2 for CTE non_colocated_subquery_2: SELECT count(*) AS cnt FROM public.events_table WHERE (event_type OPERATOR(pg_catalog.=) ANY (SELECT events_table_1.event_type FROM public.events_table events_table_1 WHERE (events_table_1.user_id OPERATOR(pg_catalog.<) 4)))
-DEBUG:  generating subplan 89_1 for subquery SELECT event_type FROM public.events_table WHERE (user_id OPERATOR(pg_catalog.<) 4)
+DEBUG:  generating subplan 90_1 for subquery SELECT event_type FROM public.events_table WHERE (user_id OPERATOR(pg_catalog.<) 4)
-DEBUG:  Plan 89 query after replacing subqueries and CTEs: SELECT count(*) AS cnt FROM public.events_table WHERE (event_type OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.event_type FROM read_intermediate_result('89_1'::text, 'binary'::citus_copy_format) intermediate_result(event_type integer)))
+DEBUG:  Plan 90 query after replacing subqueries and CTEs: SELECT count(*) AS cnt FROM public.events_table WHERE (event_type OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.event_type FROM read_intermediate_result('90_1'::text, 'binary'::citus_copy_format) intermediate_result(event_type integer)))
-DEBUG:  Plan 86 query after replacing subqueries and CTEs: SELECT non_colocated_subquery.value_2, non_colocated_subquery_2.cnt FROM (SELECT intermediate_result.value_2 FROM read_intermediate_result('86_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) non_colocated_subquery, (SELECT intermediate_result.cnt FROM read_intermediate_result('86_2'::text, 'binary'::citus_copy_format) intermediate_result(cnt bigint)) non_colocated_subquery_2 WHERE (non_colocated_subquery.value_2 OPERATOR(pg_catalog.<>) non_colocated_subquery_2.cnt)
+DEBUG:  Plan 87 query after replacing subqueries and CTEs: SELECT non_colocated_subquery.value_2, non_colocated_subquery_2.cnt FROM (SELECT intermediate_result.value_2 FROM read_intermediate_result('87_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) non_colocated_subquery, (SELECT intermediate_result.cnt FROM read_intermediate_result('87_2'::text, 'binary'::citus_copy_format) intermediate_result(cnt bigint)) non_colocated_subquery_2 WHERE (non_colocated_subquery.value_2 OPERATOR(pg_catalog.<>) non_colocated_subquery_2.cnt)
  valid 
 -------
  t
@@ -848,9 +853,9 @@ SELECT true AS valid FROM explain_json_2($$
         AND 
         foo.value_2 = baz.value_2
 $$);
-DEBUG:  generating subplan 91_1 for subquery SELECT users_table_local.value_2 FROM non_colocated_subquery.users_table_local, non_colocated_subquery.events_table_local WHERE ((users_table_local.user_id OPERATOR(pg_catalog.=) events_table_local.user_id) AND (events_table_local.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))
+DEBUG:  generating subplan 92_1 for subquery SELECT users_table_local.value_2 FROM non_colocated_subquery.users_table_local, non_colocated_subquery.events_table_local WHERE ((users_table_local.user_id OPERATOR(pg_catalog.=) events_table_local.user_id) AND (events_table_local.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8])))
-DEBUG:  generating subplan 91_2 for subquery SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[9, 10, 11, 12])))
+DEBUG:  generating subplan 92_2 for subquery SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[9, 10, 11, 12])))
-DEBUG:  Plan 91 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT intermediate_result.value_2 FROM read_intermediate_result('91_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) bar, (SELECT intermediate_result.value_2 FROM read_intermediate_result('91_2'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) baz WHERE ((foo.value_2 OPERATOR(pg_catalog.=) bar.value_2) AND (foo.value_2 OPERATOR(pg_catalog.=) baz.value_2))
+DEBUG:  Plan 92 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (SELECT users_table.value_2 FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])))) foo, (SELECT intermediate_result.value_2 FROM read_intermediate_result('92_1'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) bar, (SELECT intermediate_result.value_2 FROM read_intermediate_result('92_2'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer)) baz WHERE ((foo.value_2 OPERATOR(pg_catalog.=) bar.value_2) AND (foo.value_2 OPERATOR(pg_catalog.=) baz.value_2))
  valid 
 -------
  t
@@ -885,10 +890,10 @@ SELECT true AS valid FROM explain_json_2($$
             AND
         foo.user_id IN (SELECT users_table.user_id FROM users_table, events_table WHERE users_table.user_id = events_table.user_id AND event_type IN (1,2))
 $$);
-DEBUG:  generating subplan 93_1 for subquery SELECT value_1, value_2 FROM public.users_table
+DEBUG:  generating subplan 94_1 for subquery SELECT value_1, value_2 FROM public.users_table
-DEBUG:  generating subplan 93_2 for subquery SELECT value_1 FROM public.users_table WHERE (value_2 OPERATOR(pg_catalog.<) 1)
+DEBUG:  generating subplan 94_2 for subquery SELECT value_1 FROM public.users_table WHERE (value_2 OPERATOR(pg_catalog.<) 1)
-DEBUG:  generating subplan 93_3 for subquery SELECT value_2 FROM public.users_table WHERE (value_1 OPERATOR(pg_catalog.<) 2)
+DEBUG:  generating subplan 94_3 for subquery SELECT value_2 FROM public.users_table WHERE (value_1 OPERATOR(pg_catalog.<) 2)
-DEBUG:  Plan 93 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (((SELECT users_table.user_id FROM public.users_table) foo JOIN (SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])) UNION SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8]))) a ON ((a.user_id OPERATOR(pg_catalog.=) foo.user_id))) JOIN (SELECT intermediate_result.value_1, intermediate_result.value_2 FROM read_intermediate_result('93_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, value_2 integer)) bar ON ((foo.user_id OPERATOR(pg_catalog.=) bar.value_1))) WHERE ((bar.value_2 OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.value_1 FROM read_intermediate_result('93_2'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer))) AND (bar.value_1 OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.value_2 FROM read_intermediate_result('93_3'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer))) AND (foo.user_id OPERATOR(pg_catalog.=) ANY (SELECT users_table.user_id FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2]))))))
+DEBUG:  Plan 94 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (((SELECT users_table.user_id FROM public.users_table) foo JOIN (SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2, 3, 4])) UNION SELECT users_table.user_id FROM public.users_table WHERE (users_table.user_id OPERATOR(pg_catalog.=) ANY (ARRAY[5, 6, 7, 8]))) a ON ((a.user_id OPERATOR(pg_catalog.=) foo.user_id))) JOIN (SELECT intermediate_result.value_1, intermediate_result.value_2 FROM read_intermediate_result('94_1'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer, value_2 integer)) bar ON ((foo.user_id OPERATOR(pg_catalog.=) bar.value_1))) WHERE ((bar.value_2 OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.value_1 FROM read_intermediate_result('94_2'::text, 'binary'::citus_copy_format) intermediate_result(value_1 integer))) AND (bar.value_1 OPERATOR(pg_catalog.=) ANY (SELECT intermediate_result.value_2 FROM read_intermediate_result('94_3'::text, 'binary'::citus_copy_format) intermediate_result(value_2 integer))) AND (foo.user_id OPERATOR(pg_catalog.=) ANY (SELECT users_table.user_id FROM public.users_table, public.events_table WHERE ((users_table.user_id OPERATOR(pg_catalog.=) events_table.user_id) AND (events_table.event_type OPERATOR(pg_catalog.=) ANY (ARRAY[1, 2]))))))
 ERROR:  cannot pushdown the subquery
 -- make sure that we don't pick the refeence table as 
 -- the anchor
@@ -903,8 +908,8 @@ SELECT true AS valid FROM explain_json_2($$
       users_table_ref.user_id = foo.user_id
       AND foo.user_id = bar.value_2;
 $$);
-DEBUG:  generating subplan 97_1 for subquery SELECT user_id, value_2 FROM public.events_table
+DEBUG:  generating subplan 98_1 for subquery SELECT user_id, value_2 FROM public.events_table
-DEBUG:  Plan 97 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM public.users_reference_table users_table_ref, (SELECT users_table.user_id FROM public.users_table) foo, (SELECT intermediate_result.user_id, intermediate_result.value_2 FROM read_intermediate_result('97_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer, value_2 integer)) bar WHERE ((users_table_ref.user_id OPERATOR(pg_catalog.=) foo.user_id) AND (foo.user_id OPERATOR(pg_catalog.=) bar.value_2))
+DEBUG:  Plan 98 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM public.users_reference_table users_table_ref, (SELECT users_table.user_id FROM public.users_table) foo, (SELECT intermediate_result.user_id, intermediate_result.value_2 FROM read_intermediate_result('98_1'::text, 'binary'::citus_copy_format) intermediate_result(user_id integer, value_2 integer)) bar WHERE ((users_table_ref.user_id OPERATOR(pg_catalog.=) foo.user_id) AND (foo.user_id OPERATOR(pg_catalog.=) bar.value_2))
  valid 
 -------
  t