feat(planner): Add feature config for PushPartialAggregationThroughJoin

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

@@ -370,6 +370,19 @@ queries that have very selective joins.
 
 The corresponding configuration property is :ref:`admin/properties:\`\`optimizer.push-aggregation-through-join\`\``.
 
+``push_partial_aggregation_through_join``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+When a partial aggregation is above an inner join and all aggregation inputs come from
+only one side of the join, the partial aggregation is pushed below the join to that side.
+This reduces the amount of data flowing into the join operator, which can improve
+performance by allowing the aggregation to pre-reduce data before the join is performed.
+
+The corresponding configuration property is :ref:`admin/properties:\`\`optimizer.push-partial-aggregation-through-join\`\``.
+
 ``push_table_write_through_union``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

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

@@ -944,6 +944,19 @@ queries that have very selective joins.
 
 The corresponding session property is :ref:`admin/properties-session:\`\`push_aggregation_through_join\`\``.
 
+``optimizer.push-partial-aggregation-through-join``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+When a partial aggregation is above an inner join and all aggregation inputs come from
+only one side of the join, the partial aggregation is pushed below the join to that side.
+This reduces the amount of data flowing into the join operator, which can improve
+performance by allowing the aggregation to pre-reduce data before the join is performed.
+
+The corresponding session property is :ref:`admin/properties-session:\`\`push_partial_aggregation_through_join\`\``.
+
 ``optimizer.push-table-write-through-union``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

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

@@ -939,7 +939,7 @@ public SystemSessionProperties(
                 booleanProperty(
                         PUSH_PARTIAL_AGGREGATION_THROUGH_JOIN,
                         "Push partial aggregations below joins",
-                        false,
+                        featuresConfig.isPushPartialAggregationThroughJoin(),
                         false),
                 booleanProperty(
                         PARSE_DECIMAL_LITERALS_AS_DOUBLE,

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

@@ -155,6 +155,7 @@ public class FeaturesConfig
     private double defaultJoinSelectivityCoefficient;
     private double defaultWriterReplicationCoefficient = 3;
     private boolean pushAggregationThroughJoin = true;
+    private boolean pushPartialAggregationThroughJoin;
     private boolean pushSemiJoinThroughUnion;
     private boolean pushdownThroughUnnest;
     private double memoryRevokingTarget = 0.5;
@@ -1678,6 +1679,19 @@ public FeaturesConfig setPushAggregationThroughJoin(boolean value)
         return this;
     }
 
+    public boolean isPushPartialAggregationThroughJoin()
+    {
+        return pushPartialAggregationThroughJoin;
+    }
+
+    @Config("optimizer.push-partial-aggregation-through-join")
+    @ConfigDescription("Push partial aggregations below joins")
+    public FeaturesConfig setPushPartialAggregationThroughJoin(boolean pushPartialAggregationThroughJoin)
+    {
+        this.pushPartialAggregationThroughJoin = pushPartialAggregationThroughJoin;
+        return this;
+    }
+
     public boolean isPushSemiJoinThroughUnion()
     {
         return pushSemiJoinThroughUnion;

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

@@ -136,6 +136,7 @@ public void testDefaults()
                 .setExchangeChecksumEnabled(false)
                 .setEnableIntermediateAggregations(false)
                 .setPushAggregationThroughJoin(true)
+                .setPushPartialAggregationThroughJoin(false)
                 .setPushSemiJoinThroughUnion(false)
                 .setPushdownThroughUnnest(false)
                 .setForceSingleNodeOutput(true)
@@ -358,6 +359,7 @@ public void testExplicitPropertyMappings()
                 .put("optimizer.retry-query-with-history-based-optimization", "true")
                 .put("optimizer.treat-low-confidence-zero-estimation-as-unknown", "true")
                 .put("optimizer.push-aggregation-through-join", "false")
+                .put("optimizer.push-partial-aggregation-through-join", "true")
                 .put("optimizer.push-semi-join-through-union", "true")
                 .put("optimizer.pushdown-through-unnest", "true")
                 .put("optimizer.aggregation-partition-merging", "top_down")
@@ -765,6 +767,7 @@ public void testExplicitPropertyMappings()
                 .setSkipPushdownThroughExchangeForRemoteProjection(true)
                 .setUseConnectorProvidedSerializationCodecs(true)
                 .setRemoteFunctionNamesForFixedParallelism("remote_.*")
+                .setPushPartialAggregationThroughJoin(true)
                 .setRemoteFunctionFixedParallelismTaskCount(100);
         assertFullMapping(properties, expected);
     }