update plans

Author: comphead

native/core/src/execution/planner.rs

@@ -967,21 +967,34 @@ impl PhysicalPlanner {
                 let group_by = PhysicalGroupBy::new_single(group_exprs?);
                 let schema = child.schema();
 
+                let has_mixed_partial_merge = !agg.expr_modes.is_empty()
+                    && agg.expr_modes.contains(&2)
+                    && agg.expr_modes.iter().any(|&m| m != 2);
+
                 let mode = match agg.mode {
-                    0 => DFAggregateMode::Partial,
+                    0 => {
+                        if has_mixed_partial_merge {
+                            // Mixed {Partial, PartialMerge}: use Partial mode, wrap
+                            // PartialMerge expressions with MergeAsPartial.
+                            DFAggregateMode::Partial
+                        } else {
+                            DFAggregateMode::Partial
+                        }
+                    }
                     1 => DFAggregateMode::Final,
-                    2 => DFAggregateMode::Partial, // PartialMerge: Partial + MergeAsPartial
+                    // Uniform PartialMerge maps directly to DataFusion's PartialReduce
+                    // which has merge input + state output semantics.
+                    2 => DFAggregateMode::PartialReduce,
                     other => {
                         return Err(ExecutionError::GeneralError(format!(
                             "Unsupported aggregate mode: {other}"
                         )))
                     }
                 };
 
-                // Check if any expression uses PartialMerge mode (2). When present,
-                // those expressions are wrapped with MergeAsPartial to get merge
-                // semantics inside a Partial-mode AggregateExec.
-                let has_partial_merge = agg.mode == 2 || agg.expr_modes.contains(&2);
+                // For mixed mode, PartialMerge expressions need MergeAsPartial wrappers
+                // to get merge semantics inside a Partial-mode AggregateExec.
+                let has_partial_merge = has_mixed_partial_merge;
 
                 let agg_exprs: PhyAggResult = agg
                     .agg_exprs

spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala

@@ -712,7 +712,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
         "spark.comet.exec.shuffle.fallbackToColumnar" -> "false",
         "spark.comet.cast.allowIncompatible" -> "true",
-        "spark.sql.adaptive.enabled" -> "false",
+        "spark.sql.adaptive.enabled" -> "true",
         "spark.comet.enabled" -> "true",
         "spark.comet.expression.Cast.allowIncompatible" -> "true",
         CometConf.COMET_NATIVE_SCAN_IMPL.key -> "native_iceberg_compat") {