[BugFix] fix feedback npe when processing local agg

Signed-off-by: stephen <[email protected]>
StarRocks · Nov 12, 2024 · c1f762d · c1f762d
1 parent bb07a54
commit c1f762d
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 7 deletions.
diff --git a/be/src/exec/aggregate/aggregate_blocking_node.cpp b/be/src/exec/aggregate/aggregate_blocking_node.cpp
@@ -271,6 +271,8 @@ pipeline::OpFactories AggregateBlockingNode::decompose_to_pipeline(pipeline::Pip
             if (!has_group_by_keys) {
                 ops_with_sink =
                         context->maybe_interpolate_local_passthrough_exchange(runtime_state(), id(), ops_with_sink);
+            } else if (agg_node.local_not_split) {
+                // ignore
             } else if (could_local_shuffle) {
                 ops_with_sink = try_interpolate_local_shuffle(ops_with_sink);
             }

diff --git a/fe/fe-core/src/main/java/com/starrocks/planner/AggregationNode.java b/fe/fe-core/src/main/java/com/starrocks/planner/AggregationNode.java
@@ -101,6 +101,8 @@ public class AggregationNode extends PlanNode {
     // OlapScanNode and these PlanNodes have the same data partition policy.
     private boolean identicallyDistributed = false;
 
+    private boolean locallNotSplit = false;
+
     /**
      * Create an agg node that is not an intermediate node.
      * isIntermediate is true if it is a slave node in a 2-part agg plan.
@@ -121,6 +123,14 @@ public void unsetNeedsFinalize() {
         updateplanNodeName();
     }
 
+    public boolean isLocallNotSplit() {
+        return locallNotSplit;
+    }
+
+    public void setLocallNotSplit(boolean locallNotSplit) {
+        this.locallNotSplit = locallNotSplit;
+    }
+
     /**
      * Sets this node as a preaggregation. Only valid to call this if it is not marked
      * as a preaggregation
@@ -235,7 +245,7 @@ protected void toThrift(TPlanNode msg) {
         }
         msg.agg_node.setUse_sort_agg(useSortAgg);
         msg.agg_node.setUse_per_bucket_optimize(usePerBucketOptimize);
-
+        msg.agg_node.setLocal_not_split(locallNotSplit);
         List<Expr> groupingExprs = aggInfo.getGroupingExprs();
         if (groupingExprs != null) {
             msg.agg_node.setGrouping_exprs(Expr.treesToThrift(groupingExprs));

diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/feedback/analyzer/StreamingAggTuningAnalyzer.java b/fe/fe-core/src/main/java/com/starrocks/qe/feedback/analyzer/StreamingAggTuningAnalyzer.java
@@ -79,11 +79,13 @@ public Void visitPhysicalHashAggregate(OptExpression optExpression, Void context
                 double inputRows = skeletonNode.getNodeExecStats().getPushRows();
                 double streamingOutputRows = skeletonNode.getNodeExecStats().getPullRows();
                 BlockingAggNode blockingAggNode = findBlockingAggNode(skeletonNode);
-                double blockingOutputRows = blockingAggNode.getNodeExecStats().getPullRows();
-                if (blockingOutputRows < inputRows && (inputRows / streamingOutputRows) < STREAMING_AGGREGATION_THRESHOLD
-                        && (inputRows / blockingOutputRows) > AGGREGATION_THRESHOLD) {
-                    tuningGuides.addTuningGuide(skeletonNode.getNodeId(),
-                            new StreamingAggTuningGuide((StreamingAggNode) skeletonNode));
+                if (blockingAggNode != null) {
+                    double blockingOutputRows = blockingAggNode.getNodeExecStats().getPullRows();
+                    if (blockingOutputRows < inputRows && (inputRows / streamingOutputRows) < STREAMING_AGGREGATION_THRESHOLD
+                            && (inputRows / blockingOutputRows) > AGGREGATION_THRESHOLD) {
+                        tuningGuides.addTuningGuide(skeletonNode.getNodeId(),
+                                new StreamingAggTuningGuide((StreamingAggNode) skeletonNode));
+                    }
                 }
             }
             visit(optExpression, context);

diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/feedback/skeleton/SkeletonBuilder.java b/fe/fe-core/src/main/java/com/starrocks/qe/feedback/skeleton/SkeletonBuilder.java
@@ -97,7 +97,7 @@ public SkeletonNode visitPhysicalJoin(OptExpression optExpression, SkeletonNode
     public SkeletonNode visitPhysicalHashAggregate(OptExpression optExpression, SkeletonNode parent) {
         int planNodeId = optExpression.getOp().getPlanNodeId();
         PhysicalHashAggregateOperator aggOperator = (PhysicalHashAggregateOperator) optExpression.getOp();
-        if (aggOperator.getType().isAnyGlobal()) {
+        if (aggOperator.getType().isAnyGlobal() || !aggOperator.isSplit()) {
             BlockingAggNode node = new BlockingAggNode(optExpression, nodeExecStatsMap.get(planNodeId), parent);
             visitChildren(node, optExpression.getInputs());
             fillNodeId(optExpression.getOp(), node);

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java b/fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java
@@ -2222,6 +2222,7 @@ public PlanFragment visitPhysicalHashAggregate(OptExpression optExpr, ExecPlan c
                     aggregationNode =
                             new AggregationNode(context.getNextNodeId(), inputFragment.getPlanRoot(),
                                     aggInfo);
+                    aggregationNode.setGlobalNotSplit(true);
                 } else {
                     aggregateExprList.forEach(FunctionCallExpr::setMergeAggFn);
                     AggregateInfo aggInfo = AggregateInfo.create(

diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
@@ -818,6 +818,8 @@ struct TAggregationNode {
 
   // enable runtime limit, pipelines share one limit
   29: optional bool enable_pipeline_share_limit = false
+
+  30: optional bool local_not_split = false
 }
 
 struct TRepeatNode {