feat(optimizer): Push projections through cross joins

presto-docs/src/main/sphinx/admin/properties-session.rst

@@ -401,6 +401,22 @@ performance by allowing the aggregation to pre-reduce data before the join is pe
 
 The corresponding configuration property is :ref:`admin/properties:\`\`optimizer.push-partial-aggregation-through-join\`\``.
 
+``push_projection_through_cross_join``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+When enabled, pushes projection expressions through cross join nodes so that each
+expression is evaluated only on the side of the cross join that provides its input
+variables. This reduces the number of columns flowing through the cross join and
+avoids recomputing expressions on the multiplied output rows.
+
+Only deterministic expressions are pushed. Expressions that reference variables from
+both sides of the cross join, or constant expressions, remain above the join.
+
+The corresponding configuration property is :ref:`admin/properties:\`\`optimizer.push-projection-through-cross-join\`\``.
+
 ``push_table_write_through_union``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

presto-docs/src/main/sphinx/admin/properties.rst

@@ -969,6 +969,22 @@ performance by allowing the aggregation to pre-reduce data before the join is pe
 
 The corresponding session property is :ref:`admin/properties-session:\`\`push_partial_aggregation_through_join\`\``.
 
+``optimizer.push-projection-through-cross-join``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+When enabled, pushes projection expressions through cross join nodes so that each
+expression is evaluated only on the side of the cross join that provides its input
+variables. This reduces the number of columns flowing through the cross join and
+avoids recomputing expressions on the multiplied output rows.
+
+Only deterministic expressions are pushed. Expressions that reference variables from
+both sides of the cross join, or constant expressions, remain above the join.
+
+The corresponding session property is :ref:`admin/properties-session:\`\`push_projection_through_cross_join\`\``.
+
 ``optimizer.push-table-write-through-union``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

presto-hive/src/test/java/com/facebook/presto/hive/benchmark/BenchmarkPushProjectionThroughCrossJoin.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.hive.benchmark;
+
+import org.testng.annotations.Test;
+
+/**
+ * Benchmarks the PushProjectionThroughCrossJoin optimization.
+ *
+ * <p>Uses a real CROSS JOIN between lineitem and a subquery so that the
+ * plan produces a JoinNode with isCrossJoin(). CROSS JOIN UNNEST produces
+ * an UnnestNode instead, which this rule does not target.
+ *
+ * <p>Run via:
+ * <pre>
+ * mvn test -pl presto-hive \
+ *   -Dtest=BenchmarkPushProjectionThroughCrossJoin \
+ *   -DfailIfNoTests=false
+ * </pre>
+ */
+public final class BenchmarkPushProjectionThroughCrossJoin
+{
+    private static final String QUERY =
+            "SELECT " +
+            "  regexp_replace(l.comment, '[aeiou]', '*') AS l_redacted, " +
+            "  regexp_extract(l.comment, '\\w+') AS l_first_word, " +
+            "  upper(reverse(l.shipinstruct)) AS l_instruct, " +
+            "  regexp_replace(n.comment, '[aeiou]', '*') AS n_redacted, " +
+            "  upper(reverse(n.name)) AS n_reversed, " +
+            "  length(l.comment) + n.nationkey AS mixed " +
+            "FROM lineitem l " +
+            "CROSS JOIN nation n";
+
+    @Test
+    public void benchmark()
+            throws Exception
+    {
+        try (HiveDistributedBenchmarkRunner runner =
+                     new HiveDistributedBenchmarkRunner(3, 5)) {
+            runner.addScenario("baseline", builder -> {
+                builder.setSystemProperty("push_projection_through_cross_join", "false");
+            });
+
+            runner.addScenario("push_projection_through_cross_join", builder -> {
+                builder.setSystemProperty("push_projection_through_cross_join", "true");
+            });
+
+            runner.runWithVerification(QUERY);
+        }
+    }
+
+    public static void main(String[] args)
+            throws Exception
+    {
+        new BenchmarkPushProjectionThroughCrossJoin().benchmark();
+    }
+}

presto-main-base/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -179,6 +179,7 @@ public final class SystemSessionProperties
     public static final String SIMPLIFY_AGGREGATIONS_OVER_CONSTANT = "simplify_aggregations_over_constant";
     public static final String PUSH_PARTIAL_AGGREGATION_THROUGH_JOIN = "push_partial_aggregation_through_join";
     public static final String PRE_AGGREGATE_BEFORE_GROUPING_SETS = "pre_aggregate_before_grouping_sets";
+    public static final String PUSH_PROJECTION_THROUGH_CROSS_JOIN = "push_projection_through_cross_join";
     public static final String PARSE_DECIMAL_LITERALS_AS_DOUBLE = "parse_decimal_literals_as_double";
     public static final String FORCE_SINGLE_NODE_OUTPUT = "force_single_node_output";
     public static final String FILTER_AND_PROJECT_MIN_OUTPUT_PAGE_SIZE = "filter_and_project_min_output_page_size";
@@ -961,6 +962,11 @@ public SystemSessionProperties(
                         "Pre-aggregate data before GroupId node to reduce row multiplication in grouping sets queries",
                         featuresConfig.isPreAggregateBeforeGroupingSets(),
                         false),
+                booleanProperty(
+                        PUSH_PROJECTION_THROUGH_CROSS_JOIN,
+                        "Push projections that reference only one side of a cross join below the join to evaluate on fewer rows",
+                        featuresConfig.isPushProjectionThroughCrossJoin(),
+                        false),
                 booleanProperty(
                         PARSE_DECIMAL_LITERALS_AS_DOUBLE,
                         "Parse decimal literals as DOUBLE instead of DECIMAL",
@@ -2689,6 +2695,11 @@ public static boolean isPreAggregateBeforeGroupingSets(Session session)
         return session.getSystemProperty(PRE_AGGREGATE_BEFORE_GROUPING_SETS, Boolean.class);
     }
 
+    public static boolean isPushProjectionThroughCrossJoin(Session session)
+    {
+        return session.getSystemProperty(PUSH_PROJECTION_THROUGH_CROSS_JOIN, Boolean.class);
+    }
+
     public static boolean isParseDecimalLiteralsAsDouble(Session session)
     {
         return session.getSystemProperty(PARSE_DECIMAL_LITERALS_AS_DOUBLE, Boolean.class);

presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/FeaturesConfig.java

@@ -161,6 +161,7 @@ public class FeaturesConfig
     private boolean pushdownThroughUnnest;
     private boolean simplifyAggregationsOverConstant;
     private boolean preAggregateBeforeGroupingSets;
+    private boolean pushProjectionThroughCrossJoin;
     private double memoryRevokingTarget = 0.5;
     private double memoryRevokingThreshold = 0.9;
     private boolean useMarkDistinct = true;
@@ -1757,6 +1758,18 @@ public FeaturesConfig setPreAggregateBeforeGroupingSets(boolean preAggregateBefo
         return this;
     }
 
+    public boolean isPushProjectionThroughCrossJoin()
+    {
+        return pushProjectionThroughCrossJoin;
+    }
+
+    @Config("optimizer.push-projection-through-cross-join")
+    public FeaturesConfig setPushProjectionThroughCrossJoin(boolean pushProjectionThroughCrossJoin)
+    {
+        this.pushProjectionThroughCrossJoin = pushProjectionThroughCrossJoin;
+        return this;
+    }
+
     public boolean isForceSingleNodeOutput()
     {
         return forceSingleNodeOutput;

presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -106,6 +106,7 @@
 import com.facebook.presto.sql.planner.iterative.rule.PushOffsetThroughProject;
 import com.facebook.presto.sql.planner.iterative.rule.PushPartialAggregationThroughExchange;
 import com.facebook.presto.sql.planner.iterative.rule.PushPartialAggregationThroughJoin;
+import com.facebook.presto.sql.planner.iterative.rule.PushProjectionThroughCrossJoin;
 import com.facebook.presto.sql.planner.iterative.rule.PushProjectionThroughExchange;
 import com.facebook.presto.sql.planner.iterative.rule.PushProjectionThroughUnion;
 import com.facebook.presto.sql.planner.iterative.rule.PushRemoteExchangeThroughAssignUniqueId;
@@ -362,6 +363,7 @@ public PlanOptimizers(
                 ImmutableSet.of(
                         new PushProjectionThroughUnion(),
                         new PushProjectionThroughExchange(),
+                        new PushProjectionThroughCrossJoin(metadata.getFunctionAndTypeManager()),
                         new PushdownThroughUnnest(metadata.getFunctionAndTypeManager())));
 
         IterativeOptimizer simplifyRowExpressionOptimizer = new IterativeOptimizer(