Make equi join clauses normalized according to join sides

Author: sopel39

presto-main/src/main/java/io/prestosql/sql/planner/plan/JoinNode.java

@@ -101,16 +101,24 @@ public JoinNode(
         this.spillable = spillable;
         this.dynamicFilters = ImmutableMap.copyOf(requireNonNull(dynamicFilters, "dynamicFilters is null"));
 
+        Set<Symbol> leftSymbols = ImmutableSet.copyOf(left.getOutputSymbols());
+        Set<Symbol> rightSymbols = ImmutableSet.copyOf(right.getOutputSymbols());
         Set<Symbol> inputSymbols = ImmutableSet.<Symbol>builder()
-                .addAll(left.getOutputSymbols())
-                .addAll(right.getOutputSymbols())
+                .addAll(leftSymbols)
+                .addAll(rightSymbols)
                 .build();
         checkArgument(new HashSet<>(inputSymbols).containsAll(outputSymbols), "Left and right join inputs do not contain all output symbols");
         checkArgument(!isCrossJoin() || inputSymbols.size() == outputSymbols.size(), "Cross join does not support output symbols pruning or reordering");
 
         checkArgument(!(criteria.isEmpty() && leftHashSymbol.isPresent()), "Left hash symbol is only valid in an equijoin");
         checkArgument(!(criteria.isEmpty() && rightHashSymbol.isPresent()), "Right hash symbol is only valid in an equijoin");
 
+        criteria.forEach(equiJoinClause ->
+                checkArgument(
+                        leftSymbols.contains(equiJoinClause.getLeft()) &&
+                                rightSymbols.contains(equiJoinClause.getRight()),
+                        "Equality join criteria should be normalized according to join sides: %s", equiJoinClause));
+
         if (distributionType.isPresent()) {
             // The implementation of full outer join only works if the data is hash partitioned.
             checkArgument(
@@ -127,7 +135,7 @@ public JoinNode(
         }
 
         for (Symbol symbol : dynamicFilters.values()) {
-            checkArgument(right.getOutputSymbols().contains(symbol), "Right join input doesn't contain symbol for dynamic filter: %s", symbol);
+            checkArgument(rightSymbols.contains(symbol), "Right join input doesn't contain symbol for dynamic filter: %s", symbol);
         }
     }
 

presto-main/src/test/java/io/prestosql/sql/planner/iterative/rule/TestEliminateCrossJoins.java

@@ -118,7 +118,7 @@ public void testJoinOrder()
                                 values("b")),
                         values("c"),
                         "a", "c",
-                        "c", "b");
+                        "b", "c");
 
         JoinGraph joinGraph = JoinGraph.buildFrom(plan, noLookup(), new PlanNodeIdAllocator());
 
@@ -137,7 +137,7 @@ public void testJoinOrderWithRealCrossJoin()
                                 values("b")),
                         values("c"),
                         "a", "c",
-                        "c", "b");
+                        "b", "c");
 
         PlanNode rightPlan =
                 joinNode(
@@ -146,7 +146,7 @@ public void testJoinOrderWithRealCrossJoin()
                                 values("y")),
                         values("z"),
                         "x", "z",
-                        "z", "y");
+                        "y", "z");
 
         PlanNode plan = joinNode(leftPlan, rightPlan);
 
@@ -167,8 +167,8 @@ public void testJoinOrderWithMultipleEdgesBetweenNodes()
                                 values("b1", "b2")),
                         values("c1", "c2"),
                         "a", "c1",
-                        "c1", "b1",
-                        "c2", "b2");
+                        "b1", "c1",
+                        "b2", "c2");
 
         JoinGraph joinGraph = JoinGraph.buildFrom(plan, noLookup(), new PlanNodeIdAllocator());
 
@@ -187,7 +187,7 @@ public void testDoesNotChangeOrderWithoutCrossJoin()
                                 values("b"),
                                 "a", "b"),
                         values("c"),
-                        "c", "b");
+                        "b", "c");
 
         JoinGraph joinGraph = JoinGraph.buildFrom(plan, noLookup(), new PlanNodeIdAllocator());
 
@@ -205,7 +205,7 @@ public void testDoNotReorderCrossJoins()
                                 values("a"),
                                 values("b")),
                         values("c"),
-                        "c", "b");
+                        "b", "c");
 
         JoinGraph joinGraph = JoinGraph.buildFrom(plan, noLookup(), new PlanNodeIdAllocator());
 
@@ -290,7 +290,7 @@ public void testGiveUpOnComplexProjections()
                                 new SymbolReference("b")),
                         values("c"),
                         "a2", "c",
-                        "c", "b");
+                        "b", "c");
 
         assertEquals(JoinGraph.buildFrom(plan, noLookup(), new PlanNodeIdAllocator()).size(), 2);
     }
@@ -309,8 +309,8 @@ private Function<PlanBuilder, PlanNode> crossJoinAndJoin(JoinNode.Type secondJoi
                             p.values(axSymbol),
                             p.values(bySymbol)),
                     p.values(cxSymbol, cySymbol),
-                    new EquiJoinClause(cxSymbol, axSymbol),
-                    new EquiJoinClause(cySymbol, bySymbol));
+                    new EquiJoinClause(axSymbol, cxSymbol),
+                    new EquiJoinClause(bySymbol, cySymbol));
         };
     }
 

presto-main/src/test/java/io/prestosql/sql/planner/iterative/rule/TestReorderJoins.java

@@ -284,7 +284,7 @@ public void testReplicatedScalarJoinEvenWhereSessionRequiresRepartitioned()
                                 INNER,
                                 p.values(new PlanNodeId("valuesB"), ImmutableList.of(p.symbol("B1")), TWO_ROWS),
                                 p.values(new PlanNodeId("valuesA"), p.symbol("A1")), // matches isAtMostScalar
-                                ImmutableList.of(new EquiJoinClause(p.symbol("A1"), p.symbol("B1"))),
+                                ImmutableList.of(new EquiJoinClause(p.symbol("B1"), p.symbol("A1"))),
                                 ImmutableList.of(p.symbol("A1"), p.symbol("B1")),
                                 Optional.empty()))
                 .overrideStats("valuesA", valuesA)