[fix](local shuffle) Require hash input for distinct finalize agg without group keys

fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java

@@ -335,7 +335,13 @@ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
             // early return also catches FIRST_MERGE, dropping the HASH requirement and
             // causing wrong-result (e.g. PASSTHROUGH over serial child breaks the
             // group-by-key invariant — DORIS-25413).
-            if (!hasKeys) {
+            if (!hasPartitionRequirement()) {
+                // No group keys and no DISTINCT aggregates (e.g. COUNT(*)): the input
+                // distribution is irrelevant. Any agg with group or distinct keys keeps
+                // its partition requirement — a finalize agg emits per-instance scalar
+                // values (sum0(multi_distinct_count(...)) above) that the parent sums,
+                // so same-key rows must stay in a single instance. This mirrors BE's
+                // `_partition_exprs` (grouping exprs, or distinct/distribute exprs).
                 requireChild = needsFinalize
                         ? LocalExchangeTypeRequire.noRequire()
                         : baseClassRequire(connectContext);
@@ -348,13 +354,16 @@ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
                 // FIRST_MERGE (correctness) or finalize+colocate → HASH.
                 requireChild = parentRequire.autoRequireHash();
             } else if (hasPartitionExprs(parentRequire)) {
-                // FE-only heuristic: finalize non-colocate with parent hash requirement
-                // → inherit parent's specific hash type.
+                // finalize non-colocate with a parent hash requirement → inherit the
+                // parent's specific hash type.
                 requireChild = parentRequire.autoRequireHash();
             } else {
-                // FE-only heuristic: finalize non-colocate without parent hash → skip
-                // LE (child Exchange already provides hash distribution).
-                requireChild = LocalExchangeTypeRequire.noRequire();
+                // finalize non-colocate without a parent hash requirement: the input
+                // must still be key-aligned (group/distinct key), so require HASH
+                // explicitly instead of trusting the child's distribution. When the
+                // child already provides hash distribution, satisfy() passes and no
+                // LE is inserted, so this is safe and free in the common case.
+                requireChild = LocalExchangeTypeRequire.requireHash();
             }
         }
 
@@ -371,6 +380,32 @@ private LocalExchangeTypeRequire baseClassRequire(ConnectContext connectContext)
                 : LocalExchangeTypeRequire.noRequire();
     }
 
+    /**
+     * Whether this agg needs key-aligned (hash-partitioned) input from its child.
+     * Mirrors BE AggSinkOperatorX::update_operator's `_partition_exprs`: partition
+     * exprs are the grouping exprs, or the distinct/distribute exprs when the agg
+     * has DISTINCT functions. A finalize agg that emits per-instance scalar values
+     * (e.g. the sum0(multi_distinct_count(...)) above) is only correct when
+     * same-key rows stay in a single instance — overlapping keys across instances
+     * get summed multiple times by the parent.
+     */
+    private boolean hasPartitionRequirement() {
+        return !aggInfo.getGroupingExprs().isEmpty() || hasDistinctAggregate();
+    }
+
+    private boolean hasDistinctAggregate() {
+        // Multi-distinct aggregates are detected by function name. Nereids rewrites
+        // count/sum/group_concat(distinct ...) into dedicated MultiDistinct* functions
+        // constructed with distinct=false, so by this legacy FunctionCallExpr layer
+        // isDistinct() is already false and the function name is the only signal.
+        return aggInfo.getAggregateExprs().stream()
+                .map(FunctionCallExpr::getFnName)
+                .filter(name -> name != null)
+                .map(name -> name.getFunction())
+                .filter(name -> name != null)
+                .anyMatch(name -> name.startsWith("multi_distinct_"));
+    }
+
     @Override
     protected List<Expr> getSemanticPartitionExprs() {
         return aggInfo.getGroupingExprs();
@@ -386,18 +421,7 @@ protected List<Expr> getLocalExchangeDistributeExprs(int childIndex, boolean fol
         // chain scatters same-group rows across N instances, leaving partial_preagg essentially a
         // no-op and breaking row-arrival order at downstream merge-finalize (e.g. group_concat).
         List<Expr> childDist = getChildDistributeExprList(childIndex);
-        // Multi-distinct aggregates are detected by function name. Nereids rewrites
-        // count/sum(distinct ...) into dedicated MultiDistinct* functions constructed with
-        // distinct=false and a "multi_distinct_" name, so by this legacy FunctionCallExpr layer
-        // isDistinct() is already false and the function name is the only remaining signal —
-        // there is no structural flag to test here.
-        boolean hasDistinct = aggInfo.getAggregateExprs().stream()
-                .map(FunctionCallExpr::getFnName)
-                .filter(name -> name != null)
-                .map(name -> name.getFunction())
-                .filter(name -> name != null)
-                .anyMatch(name -> name.startsWith("multi_distinct_"));
-        if (childDist != null && !childDist.isEmpty() && (followedByShuffled || hasDistinct)) {
+        if (childDist != null && !childDist.isEmpty() && (followedByShuffled || hasDistinctAggregate())) {
             return childDist;
         }
         return Lists.newArrayList(aggInfo.getGroupingExprs());
@@ -406,13 +430,14 @@ protected List<Expr> getLocalExchangeDistributeExprs(int childIndex, boolean fol
     @Override
     public boolean requiresShuffleForCorrectness() {
         // Mirrors BE's AggSinkOperatorX::is_shuffled_operator() exactly:
-        //   finalize agg with group keys needs hash-distributed input for correctness.
+        //   finalize agg with partition exprs (group keys or DISTINCT aggregates)
+        //   needs hash-distributed input for correctness.
         // GLOBAL dedup (!needsFinalize) is intentionally NOT included here — if a
         // GLOBAL dedup exists, a finalize agg always sits above it (e.g. DISTINCT_GLOBAL
         // above DISTINCT_LOCAL/GLOBAL_DEDUP), and the finalize agg propagates the flag
         // down via inheritedShuffled. A solo finalize agg satisfies hash distribution
         // through its own child requirement.
-        return needsFinalize && !aggInfo.getGroupingExprs().isEmpty();
+        return needsFinalize && hasPartitionRequirement();
     }
 
     private boolean canUseDistinctStreamingAgg(SessionVariable sessionVariable) {