[WIP] adjust agg pushdown strategy

Signed-off-by: stephen <[email protected]>
StarRocks · Dec 31, 2024 · a33f129 · a33f129
1 parent 3f07430
commit a33f129
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 13 deletions.
diff --git a/be/src/exec/aggregator.cpp b/be/src/exec/aggregator.cpp
@@ -1359,6 +1359,7 @@ void Aggregator::_init_agg_hash_variant(HashVariantType& hash_variant) {
             }
         }
     }
+
     VLOG_ROW << "hash type is "
              << static_cast<typename std::underlying_type<typename HashVariantType::Type>::type>(type);
     hash_variant.init(_state, type, _agg_stat);

diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java b/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
@@ -364,6 +364,8 @@ public class SessionVariable implements Serializable, Writable, Cloneable {
     public static final String CBO_PRUNE_SHUFFLE_COLUMN_RATE = "cbo_prune_shuffle_column_rate";
     public static final String CBO_PUSH_DOWN_AGGREGATE_MODE = "cbo_push_down_aggregate_mode";
     public static final String CBO_PUSH_DOWN_AGGREGATE_ON_BROADCAST_JOIN = "cbo_push_down_aggregate_on_broadcast_join";
+    public static final String CBO_PUSH_DOWN_AGGREGATE_ON_BROADCAST_JOIN_ROW_COUNT_LIMIT =
+            "cbo_push_down_aggregate_on_broadcast_join_row_count_limit";
 
     public static final String CBO_PUSH_DOWN_DISTINCT_BELOW_WINDOW = "cbo_push_down_distinct_below_window";
     public static final String CBO_PUSH_DOWN_AGGREGATE = "cbo_push_down_aggregate";
@@ -1532,6 +1534,9 @@ public static MaterializedViewRewriteMode parse(String str) {
     @VarAttr(name = CBO_PUSH_DOWN_AGGREGATE_ON_BROADCAST_JOIN, flag = VariableMgr.INVISIBLE)
     private boolean cboPushDownAggregateOnBroadcastJoin = true;
 
+    @VarAttr(name = CBO_PUSH_DOWN_AGGREGATE_ON_BROADCAST_JOIN_ROW_COUNT_LIMIT, flag = VariableMgr.INVISIBLE)
+    private long cboPushDownAggregateOnBroadcastJoinRowCountLimit = 250000;
+
     // auto, global, local
     @VarAttr(name = CBO_PUSH_DOWN_AGGREGATE, flag = VariableMgr.INVISIBLE)
     private String cboPushDownAggregate = "global";
@@ -3595,6 +3600,14 @@ public void setCboPushDownAggregateOnBroadcastJoin(boolean cboPushDownAggregateO
         this.cboPushDownAggregateOnBroadcastJoin = cboPushDownAggregateOnBroadcastJoin;
     }
 
+    public long getCboPushDownAggregateOnBroadcastJoinRowCountLimit() {
+        return cboPushDownAggregateOnBroadcastJoinRowCountLimit;
+    }
+
+    public void setCboPushDownAggregateOnBroadcastJoinRowCountLimit(long cboPushDownAggregateOnBroadcastJoinRowCountLimit) {
+        this.cboPushDownAggregateOnBroadcastJoinRowCountLimit = cboPushDownAggregateOnBroadcastJoinRowCountLimit;
+    }
+
     public String getCboPushDownAggregate() {
         return cboPushDownAggregate;
     }

diff --git a/...src/main/java/com/starrocks/sql/optimizer/rule/tree/pdagg/PushDownAggregateCollector.java b/...src/main/java/com/starrocks/sql/optimizer/rule/tree/pdagg/PushDownAggregateCollector.java
@@ -467,6 +467,11 @@ private boolean checkStatistics(AggregatePushDownContext context, ColumnRefSet g
             return false;
         }
 
+        // target is the immediate child of a small broadcast join
+        if (pushDownMode == PUSH_DOWN_AGG_AUTO && context.immediateChildOfSmallBroadcastJoin) {
+            return false;
+        }
+
         List<ColumnStatistic> lower = Lists.newArrayList();
         List<ColumnStatistic> medium = Lists.newArrayList();
         List<ColumnStatistic> high = Lists.newArrayList();
@@ -516,15 +521,9 @@ private boolean checkStatistics(AggregatePushDownContext context, ColumnRefSet g
             }
         }
 
-        // 2. forbidden rules
-        // 2.1 target is the immediate child of a small broadcast join and the cardinality of the aggregation is not lower.
-        if (pushDownMode == PUSH_DOWN_AGG_AUTO && context.immediateChildOfSmallBroadcastJoin) {
-            return false;
-        }
-
-        // 2.2 high cardinality >= 2
-        // 2.3 medium cardinality > 2
-        // 2.4 high cardinality = 1 and medium cardinality > 0
+        // 2.1 high cardinality >= 2
+        // 2.2 medium cardinality > 2
+        // 2.3 high cardinality = 1 and medium cardinality > 0
         if (high.size() >= 2 || medium.size() > 2 || (high.size() == 1 && !medium.isEmpty())) {
             return false;
         }
@@ -553,9 +552,9 @@ private boolean checkStatistics(AggregatePushDownContext context, ColumnRefSet g
         return false;
     }
 
-    // high(2): cardinality/count > MEDIUM_AGGREGATE
-    // medium(1): cardinality/count <= MEDIUM_AGGREGATE and > LOW_AGGREGATE
-    // lower(0): cardinality/count < LOW_AGGREGATE
+    // high(2): row_count / cardinality < MEDIUM_AGGREGATE_EFFECT_COEFFICIENT
+    // medium(1): row_count / cardinality >= MEDIUM_AGGREGATE_EFFECT_COEFFICIENT and < LOW_AGGREGATE_EFFECT_COEFFICIENT
+    // lower(0): row_count / cardinality >= LOW_AGGREGATE_EFFECT_COEFFICIENT
     private int groupByCardinality(ColumnStatistic statistic, double rowCount) {
         if (statistic.isUnknown()) {
             return 2;
@@ -586,7 +585,7 @@ private boolean isSmallBroadcastJoin(OptExpression optExpression) {
         }
         double rightRows = rightStatistics.getOutputRowCount();
         return rightRows <= sessionVariable.getBroadcastRowCountLimit() &&
-                rightRows <= StatisticsEstimateCoefficient.SMALL_BROADCAST_JOIN_ROW_COUNT_UPPER_BOUND;
+                rightRows <= sessionVariable.getCboPushDownAggregateOnBroadcastJoinRowCountLimit();
     }
 
     /**
-Original file line number
+Diff line change
@@ Expand Up @@
                 }
             }
         }
         VLOG_ROW << "hash type is "
                  << static_cast<typename std::underlying_type<typename HashVariantType::Type>::type>(type);
         hash_variant.init(_state, type, _agg_stat);
@@ Expand Down @@