You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by co...@apache.org on 2026/05/04 19:59:35 UTC
(datafusion-comet) branch main updated: feat: support `PartialMerge` aggregation mode (#4003)
This is an automated email from the ASF dual-hosted git repository.
comphead pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new da187f2aa feat: support `PartialMerge` aggregation mode (#4003)
da187f2aa is described below
commit da187f2aa4e68e00bb94310b657f61aff99da350
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Mon May 4 12:59:30 2026 -0700
feat: support `PartialMerge` aggregation mode (#4003)
---
native/common/src/error.rs | 15 +-
native/core/src/execution/merge_as_partial.rs | 245 +++++++++++++
native/core/src/execution/mod.rs | 1 +
native/core/src/execution/planner.rs | 89 ++++-
native/proto/src/proto/operator.proto | 8 +
native/spark-expr/src/agg_funcs/avg_decimal.rs | 8 +-
native/spark-expr/src/agg_funcs/sum_decimal.rs | 112 +++---
native/spark-expr/src/agg_funcs/sum_int.rs | 138 ++++----
native/spark-expr/src/lib.rs | 6 +-
.../org/apache/comet/rules/CometExecRule.scala | 6 +-
.../org/apache/comet/rules/CometScanRule.scala | 7 +-
.../org/apache/comet/serde/QueryPlanSerde.scala | 16 +-
.../org/apache/spark/sql/comet/operators.scala | 120 +++++--
.../apache/comet/shims/ShimCometStreaming.scala | 31 ++
.../apache/comet/shims/ShimCometStreaming.scala | 29 ++
.../sql/comet/shims/ShimSparkErrorConverter.scala | 5 +-
.../expressions/aggregate/partial_merge.sql | 380 +++++++++++++++++++++
.../q16.native_datafusion/extended.txt | 79 +++--
.../q16.native_iceberg_compat/extended.txt | 79 +++--
.../q28.native_datafusion/extended.txt | 104 +++---
.../q28.native_iceberg_compat/extended.txt | 104 +++---
.../q94.native_datafusion/extended.txt | 79 +++--
.../q94.native_iceberg_compat/extended.txt | 79 +++--
.../q95.native_datafusion/extended.txt | 123 ++++---
.../q95.native_iceberg_compat/extended.txt | 123 ++++---
.../q16.native_datafusion/extended.txt | 79 +++--
.../q16.native_iceberg_compat/extended.txt | 79 +++--
.../q28.native_datafusion/extended.txt | 104 +++---
.../q28.native_iceberg_compat/extended.txt | 104 +++---
.../q94.native_datafusion/extended.txt | 79 +++--
.../q94.native_iceberg_compat/extended.txt | 79 +++--
.../q95.native_datafusion/extended.txt | 123 ++++---
.../q95.native_iceberg_compat/extended.txt | 123 ++++---
.../apache/comet/exec/CometAggregateSuite.scala | 133 ++++++--
.../org/apache/comet/exec/CometExecSuite.scala | 6 +-
35 files changed, 1892 insertions(+), 1003 deletions(-)
diff --git a/native/common/src/error.rs b/native/common/src/error.rs
index 5dff9bd86..ff9059571 100644
--- a/native/common/src/error.rs
+++ b/native/common/src/error.rs
@@ -72,8 +72,8 @@ pub enum SparkError {
#[error("[ARITHMETIC_OVERFLOW] Overflow in integral divide. Use `try_divide` to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
IntegralDivideOverflow,
- #[error("[ARITHMETIC_OVERFLOW] Overflow in sum of decimals. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
- DecimalSumOverflow,
+ #[error("[ARITHMETIC_OVERFLOW] Overflow in sum of decimals. Use `try_{function_name}` to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
+ DecimalSumOverflow { function_name: String },
#[error("[DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
DivideByZero,
@@ -240,7 +240,7 @@ impl SparkError {
SparkError::CannotParseDecimal => "CannotParseDecimal",
SparkError::ArithmeticOverflow { .. } => "ArithmeticOverflow",
SparkError::IntegralDivideOverflow => "IntegralDivideOverflow",
- SparkError::DecimalSumOverflow => "DecimalSumOverflow",
+ SparkError::DecimalSumOverflow { .. } => "DecimalSumOverflow",
SparkError::DivideByZero => "DivideByZero",
SparkError::RemainderByZero => "RemainderByZero",
SparkError::IntervalDividedByZero => "IntervalDividedByZero",
@@ -336,6 +336,11 @@ impl SparkError {
"fromType": from_type,
})
}
+ SparkError::DecimalSumOverflow { function_name } => {
+ serde_json::json!({
+ "functionName": function_name,
+ })
+ }
SparkError::BinaryArithmeticOverflow {
value1,
symbol,
@@ -523,7 +528,7 @@ impl SparkError {
| SparkError::NumericOutOfRange { .. } // Comet-specific extension
| SparkError::ArithmeticOverflow { .. }
| SparkError::IntegralDivideOverflow
- | SparkError::DecimalSumOverflow
+ | SparkError::DecimalSumOverflow { .. }
| SparkError::BinaryArithmeticOverflow { .. }
| SparkError::IntervalArithmeticOverflowWithSuggestion { .. }
| SparkError::IntervalArithmeticOverflowWithoutSuggestion
@@ -601,7 +606,7 @@ impl SparkError {
SparkError::IntervalDividedByZero => Some("INTERVAL_DIVIDED_BY_ZERO"),
SparkError::ArithmeticOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
SparkError::IntegralDivideOverflow => Some("ARITHMETIC_OVERFLOW"),
- SparkError::DecimalSumOverflow => Some("ARITHMETIC_OVERFLOW"),
+ SparkError::DecimalSumOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
SparkError::BinaryArithmeticOverflow { .. } => Some("BINARY_ARITHMETIC_OVERFLOW"),
SparkError::IntervalArithmeticOverflowWithSuggestion { .. } => {
Some("INTERVAL_ARITHMETIC_OVERFLOW")
diff --git a/native/core/src/execution/merge_as_partial.rs b/native/core/src/execution/merge_as_partial.rs
new file mode 100644
index 000000000..1c15ce832
--- /dev/null
+++ b/native/core/src/execution/merge_as_partial.rs
@@ -0,0 +1,245 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! MergeAsPartial wrapper for implementing Spark's PartialMerge aggregate mode.
+//!
+//! Spark's PartialMerge mode merges intermediate state buffers and outputs intermediate
+//! state (not final values). DataFusion has no equivalent mode — `Partial` calls
+//! `update_batch` and outputs state, while `Final` calls `merge_batch` and outputs
+//! evaluated results.
+//!
+//! This wrapper bridges the gap: it operates under DataFusion's `Partial` mode (which
+//! outputs state) but redirects `update_batch` calls to `merge_batch`, giving merge
+//! semantics with state output.
+
+use std::any::Any;
+use std::fmt::Debug;
+use std::hash::{Hash, Hasher};
+
+use arrow::array::{ArrayRef, BooleanArray};
+use arrow::datatypes::{DataType, FieldRef};
+use datafusion::common::Result;
+use datafusion::logical_expr::function::AccumulatorArgs;
+use datafusion::logical_expr::function::StateFieldsArgs;
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDF, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF,
+ Signature, Volatility,
+};
+use datafusion::physical_expr::aggregate::AggregateFunctionExpr;
+use datafusion::scalar::ScalarValue;
+
+/// An AggregateUDF wrapper that gives merge semantics in Partial mode.
+///
+/// When DataFusion runs an AggregateExec in Partial mode, it calls `update_batch`
+/// on each accumulator and outputs `state()`. This wrapper intercepts `update_batch`
+/// and redirects it to `merge_batch` on the inner accumulator, effectively
+/// implementing PartialMerge: merge inputs, output state.
+///
+/// We store the inner AggregateUDF (not the AggregateFunctionExpr) to avoid keeping
+/// references to UnboundColumn expressions that would panic if evaluated.
+#[derive(Debug)]
+pub struct MergeAsPartialUDF {
+ /// The inner aggregate UDF, cloned from the original expression.
+ inner_udf: AggregateUDF,
+ /// Pre-computed return type from the original expression.
+ return_type: DataType,
+ /// Pre-computed state fields from the original expression.
+ cached_state_fields: Vec<FieldRef>,
+ /// Cached signature that accepts state field types.
+ signature: Signature,
+ /// Name for this wrapper.
+ name: String,
+}
+
+impl PartialEq for MergeAsPartialUDF {
+ fn eq(&self, other: &Self) -> bool {
+ self.name == other.name
+ }
+}
+
+impl Eq for MergeAsPartialUDF {}
+
+impl Hash for MergeAsPartialUDF {
+ fn hash<H: Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ }
+}
+
+impl MergeAsPartialUDF {
+ pub fn new(inner_expr: &AggregateFunctionExpr) -> Result<Self> {
+ let name = format!("merge_as_partial_{}", inner_expr.name());
+ let return_type = inner_expr.field().data_type().clone();
+ let cached_state_fields = inner_expr.state_fields()?;
+
+ // Use a permissive signature since we accept state field types which
+ // vary per aggregate function.
+ let signature = Signature::variadic_any(Volatility::Immutable);
+
+ Ok(Self {
+ inner_udf: inner_expr.fun().clone(),
+ return_type,
+ cached_state_fields,
+ signature,
+ name,
+ })
+ }
+}
+
+impl AggregateUDFImpl for MergeAsPartialUDF {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ // In Partial mode, return_type isn't used for output schema (state_fields is).
+ // Return the inner function's return type for consistency.
+ Ok(self.return_type.clone())
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // Cached at construction: state schema depends on the inner aggregate's
+ // return type, not on StateFieldsArgs.
+ Ok(self.cached_state_fields.clone())
+ }
+
+ fn accumulator(&self, args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
+ // args.exprs are state-typed (match this wrapper's signature), not the
+ // inner aggregate's original inputs. Safe for built-ins (SUM/COUNT/
+ // MIN/MAX/AVG) which build accumulators from return_type; aggregates
+ // that inspect args.exprs types would need reconsideration.
+ let inner_acc = self.inner_udf.accumulator(args)?;
+ Ok(Box::new(MergeAsPartialAccumulator { inner: inner_acc }))
+ }
+
+ fn groups_accumulator_supported(&self, args: AccumulatorArgs) -> bool {
+ // See `accumulator`: args.exprs are state-typed.
+ self.inner_udf.groups_accumulator_supported(args)
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ // See `accumulator`: args.exprs are state-typed.
+ let inner_acc = self.inner_udf.create_groups_accumulator(args)?;
+ Ok(Box::new(MergeAsPartialGroupsAccumulator {
+ inner: inner_acc,
+ }))
+ }
+
+ fn reverse_expr(&self) -> ReversedUDAF {
+ ReversedUDAF::NotSupported
+ }
+
+ fn default_value(&self, data_type: &DataType) -> Result<ScalarValue> {
+ ScalarValue::try_from(data_type)
+ }
+
+ fn is_descending(&self) -> Option<bool> {
+ None
+ }
+}
+
+/// Accumulator wrapper that redirects update_batch to merge_batch.
+struct MergeAsPartialAccumulator {
+ inner: Box<dyn Accumulator>,
+}
+
+impl Debug for MergeAsPartialAccumulator {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("MergeAsPartialAccumulator").finish()
+ }
+}
+
+impl Accumulator for MergeAsPartialAccumulator {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ // Redirect update to merge — this is the key trick.
+ self.inner.merge_batch(values)
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ self.inner.merge_batch(states)
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ self.inner.evaluate()
+ }
+
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ self.inner.state()
+ }
+
+ fn size(&self) -> usize {
+ self.inner.size()
+ }
+}
+
+/// GroupsAccumulator wrapper that redirects update_batch to merge_batch.
+struct MergeAsPartialGroupsAccumulator {
+ inner: Box<dyn GroupsAccumulator>,
+}
+
+impl Debug for MergeAsPartialGroupsAccumulator {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("MergeAsPartialGroupsAccumulator").finish()
+ }
+}
+
+impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
+ fn update_batch(
+ &mut self,
+ values: &[ArrayRef],
+ group_indices: &[usize],
+ opt_filter: Option<&BooleanArray>,
+ total_num_groups: usize,
+ ) -> Result<()> {
+ // Redirect update to merge — this is the key trick.
+ self.inner
+ .merge_batch(values, group_indices, opt_filter, total_num_groups)
+ }
+
+ fn merge_batch(
+ &mut self,
+ values: &[ArrayRef],
+ group_indices: &[usize],
+ opt_filter: Option<&BooleanArray>,
+ total_num_groups: usize,
+ ) -> Result<()> {
+ self.inner
+ .merge_batch(values, group_indices, opt_filter, total_num_groups)
+ }
+
+ fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {
+ self.inner.evaluate(emit_to)
+ }
+
+ fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
+ self.inner.state(emit_to)
+ }
+
+ fn size(&self) -> usize {
+ self.inner.size()
+ }
+}
diff --git a/native/core/src/execution/mod.rs b/native/core/src/execution/mod.rs
index f556fce41..ec247f72b 100644
--- a/native/core/src/execution/mod.rs
+++ b/native/core/src/execution/mod.rs
@@ -19,6 +19,7 @@
pub mod columnar_to_row;
pub mod expressions;
pub mod jni_api;
+pub(crate) mod merge_as_partial;
pub(crate) mod metrics;
pub mod operators;
pub(crate) mod planner;
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index ef0250bab..5522e5b90 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -116,8 +116,8 @@ use datafusion_comet_proto::{
},
spark_operator::{
self, lower_window_frame_bound::LowerFrameBoundStruct, operator::OpStruct,
- upper_window_frame_bound::UpperFrameBoundStruct, BuildSide,
- CompressionCodec as SparkCompressionCodec, JoinType, Operator, WindowFrameType,
+ upper_window_frame_bound::UpperFrameBoundStruct, AggregateMode as ProtoAggregateMode,
+ BuildSide, CompressionCodec as SparkCompressionCodec, JoinType, Operator, WindowFrameType,
},
spark_partitioning::{partitioning::PartitioningStruct, Partitioning as SparkPartitioning},
};
@@ -985,19 +985,94 @@ impl PhysicalPlanner {
let group_by = PhysicalGroupBy::new_single(group_exprs?);
let schema = child.schema();
- let mode = if agg.mode == 0 {
- DFAggregateMode::Partial
- } else {
- DFAggregateMode::Final
+ let proto_mode = ProtoAggregateMode::try_from(agg.mode).map_err(|_| {
+ ExecutionError::GeneralError(format!(
+ "Unsupported aggregate mode: {}",
+ agg.mode
+ ))
+ })?;
+ let mode = match proto_mode {
+ ProtoAggregateMode::Partial => DFAggregateMode::Partial,
+ ProtoAggregateMode::Final => DFAggregateMode::Final,
+ // PartialMerge: Partial + MergeAsPartial
+ ProtoAggregateMode::PartialMerge => DFAggregateMode::Partial,
};
+ // Check if any expression uses PartialMerge mode. When present,
+ // those expressions are wrapped with MergeAsPartial to get merge
+ // semantics inside a Partial-mode AggregateExec.
+ let partial_merge_value = ProtoAggregateMode::PartialMerge as i32;
+ let has_partial_merge = proto_mode == ProtoAggregateMode::PartialMerge
+ || agg.expr_modes.contains(&partial_merge_value);
+
let agg_exprs: PhyAggResult = agg
.agg_exprs
.iter()
.map(|expr| self.create_agg_expr(expr, Arc::clone(&schema)))
.collect();
- let aggr_expr = agg_exprs?.into_iter().map(Arc::new).collect();
+ let aggr_expr: Vec<Arc<AggregateFunctionExpr>> = if has_partial_merge {
+ // Wrap PartialMerge expressions with MergeAsPartial.
+ // State fields in the child's output start at initial_input_buffer_offset.
+ let mut state_offset = agg.initial_input_buffer_offset as usize;
+ let per_expr_modes: Vec<i32> = if !agg.expr_modes.is_empty() {
+ agg.expr_modes.clone()
+ } else {
+ vec![agg.mode; agg.agg_exprs.len()]
+ };
+
+ agg_exprs?
+ .into_iter()
+ .enumerate()
+ .map(|(idx, expr)| {
+ if per_expr_modes[idx] == partial_merge_value {
+ // PartialMerge: wrap with MergeAsPartial
+ let state_fields = expr
+ .state_fields()
+ .map_err(|e| ExecutionError::GeneralError(e.to_string()))?;
+ let num_state_fields = state_fields.len();
+
+ let state_cols: Vec<Arc<dyn PhysicalExpr>> = (0..num_state_fields)
+ .map(|i| {
+ let col_idx = state_offset + i;
+ let field = schema.field(col_idx);
+ Arc::new(Column::new(field.name(), col_idx))
+ as Arc<dyn PhysicalExpr>
+ })
+ .collect();
+ state_offset += num_state_fields;
+
+ let merge_udf =
+ crate::execution::merge_as_partial::MergeAsPartialUDF::new(
+ &expr,
+ )
+ .map_err(|e| ExecutionError::DataFusionError(e.to_string()))?;
+ let merge_udf_arc = Arc::new(
+ datafusion::logical_expr::AggregateUDF::new_from_impl(
+ merge_udf,
+ ),
+ );
+
+ let merge_expr =
+ AggregateExprBuilder::new(merge_udf_arc, state_cols)
+ .schema(Arc::clone(&schema))
+ .alias(format!("col_{idx}"))
+ .with_ignore_nulls(expr.ignore_nulls())
+ .with_distinct(expr.is_distinct())
+ .build()
+ .map_err(|e| {
+ ExecutionError::DataFusionError(e.to_string())
+ })?;
+
+ Ok(Arc::new(merge_expr))
+ } else {
+ Ok(Arc::new(expr))
+ }
+ })
+ .collect::<Result<Vec<_>, ExecutionError>>()?
+ } else {
+ agg_exprs?.into_iter().map(Arc::new).collect()
+ };
// Build per-aggregate filter expressions from the FILTER (WHERE ...) clause.
// Filters are only present in Partial mode; Final/PartialMerge always get None.
diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index 1efa1dcc1..dd85d47e6 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -270,6 +270,13 @@ message HashAggregate {
repeated spark.spark_expression.AggExpr agg_exprs = 2;
repeated spark.spark_expression.Expr result_exprs = 3;
AggregateMode mode = 5;
+ // Per-expression modes for mixed-mode aggregates (e.g., PartialMerge + Partial).
+ // When set, each entry corresponds to agg_exprs at the same index.
+ // When empty, all expressions use the `mode` field.
+ repeated AggregateMode expr_modes = 6;
+ // Offset in the child's output where aggregate buffer attributes start.
+ // Used by PartialMerge to locate state fields in the input.
+ int32 initial_input_buffer_offset = 7;
}
message Limit {
@@ -319,6 +326,7 @@ message ParquetWriter {
enum AggregateMode {
Partial = 0;
Final = 1;
+ PartialMerge = 2;
}
message Expand {
diff --git a/native/spark-expr/src/agg_funcs/avg_decimal.rs b/native/spark-expr/src/agg_funcs/avg_decimal.rs
index 9e8a31afa..05d74e1f9 100644
--- a/native/spark-expr/src/agg_funcs/avg_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs
@@ -367,13 +367,13 @@ impl Accumulator for AvgDecimalAccumulator {
// Check for overflow during sum accumulation in ANSI mode.
// This matches Spark's DecimalDivideWithOverflowCheck behavior.
if self.sum.is_none() && !self.is_empty && self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("avg");
return Err(self.wrap_error_with_context(error));
}
// Also check if is_not_null is false (indicates overflow)
if !self.is_not_null && self.count > 0 && self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("avg");
return Err(self.wrap_error_with_context(error));
}
@@ -573,7 +573,7 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator {
if is_overflow || !is_valid_decimal_precision(new_sum, self.sum_precision) {
if self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("avg");
return Err(self.wrap_error_with_context(error));
}
self.is_not_null.set_bit(group_index, false);
@@ -609,7 +609,7 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator {
// Check for overflow during sum accumulation in ANSI mode.
// This matches Spark's DecimalDivideWithOverflowCheck behavior.
if !is_not_null && count > 0 && self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("avg");
return Err(self.wrap_error_with_context(error));
}
diff --git a/native/spark-expr/src/agg_funcs/sum_decimal.rs b/native/spark-expr/src/agg_funcs/sum_decimal.rs
index 46db7f36b..180e884bf 100644
--- a/native/spark-expr/src/agg_funcs/sum_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs
@@ -214,7 +214,7 @@ impl SumDecimalAccumulator {
if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
}
self.sum = None;
@@ -311,57 +311,61 @@ impl Accumulator for SumDecimalAccumulator {
}
fn merge_batch(&mut self, states: &[ArrayRef]) -> DFResult<()> {
- // For decimal sum, always expect 2 state arrays regardless of eval_mode
assert_eq!(
states.len(),
2,
- "Expect two elements in 'states' but found {}",
+ "expected (sum, is_empty), got {}",
states.len()
);
- assert_eq!(states[0].len(), 1);
- assert_eq!(states[1].len(), 1);
+ assert_eq!(states[0].len(), states[1].len());
- let that_sum_array = states[0].as_primitive::<Decimal128Type>();
- let that_sum = if that_sum_array.is_null(0) {
- None
- } else {
- Some(that_sum_array.value(0))
- };
+ let sum_array = states[0].as_primitive::<Decimal128Type>();
+ let is_empty_array = states[1].as_boolean();
- let that_is_empty = states[1].as_boolean().value(0);
- let that_overflowed = !that_is_empty && that_sum.is_none();
- let this_overflowed = !self.is_empty && self.sum.is_none();
+ // May be called with multiple state rows (e.g. under MergeAsPartial or
+ // non-grouped aggregates merging state from several partitions).
+ for i in 0..sum_array.len() {
+ let that_sum = if sum_array.is_null(i) {
+ None
+ } else {
+ Some(sum_array.value(i))
+ };
+ let that_is_empty = is_empty_array.value(i);
+ let that_overflowed = !that_is_empty && that_sum.is_none();
+ let this_overflowed = !self.is_empty && self.sum.is_none();
- if that_overflowed || this_overflowed {
- self.sum = None;
- self.is_empty = false;
- return Ok(());
- }
+ // Overflow is sticky; ANSI error was raised upstream.
+ if that_overflowed || this_overflowed {
+ self.sum = None;
+ self.is_empty = false;
+ continue;
+ }
- if that_is_empty {
- return Ok(());
- }
+ if that_is_empty {
+ continue;
+ }
- if self.is_empty {
- self.sum = that_sum;
- self.is_empty = false;
- return Ok(());
- }
+ if self.is_empty {
+ self.sum = that_sum;
+ self.is_empty = false;
+ continue;
+ }
- let left = self.sum.unwrap();
- let right = that_sum.unwrap();
- let (new_sum, is_overflow) = left.overflowing_add(right);
+ let left = self.sum.unwrap();
+ let right = that_sum.unwrap();
+ let (new_sum, is_overflow) = left.overflowing_add(right);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
- if self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
- return Err(self.wrap_error_with_context(error));
+ if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
+ if self.eval_mode == EvalMode::Ansi {
+ let error = decimal_sum_overflow_error("sum");
+ return Err(self.wrap_error_with_context(error));
+ } else {
+ self.sum = None;
+ self.is_empty = false;
+ }
} else {
- self.sum = None;
- self.is_empty = false;
+ self.sum = Some(new_sum);
}
- } else {
- self.sum = Some(new_sum);
}
Ok(())
@@ -426,7 +430,7 @@ impl SumDecimalGroupsAccumulator {
if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
}
self.sum[group_index] = None;
@@ -595,7 +599,7 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator {
if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
- let error = decimal_sum_overflow_error();
+ let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
} else {
self.sum[group_index] = None;
@@ -771,4 +775,32 @@ mod tests {
let result = result.as_any().downcast_ref::<Decimal128Array>().unwrap();
assert_eq!(result.value(0), 40); // 10 + 30 = 40
}
+
+ /// `merge_batch` must consume every row — previously it asserted
+ /// `states[*].len() == 1` and silently dropped the rest.
+ #[test]
+ fn test_accumulator_merge_batch_multi_row() {
+ let data_type = DataType::Decimal128(10, 2);
+ let mut acc = SumDecimalAccumulator::new(
+ 10,
+ 2,
+ EvalMode::Legacy,
+ None,
+ crate::create_query_context_map(),
+ );
+
+ // sums [100, 200, null, 300], is_empty [false, false, true, false] -> 600
+ let sums: ArrayRef = Arc::new(
+ Decimal128Array::from(vec![Some(100i128), Some(200), None, Some(300)])
+ .with_data_type(data_type.clone()),
+ );
+ let is_empty: ArrayRef = Arc::new(BooleanArray::from(vec![false, false, true, false]));
+ acc.merge_batch(&[sums, is_empty]).unwrap();
+
+ let evaluated = acc.evaluate().unwrap();
+ match evaluated {
+ ScalarValue::Decimal128(Some(v), _, _) => assert_eq!(v, 600),
+ other => panic!("unexpected result: {other:?}"),
+ }
+ }
}
diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs
index 781528521..5ce606677 100644
--- a/native/spark-expr/src/agg_funcs/sum_int.rs
+++ b/native/spark-expr/src/agg_funcs/sum_int.rs
@@ -173,30 +173,8 @@ impl Accumulator for SumIntegerAccumulatorLegacy {
}
fn merge_batch(&mut self, states: &[ArrayRef]) -> DFResult<()> {
- if states.len() != 1 {
- return Err(DataFusionError::Internal(format!(
- "Invalid state while merging batch. Expected 1 element but found {}",
- states.len()
- )));
- }
-
- let that_sum_array = states[0].as_primitive::<Int64Type>();
- let that_sum = if that_sum_array.is_null(0) {
- None
- } else {
- Some(that_sum_array.value(0))
- };
-
- if that_sum.is_none() {
- return Ok(());
- }
- if self.sum.is_none() {
- self.sum = that_sum;
- return Ok(());
- }
-
- self.sum = Some(self.sum.unwrap().add_wrapping(that_sum.unwrap()));
- Ok(())
+ // Merging partial sums is the same as summing values
+ self.update_batch(states)
}
}
@@ -268,35 +246,8 @@ impl Accumulator for SumIntegerAccumulatorAnsi {
}
fn merge_batch(&mut self, states: &[ArrayRef]) -> DFResult<()> {
- if states.len() != 1 {
- return Err(DataFusionError::Internal(format!(
- "Invalid state while merging batch. Expected 1 element but found {}",
- states.len()
- )));
- }
-
- let that_sum_array = states[0].as_primitive::<Int64Type>();
- let that_sum = if that_sum_array.is_null(0) {
- None
- } else {
- Some(that_sum_array.value(0))
- };
-
- if that_sum.is_none() {
- return Ok(());
- }
- if self.sum.is_none() {
- self.sum = that_sum;
- return Ok(());
- }
-
- self.sum = Some(
- self.sum
- .unwrap()
- .add_checked(that_sum.unwrap())
- .map_err(|_| DataFusionError::from(arithmetic_overflow_error("integer")))?,
- );
- Ok(())
+ // Merging partial sums is the same as summing values
+ self.update_batch(states)
}
}
@@ -400,35 +351,43 @@ impl Accumulator for SumIntegerAccumulatorTry {
}
let that_sum_array = states[0].as_primitive::<Int64Type>();
- let that_sum = if that_sum_array.is_null(0) {
- None
- } else {
- Some(that_sum_array.value(0))
- };
- let that_has_all_nulls = states[1].as_boolean().value(0);
+ let that_has_all_nulls_array = states[1].as_boolean();
- let that_overflowed = !that_has_all_nulls && that_sum.is_none();
- if that_overflowed || self.overflowed() {
- self.sum = None;
- self.has_all_nulls = false;
- return Ok(());
- }
+ for row in 0..that_sum_array.len() {
+ if self.overflowed() {
+ return Ok(());
+ }
- if that_has_all_nulls {
- return Ok(());
- }
- if self.has_all_nulls {
- self.sum = that_sum;
- self.has_all_nulls = false;
- return Ok(());
- }
+ let that_sum = if that_sum_array.is_null(row) {
+ None
+ } else {
+ Some(that_sum_array.value(row))
+ };
+ let that_has_all_nulls = that_has_all_nulls_array.value(row);
- // Both sides have non-null values
- match self.sum.unwrap().add_checked(that_sum.unwrap()) {
- Ok(v) => self.sum = Some(v),
- Err(_) => {
+ let that_overflowed = !that_has_all_nulls && that_sum.is_none();
+ if that_overflowed {
self.sum = None;
self.has_all_nulls = false;
+ return Ok(());
+ }
+
+ if that_has_all_nulls {
+ continue;
+ }
+
+ if self.has_all_nulls {
+ self.sum = that_sum;
+ self.has_all_nulls = false;
+ continue;
+ }
+
+ match self.sum.unwrap().add_checked(that_sum.unwrap()) {
+ Ok(v) => self.sum = Some(v),
+ Err(_) => {
+ self.sum = None;
+ self.has_all_nulls = false;
+ }
}
}
Ok(())
@@ -1031,4 +990,29 @@ mod tests {
.collect();
assert_eq!(result, vec![Some(6)]);
}
+
+ /// Regression coverage for the scalar `Accumulator` path used when Comet wraps a PartialMerge
+ /// expression with `MergeAsPartial`: `merge_batch` has to consume every row of the incoming
+ /// state array. The previous implementation read only row 0, which silently under-counted
+ /// whenever the MergeAsPartial operator handed us a state batch with more than one row.
+ #[test]
+ fn test_legacy_accumulator_merge_batch_multi_row() {
+ let mut acc = SumIntegerAccumulatorLegacy::new();
+ let states: ArrayRef = Arc::new(Int64Array::from(vec![Some(1i64), Some(2), None, Some(3)]));
+ acc.merge_batch(&[states]).unwrap();
+ assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(6)));
+ }
+
+ #[test]
+ fn test_ansi_accumulator_merge_batch_multi_row() {
+ let mut acc = SumIntegerAccumulatorAnsi::new();
+ let states: ArrayRef = Arc::new(Int64Array::from(vec![
+ Some(10i64),
+ Some(20),
+ None,
+ Some(30),
+ ]));
+ acc.merge_batch(&[states]).unwrap();
+ assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(60)));
+ }
}
diff --git a/native/spark-expr/src/lib.rs b/native/spark-expr/src/lib.rs
index 5c476f65e..504d7814e 100644
--- a/native/spark-expr/src/lib.rs
+++ b/native/spark-expr/src/lib.rs
@@ -123,8 +123,10 @@ pub(crate) fn arithmetic_overflow_error(from_type: &str) -> SparkError {
}
}
-pub(crate) fn decimal_sum_overflow_error() -> SparkError {
- SparkError::DecimalSumOverflow
+pub(crate) fn decimal_sum_overflow_error(function_name: &str) -> SparkError {
+ SparkError::DecimalSumOverflow {
+ function_name: function_name.to_string(),
+ }
}
pub(crate) fn divide_by_zero_error() -> SparkError {
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
index 213c3d0ad..06d926dfc 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -53,7 +53,7 @@ import org.apache.comet.CometSparkSessionExtensions._
import org.apache.comet.rules.CometExecRule.allExecs
import org.apache.comet.serde._
import org.apache.comet.serde.operator._
-import org.apache.comet.shims.ShimSubqueryBroadcast
+import org.apache.comet.shims.{ShimCometStreaming, ShimSubqueryBroadcast}
object CometExecRule {
@@ -536,6 +536,10 @@ case class CometExecRule(session: SparkSession)
// We shouldn't transform Spark query plan if Comet is not loaded.
if (!isCometLoaded(conf)) return plan
+ // Comet does not support structured streaming. Fall back to Spark for any plan that
+ // belongs to a streaming query (detected via StreamSourceAwareSparkPlan.getStream).
+ if (ShimCometStreaming.isStreamingPlan(plan)) return plan
+
if (!CometConf.COMET_EXEC_ENABLED.get(conf)) {
// Comet exec is disabled, but for Spark shuffle, we still can use Comet columnar shuffle
if (isCometShuffleEnabled(conf)) {
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 2cd5ba156..1c9ec98a7 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -50,7 +50,7 @@ import org.apache.comet.objectstore.NativeConfig
import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported}
import org.apache.comet.parquet.Native
import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan}
-import org.apache.comet.shims.{CometTypeShim, ShimFileFormat, ShimSubqueryBroadcast}
+import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast}
/**
* Spark physical optimizer rule for replacing Spark scans with Comet scans.
@@ -76,6 +76,11 @@ case class CometScanRule(session: SparkSession)
private def _apply(plan: SparkPlan): SparkPlan = {
if (!isCometLoaded(conf)) return plan
+ // Comet does not support structured streaming. The parallel guard in
+ // CometExecRule only stops operator wrapping, so without this check we
+ // would still rewrite scans to CometScanExec in a streaming plan.
+ if (ShimCometStreaming.isStreamingPlan(plan)) return plan
+
def isSupportedScanNode(plan: SparkPlan): Boolean = plan match {
case _: FileSourceScanExec => true
case _: BatchScanExec => true
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index af3e27c77..1cfb58b88 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -468,15 +468,15 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
binding: Boolean,
conf: SQLConf): Option[AggExpr] = {
- // Support Count(distinct single_value)
- // COUNT(DISTINCT x) - supported
- // COUNT(DISTINCT x, x) - supported through transition to COUNT(DISTINCT x)
- // COUNT(DISTINCT x, y) - not supported
+ // Distinct aggregates with a single column are supported (e.g., COUNT(DISTINCT x),
+ // SUM(DISTINCT x), AVG(DISTINCT x)). The multi-stage plan generated by Spark
+ // guarantees distinct semantics through grouping — the native side does not need
+ // to handle deduplication.
+ // Multi-column distinct is only supported for COUNT (e.g., COUNT(DISTINCT x, y)).
if (aggExpr.isDistinct
- &&
- !(aggExpr.aggregateFunction.prettyName == "count" &&
- aggExpr.aggregateFunction.children.length == 1)) {
- withInfo(aggExpr, s"Distinct aggregate not supported for: $aggExpr")
+ && aggExpr.aggregateFunction.children.length > 1
+ && aggExpr.aggregateFunction.prettyName != "count") {
+ withInfo(aggExpr, s"Multi-column distinct aggregate not supported for: $aggExpr")
return None
}
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index 30b454591..888df13ba 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -30,7 +30,7 @@ import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeSet, Expression, ExpressionSet, Generator, NamedExpression, SortOrder}
-import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateMode, CollectSet, Final, Partial, PartialMerge}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateMode, CollectSet, Final, First, Last, Partial, PartialMerge}
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.physical._
@@ -1356,8 +1356,11 @@ trait CometBaseAggregate {
childOp: OperatorOuterClass.Operator*): Option[OperatorOuterClass.Operator] = {
val modes = aggregate.aggregateExpressions.map(_.mode).distinct
- // In distinct aggregates there can be a combination of modes
- val multiMode = modes.size > 1
+ val modeSet = modes.toSet
+ val hasPartialMerge = modeSet.contains(PartialMerge)
+ // In distinct aggregates there can be a combination of modes.
+ // We support {Partial, PartialMerge} mix; other combinations are rejected.
+ val multiMode = modes.size > 1 && modeSet != Set(Partial, PartialMerge)
// For a final mode HashAggregate, we only need to transform the HashAggregate
// if there is Comet partial aggregation.
val sparkFinalMode = modes.contains(Final) && findCometPartialAgg(aggregate.child).isEmpty
@@ -1437,33 +1440,59 @@ trait CometBaseAggregate {
hashAggBuilder.addAllResultExprs(resultExprs.map(_.get).asJava)
Some(builder.setHashAgg(hashAggBuilder).build())
} else {
- val modes = aggregateExpressions.map(_.mode).distinct
-
- if (modes.size != 1) {
- // This shouldn't happen as all aggregation expressions should share the same mode.
- // Fallback to Spark nevertheless here.
- withInfo(aggregate, "All aggregate expressions do not have the same mode")
+ // Validate mode combinations. We support:
+ // - All Partial
+ // - All Final
+ // - All PartialMerge
+ // - Mixed {Partial, PartialMerge} (for distinct aggregate plans)
+ val isMixedPartialMerge = modeSet == Set(Partial, PartialMerge)
+ if (modes.size > 1 && !isMixedPartialMerge) {
+ withInfo(aggregate, s"Unsupported mixed aggregation modes: ${modes.mkString(", ")}")
return None
}
- val mode = modes.head match {
- case Partial => CometAggregateMode.Partial
- case Final => CometAggregateMode.Final
- case _ =>
- withInfo(aggregate, s"Unsupported aggregation mode ${modes.head}")
+ // Determine the proto mode. For uniform modes, use that mode directly.
+ // For mixed {Partial, PartialMerge}, use Partial as the base mode since
+ // PartialMerge expressions are wrapped with MergeAsPartial on the native side.
+ val mode = if (isMixedPartialMerge) {
+ CometAggregateMode.Partial
+ } else {
+ modes.head match {
+ case Partial => CometAggregateMode.Partial
+ case Final => CometAggregateMode.Final
+ case PartialMerge => CometAggregateMode.PartialMerge
+ case _ =>
+ withInfo(aggregate, s"Unsupported aggregation mode ${modes.head}")
+ return None
+ }
+ }
+
+ // FIRST/LAST are order-dependent: in PartialMerge mode, DataFusion's hash
+ // table may process rows in a different order than Spark's. CollectSet is
+ // handled separately (floating-point compat in CometCollectSet; streaming
+ // in ShimCometStreaming.isStreamingPlan).
+ // https://github.com/apache/datafusion-comet/issues/4131
+ if (hasPartialMerge) {
+ val unsupportedAggs = aggregateExpressions.filter { a =>
+ a.mode == PartialMerge && (a.aggregateFunction.isInstanceOf[First] ||
+ a.aggregateFunction.isInstanceOf[Last])
+ }
+ if (unsupportedAggs.nonEmpty) {
+ withInfo(
+ aggregate,
+ "PartialMerge not supported for aggregates: " +
+ unsupportedAggs.map(_.aggregateFunction.prettyName).mkString(", "))
return None
+ }
}
- // In final mode, the aggregate expressions are bound to the output of the
- // child and partial aggregate expressions buffer attributes produced by partial
- // aggregation. This is done in Spark `HashAggregateExec` internally. In Comet,
- // we don't have to do this because we don't use the merging expression.
- val binding = mode != CometAggregateMode.Final
- // `output` is only used when `binding` is true (i.e., non-Final)
+ // Per-expression binding: Partial expressions bind to child output,
+ // PartialMerge/Final expressions do not (native planner handles their input).
val output = child.output
-
- val aggExprs =
- aggregateExpressions.map(aggExprToProto(_, output, binding, aggregate.conf))
+ val aggExprs = aggregateExpressions.map { a =>
+ val exprBinding = a.mode != PartialMerge && a.mode != Final
+ aggExprToProto(a, output, exprBinding, aggregate.conf)
+ }
if (aggExprs.exists(_.isEmpty)) {
withInfo(
@@ -1491,6 +1520,23 @@ trait CometBaseAggregate {
hashAggBuilder.addAllResultExprs(resultExprs.map(_.get).asJava)
}
hashAggBuilder.setModeValue(mode.getNumber)
+
+ // Send per-expression modes and buffer offset for PartialMerge handling
+ if (hasPartialMerge) {
+ val exprModes = aggregateExpressions.map { a =>
+ a.mode match {
+ case Partial => CometAggregateMode.Partial
+ case PartialMerge => CometAggregateMode.PartialMerge
+ case Final => CometAggregateMode.Final
+ case other =>
+ withInfo(aggregate, s"Unsupported aggregation mode $other")
+ return None
+ }
+ }
+ hashAggBuilder.addAllExprModes(exprModes.asJava)
+ hashAggBuilder.setInitialInputBufferOffset(aggregate.initialInputBufferOffset)
+ }
+
Some(builder.setHashAgg(hashAggBuilder).build())
} else {
val allChildren: Seq[Expression] =
@@ -1504,14 +1550,21 @@ trait CometBaseAggregate {
/**
* Find the first Comet partial aggregate in the plan. If it reaches a Spark HashAggregate with
- * partial mode, it will return None.
+ * partial or partial-merge mode, it will return None.
*/
private def findCometPartialAgg(plan: SparkPlan): Option[CometHashAggregateExec] = {
+ def isPartialOrMerge(mode: AggregateMode): Boolean =
+ mode == Partial || mode == PartialMerge
+
plan.collectFirst {
- case agg: CometHashAggregateExec if agg.aggregateExpressions.forall(_.mode == Partial) =>
+ case agg: CometHashAggregateExec
+ if agg.aggregateExpressions.forall(e => isPartialOrMerge(e.mode)) =>
Some(agg)
- case agg: HashAggregateExec if agg.aggregateExpressions.forall(_.mode == Partial) => None
- case agg: ObjectHashAggregateExec if agg.aggregateExpressions.forall(_.mode == Partial) =>
+ case agg: HashAggregateExec
+ if agg.aggregateExpressions.forall(e => isPartialOrMerge(e.mode)) =>
+ None
+ case agg: ObjectHashAggregateExec
+ if agg.aggregateExpressions.forall(e => isPartialOrMerge(e.mode)) =>
None
case a: AQEShuffleReadExec => findCometPartialAgg(a.child)
case s: ShuffleQueryStageExec => findCometPartialAgg(s.plan)
@@ -1608,7 +1661,7 @@ object CometObjectHashAggregateExec
* case branch here mapping it to the native state type.
*/
private def adjustOutputForNativeState(op: ObjectHashAggregateExec): Seq[Attribute] = {
- // CometBaseAggregate.doConvert guarantees all expressions share the same mode.
+ // This adjustment only applies to pure-Partial aggregates (checked below).
val modes = op.aggregateExpressions.map(_.mode).distinct
if (modes != Seq(Partial)) {
return op.output
@@ -1650,11 +1703,8 @@ case class CometHashAggregateExec(
// The aggExprs could be empty. For example, if the aggregate functions only have
// distinct aggregate functions or only have group by, the aggExprs is empty and
- // modes is empty too. If aggExprs is not empty, we need to verify all the
- // aggregates have the same mode.
+ // modes is empty too.
val modes: Seq[AggregateMode] = aggregateExpressions.map(_.mode).distinct
- assert(modes.length == 1 || modes.isEmpty)
- val mode = modes.headOption
override def producedAttributes: AttributeSet = outputSet ++ AttributeSet(resultExpressions)
@@ -1671,7 +1721,7 @@ case class CometHashAggregateExec(
}
override def stringArgs: Iterator[Any] =
- Iterator(input, mode, groupingExpressions, aggregateExpressions, child)
+ Iterator(input, modes, groupingExpressions, aggregateExpressions, child)
override def equals(obj: Any): Boolean = {
obj match {
@@ -1680,7 +1730,7 @@ case class CometHashAggregateExec(
this.groupingExpressions == other.groupingExpressions &&
this.aggregateExpressions == other.aggregateExpressions &&
this.input == other.input &&
- this.mode == other.mode &&
+ this.modes == other.modes &&
this.child == other.child &&
this.serializedPlanOpt == other.serializedPlanOpt
case _ =>
@@ -1689,7 +1739,7 @@ case class CometHashAggregateExec(
}
override def hashCode(): Int =
- Objects.hashCode(output, groupingExpressions, aggregateExpressions, input, mode, child)
+ Objects.hashCode(output, groupingExpressions, aggregateExpressions, input, modes, child)
override protected def outputExpressions: Seq[NamedExpression] = resultExpressions
}
diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/ShimCometStreaming.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimCometStreaming.scala
new file mode 100644
index 000000000..f80eab95e
--- /dev/null
+++ b/spark/src/main/spark-3.x/org/apache/comet/shims/ShimCometStreaming.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.shims
+
+import org.apache.spark.sql.execution.SparkPlan
+
+// StreamSourceAwareSparkPlan does not exist in Spark 3.x, so fall back to
+// walking the physical tree and checking each node's logical link. Inspecting
+// only the root's logical link would silently miss streaming plans whenever a
+// rule produced a fresh root node without copying the link over.
+object ShimCometStreaming {
+ def isStreamingPlan(plan: SparkPlan): Boolean =
+ plan.exists(_.logicalLink.exists(_.isStreaming))
+}
diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/ShimCometStreaming.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimCometStreaming.scala
new file mode 100644
index 000000000..ed4cb37e2
--- /dev/null
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/ShimCometStreaming.scala
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.shims
+
+import org.apache.spark.sql.execution.{SparkPlan, StreamSourceAwareSparkPlan}
+
+object ShimCometStreaming {
+ def isStreamingPlan(plan: SparkPlan): Boolean = plan.exists {
+ case p: StreamSourceAwareSparkPlan => p.getStream.isDefined
+ case _ => false
+ }
+}
diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
index 6a1e04da4..bdb10fa2f 100644
--- a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
+++ b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala
@@ -116,7 +116,10 @@ trait ShimSparkErrorConverter {
Some(QueryExecutionErrors.overflowInIntegralDivideError(context.headOption.orNull))
case "DecimalSumOverflow" =>
- Some(QueryExecutionErrors.overflowInSumOfDecimalError(context.headOption.orNull, ""))
+ val functionName = params.get("functionName").map(_.toString).getOrElse("sum")
+ Some(
+ QueryExecutionErrors
+ .overflowInSumOfDecimalError(context.headOption.orNull, s"try_$functionName"))
case "NumericValueOutOfRange" =>
val decimal = Decimal(params("value").toString)
diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/partial_merge.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/partial_merge.sql
new file mode 100644
index 000000000..1ff7e0ef6
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/partial_merge.sql
@@ -0,0 +1,380 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Tests exercising the Spark PartialMerge aggregate mode, produced when
+-- RewriteDistinctAggregates rewrites a mix of distinct and non-distinct
+-- aggregates (or multiple distinct aggregates) into an Expand-based multi-stage
+-- plan. In that plan the non-distinct aggregates flow through a middle
+-- PartialMerge stage that must merge intermediate state buffers and re-emit
+-- state (rather than evaluating a final value).
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+-- Config: spark.comet.expression.Cast.allowIncompatible=true
+
+-- ============================================================
+-- Setup: shared tables
+-- ============================================================
+
+statement
+CREATE TABLE pm_basic(i int, l bigint, d double, s string, grp string) USING parquet
+
+statement
+INSERT INTO pm_basic VALUES
+ (1, 10, 1.5, 'a', 'g1'),
+ (2, 20, 2.5, 'b', 'g1'),
+ (1, 10, 1.5, 'a', 'g1'),
+ (3, 30, 3.5, 'c', 'g2'),
+ (3, 30, 3.5, 'c', 'g2'),
+ (4, 40, 4.5, 'd', 'g2'),
+ (5, 50, 5.5, 'e', 'g3'),
+ (5, 50, 5.5, 'e', 'g3'),
+ (5, 50, 5.5, 'e', 'g3')
+
+statement
+CREATE TABLE pm_nulls(i int, l bigint, d double, s string, grp string) USING parquet
+
+statement
+INSERT INTO pm_nulls VALUES
+ (NULL, NULL, NULL, NULL, 'g1'),
+ (1, 10, 1.5, 'a', 'g1'),
+ (1, 10, 1.5, 'a', 'g1'),
+ (2, 20, 2.5, 'b', 'g1'),
+ (NULL, NULL, NULL, NULL, 'g2'),
+ (NULL, NULL, NULL, NULL, 'g2'),
+ (3, 30, 3.5, 'c', 'g2')
+
+statement
+CREATE TABLE pm_empty(i int, grp string) USING parquet
+
+statement
+CREATE TABLE pm_single(i int, grp string) USING parquet
+
+statement
+INSERT INTO pm_single VALUES (42, 'g1')
+
+statement
+CREATE TABLE pm_single_null(i int, grp string) USING parquet
+
+statement
+INSERT INTO pm_single_null VALUES (NULL, 'g1')
+
+statement
+CREATE TABLE pm_all_nulls(i int, grp string) USING parquet
+
+statement
+INSERT INTO pm_all_nulls VALUES (NULL, 'g1'), (NULL, 'g1'), (NULL, 'g2'), (NULL, 'g2')
+
+statement
+CREATE TABLE pm_decimal(v DECIMAL(10,2), w DECIMAL(20,4), grp string) USING parquet
+
+statement
+INSERT INTO pm_decimal VALUES
+ (1.23, 10.0001, 'g1'),
+ (1.23, 10.0001, 'g1'),
+ (4.56, 20.0002, 'g1'),
+ (NULL, NULL, 'g2'),
+ (7.89, 30.0003, 'g2'),
+ (7.89, 30.0003, 'g2')
+
+statement
+CREATE TABLE pm_large(k int, v int, grp string) USING parquet
+
+statement
+INSERT INTO pm_large
+SELECT
+ CAST(id % 50 AS int) AS k,
+ CAST(id AS int) AS v,
+ CASE WHEN id < 500 THEN 'g1' ELSE 'g2' END AS grp
+FROM (SELECT explode(sequence(1, 1000)) AS id)
+
+statement
+CREATE TABLE pm_boundary(i int, l bigint) USING parquet
+
+statement
+INSERT INTO pm_boundary VALUES
+ (2147483647, 9223372036854775807),
+ (-2147483648, -9223372036854775808),
+ (0, 0),
+ (1, 1),
+ (1, 1)
+
+-- ############################################################
+-- BASIC partial merge patterns
+-- ############################################################
+
+-- ============================================================
+-- basic: count(distinct) + non-distinct aggregates
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i) FROM pm_basic
+
+query
+SELECT count(DISTINCT i), count(i) FROM pm_basic
+
+query
+SELECT count(DISTINCT i), min(i), max(i) FROM pm_basic
+
+query tolerance=1e-6
+SELECT count(DISTINCT d), sum(d), avg(d) FROM pm_basic
+
+-- ============================================================
+-- basic: sum(distinct) + non-distinct aggregates
+-- ============================================================
+
+query
+SELECT sum(DISTINCT i), sum(i) FROM pm_basic
+
+query
+SELECT sum(DISTINCT l), count(l) FROM pm_basic
+
+-- ============================================================
+-- basic: count(distinct) with GROUP BY
+-- ============================================================
+
+query
+SELECT grp, count(DISTINCT i), sum(i) FROM pm_basic GROUP BY grp ORDER BY grp
+
+query
+SELECT grp, count(DISTINCT i) FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- basic: sum(distinct) with GROUP BY
+-- ============================================================
+
+query
+SELECT grp, sum(DISTINCT i), sum(i), count(i) FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ############################################################
+-- ADVANCED partial merge patterns
+-- ############################################################
+
+-- ============================================================
+-- advanced: multiple distinct aggregates on different columns
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), count(DISTINCT l), count(DISTINCT s) FROM pm_basic
+
+query
+SELECT count(DISTINCT i), count(DISTINCT l), sum(i) FROM pm_basic
+
+query tolerance=1e-6
+SELECT count(DISTINCT i), sum(DISTINCT l), avg(d) FROM pm_basic
+
+-- ============================================================
+-- advanced: non-count distinct + non-distinct (Expand pattern)
+-- Exercises the middle PartialMerge stage for a float-producing aggregate
+-- ============================================================
+
+query tolerance=1e-6
+SELECT avg(DISTINCT i), sum(i) FROM pm_basic
+
+query tolerance=1e-6
+SELECT avg(DISTINCT d), sum(d), count(d) FROM pm_basic
+
+query
+SELECT sum(DISTINCT i), max(i), min(i), count(i) FROM pm_basic
+
+-- ============================================================
+-- advanced: multi-column distinct
+-- ============================================================
+
+query
+SELECT count(DISTINCT i, l), sum(i) FROM pm_basic
+
+query
+SELECT grp, count(DISTINCT i, l), count(*) FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- advanced: multiple distinct + multiple non-distinct + GROUP BY
+-- ============================================================
+
+query tolerance=1e-6
+SELECT grp,
+ count(DISTINCT i),
+ sum(DISTINCT l),
+ sum(i),
+ avg(d),
+ count(*)
+FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- advanced: decimal partial merge (sum/avg produce decimal state)
+-- ============================================================
+
+query
+SELECT count(DISTINCT v), sum(v) FROM pm_decimal
+
+query
+SELECT grp, count(DISTINCT v), sum(v), sum(w) FROM pm_decimal GROUP BY grp ORDER BY grp
+
+query tolerance=1e-6
+SELECT grp, count(DISTINCT v), avg(v) FROM pm_decimal GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- advanced: HAVING against a partial-merged aggregate
+-- ============================================================
+
+query
+SELECT grp, count(DISTINCT i) AS c, sum(i) AS s
+FROM pm_basic
+GROUP BY grp
+HAVING sum(i) > 5
+ORDER BY grp
+
+-- ============================================================
+-- advanced: subquery feeds distinct + non-distinct aggregates
+-- ============================================================
+
+query
+SELECT count(DISTINCT x), sum(x)
+FROM (SELECT i + 1 AS x FROM pm_basic) t
+
+-- ============================================================
+-- advanced: large, multi-batch input
+-- ============================================================
+
+query
+SELECT count(DISTINCT k), sum(v) FROM pm_large
+
+query
+SELECT grp, count(DISTINCT k), sum(v), count(*) FROM pm_large GROUP BY grp ORDER BY grp
+
+-- ############################################################
+-- EDGE cases
+-- ############################################################
+
+-- ============================================================
+-- edge: NULLs interact with distinct + non-distinct
+-- count(DISTINCT) ignores NULLs; sum/count do too
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i), count(i), count(*) FROM pm_nulls
+
+query
+SELECT grp, count(DISTINCT i), sum(i), count(i), count(*)
+FROM pm_nulls GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- edge: empty table
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i), count(i) FROM pm_empty
+
+query
+SELECT grp, count(DISTINCT i), sum(i) FROM pm_empty GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- edge: single non-null row
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i), count(i) FROM pm_single
+
+-- ============================================================
+-- edge: single NULL row
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i), count(i) FROM pm_single_null
+
+-- ============================================================
+-- edge: all NULLs (per group and overall)
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), sum(i), count(i) FROM pm_all_nulls
+
+query
+SELECT grp, count(DISTINCT i), sum(i), count(i)
+FROM pm_all_nulls GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- edge: int/long boundary values with duplicates
+-- Partial merge must merge state without overflow when duplicates collapse
+-- ============================================================
+
+query
+SELECT count(DISTINCT i), count(DISTINCT l) FROM pm_boundary
+
+query
+SELECT count(DISTINCT i), sum(DISTINCT i) FROM pm_boundary
+
+-- ============================================================
+-- edge: DISTINCT inside expression + non-distinct alongside
+-- ============================================================
+
+query
+SELECT count(DISTINCT i + 1), sum(i) FROM pm_basic
+
+query
+SELECT grp, count(DISTINCT IF(i % 2 = 0, NULL, i)), sum(i)
+FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ############################################################
+-- FALLBACK cases
+-- ############################################################
+
+-- FIRST/LAST aggregates in PartialMerge mode are order-dependent and
+-- DataFusion's hash table may process rows in a different order than Spark's.
+-- See https://github.com/apache/datafusion-comet/issues/4131
+
+-- ============================================================
+-- fallback: first + distinct count triggers PartialMerge on first
+-- ============================================================
+
+query expect_fallback(PartialMerge not supported for aggregates: first)
+SELECT first(i), count(DISTINCT i) FROM pm_basic
+
+query expect_fallback(PartialMerge not supported for aggregates: first)
+SELECT grp, first(i), count(DISTINCT i) FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- fallback: last + distinct count triggers PartialMerge on last
+-- ============================================================
+
+query expect_fallback(PartialMerge not supported for aggregates: last)
+SELECT last(i), count(DISTINCT i) FROM pm_basic
+
+query expect_fallback(PartialMerge not supported for aggregates: last)
+SELECT grp, last(i), count(DISTINCT i) FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- fallback: first and last together with distinct
+-- ============================================================
+
+query expect_fallback(PartialMerge not supported for aggregates: first, last)
+SELECT first(i), last(i), count(DISTINCT i) FROM pm_basic
+
+query expect_fallback(PartialMerge not supported for aggregates: first, last)
+SELECT grp, first(i), last(i), count(DISTINCT i), sum(i)
+FROM pm_basic GROUP BY grp ORDER BY grp
+
+-- ============================================================
+-- fallback: first/last IGNORE NULLS with distinct
+-- ============================================================
+
+query expect_fallback(PartialMerge not supported for aggregates: first)
+SELECT grp, first(i) IGNORE NULLS, count(DISTINCT i)
+FROM pm_nulls GROUP BY grp ORDER BY grp
+
+query expect_fallback(PartialMerge not supported for aggregates: last)
+SELECT grp, last(i) IGNORE NULLS, count(DISTINCT i)
+FROM pm_nulls GROUP BY grp ORDER BY grp
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_datafusion/extended.txt
index cb53684dc..97b705f04 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_datafusion/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometNativeScan parquet spark_catalog.default.catalog_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.call_center
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometNativeScan parquet spark_catalog.default.catalog_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.call_center
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_iceberg_compat/extended.txt
index 0f23bf5d4..c4efd6fb1 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q16.native_iceberg_compat/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.call_center
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.call_center
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_datafusion/extended.txt
index d05989eb4..6dbeba7e9 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_datafusion/extended.txt
@@ -5,74 +5,68 @@ BroadcastNestedLoopJoin
: : : :- BroadcastNestedLoopJoin [COMET: BroadcastNestedLoopJoin is not supported]
: : : : :- CometNativeColumnarToRow
: : : : : +- CometHashAggregate
-: : : : : +- CometColumnarExchange
-: : : : : +- HashAggregate
-: : : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : : +- CometNativeColumnarToRow
-: : : : : +- CometExchange
-: : : : : +- CometHashAggregate
-: : : : : +- CometProject
-: : : : : +- CometFilter
-: : : : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometHashAggregate
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometProject
+: : : : : +- CometFilter
+: : : : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : : : +- BroadcastExchange
: : : : +- CometNativeColumnarToRow
: : : : +- CometHashAggregate
-: : : : +- CometColumnarExchange
-: : : : +- HashAggregate
-: : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : +- CometNativeColumnarToRow
-: : : : +- CometExchange
-: : : : +- CometHashAggregate
-: : : : +- CometProject
-: : : : +- CometFilter
-: : : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometHashAggregate
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometProject
+: : : : +- CometFilter
+: : : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : : +- BroadcastExchange
: : : +- CometNativeColumnarToRow
: : : +- CometHashAggregate
-: : : +- CometColumnarExchange
-: : : +- HashAggregate
-: : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : +- CometNativeColumnarToRow
-: : : +- CometExchange
-: : : +- CometHashAggregate
-: : : +- CometProject
-: : : +- CometFilter
-: : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometHashAggregate
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometProject
+: : : +- CometFilter
+: : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : +- BroadcastExchange
: : +- CometNativeColumnarToRow
: : +- CometHashAggregate
-: : +- CometColumnarExchange
-: : +- HashAggregate
-: : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : +- CometNativeColumnarToRow
-: : +- CometExchange
-: : +- CometHashAggregate
-: : +- CometProject
-: : +- CometFilter
-: : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometHashAggregate
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometProject
+: : +- CometFilter
+: : +- CometNativeScan parquet spark_catalog.default.store_sales
: +- BroadcastExchange
: +- CometNativeColumnarToRow
: +- CometHashAggregate
-: +- CometColumnarExchange
-: +- HashAggregate
-: +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: +- CometNativeColumnarToRow
-: +- CometExchange
-: +- CometHashAggregate
-: +- CometProject
-: +- CometFilter
-: +- CometNativeScan parquet spark_catalog.default.store_sales
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometHashAggregate
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometProject
+: +- CometFilter
+: +- CometNativeScan parquet spark_catalog.default.store_sales
+- BroadcastExchange
+- CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometExchange
- +- CometHashAggregate
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.store_sales
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.store_sales
-Comet accelerated 42 out of 64 eligible operators (65%). Final plan contains 12 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 54 out of 64 eligible operators (84%). Final plan contains 6 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_iceberg_compat/extended.txt
index a778e244c..35dac9a90 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q28.native_iceberg_compat/extended.txt
@@ -5,74 +5,68 @@ BroadcastNestedLoopJoin
: : : :- BroadcastNestedLoopJoin [COMET: BroadcastNestedLoopJoin is not supported]
: : : : :- CometNativeColumnarToRow
: : : : : +- CometHashAggregate
-: : : : : +- CometColumnarExchange
-: : : : : +- HashAggregate
-: : : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : : +- CometNativeColumnarToRow
-: : : : : +- CometExchange
-: : : : : +- CometHashAggregate
-: : : : : +- CometProject
-: : : : : +- CometFilter
-: : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometHashAggregate
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometProject
+: : : : : +- CometFilter
+: : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : : : +- BroadcastExchange
: : : : +- CometNativeColumnarToRow
: : : : +- CometHashAggregate
-: : : : +- CometColumnarExchange
-: : : : +- HashAggregate
-: : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : +- CometNativeColumnarToRow
-: : : : +- CometExchange
-: : : : +- CometHashAggregate
-: : : : +- CometProject
-: : : : +- CometFilter
-: : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometHashAggregate
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometProject
+: : : : +- CometFilter
+: : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : : +- BroadcastExchange
: : : +- CometNativeColumnarToRow
: : : +- CometHashAggregate
-: : : +- CometColumnarExchange
-: : : +- HashAggregate
-: : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : +- CometNativeColumnarToRow
-: : : +- CometExchange
-: : : +- CometHashAggregate
-: : : +- CometProject
-: : : +- CometFilter
-: : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometHashAggregate
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometProject
+: : : +- CometFilter
+: : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : +- BroadcastExchange
: : +- CometNativeColumnarToRow
: : +- CometHashAggregate
-: : +- CometColumnarExchange
-: : +- HashAggregate
-: : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : +- CometNativeColumnarToRow
-: : +- CometExchange
-: : +- CometHashAggregate
-: : +- CometProject
-: : +- CometFilter
-: : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometHashAggregate
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometProject
+: : +- CometFilter
+: : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: +- BroadcastExchange
: +- CometNativeColumnarToRow
: +- CometHashAggregate
-: +- CometColumnarExchange
-: +- HashAggregate
-: +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: +- CometNativeColumnarToRow
-: +- CometExchange
-: +- CometHashAggregate
-: +- CometProject
-: +- CometFilter
-: +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometHashAggregate
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometProject
+: +- CometFilter
+: +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+- BroadcastExchange
+- CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometExchange
- +- CometHashAggregate
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
-Comet accelerated 42 out of 64 eligible operators (65%). Final plan contains 12 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 54 out of 64 eligible operators (84%). Final plan contains 6 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_datafusion/extended.txt
index c91f0410e..11de43f62 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_datafusion/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometNativeScan parquet spark_catalog.default.web_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometNativeScan parquet spark_catalog.default.web_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.web_site
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_iceberg_compat/extended.txt
index ff3a1a06a..a89558eb8 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q94.native_iceberg_compat/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_datafusion/extended.txt
index 94e8b8d08..45ec12b82 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_datafusion/extended.txt
@@ -1,65 +1,64 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_returns
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometFilter
- : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_returns
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometFilter
+ : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.web_site
-Comet accelerated 59 out of 61 eligible operators (96%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 61 out of 61 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_iceberg_compat/extended.txt
index bcb3e1d44..9a692023a 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_2/q95.native_iceberg_compat/extended.txt
@@ -1,65 +1,64 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometFilter
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometFilter
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
-Comet accelerated 59 out of 61 eligible operators (96%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 61 out of 61 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_datafusion/extended.txt
index cb53684dc..97b705f04 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_datafusion/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometNativeScan parquet spark_catalog.default.catalog_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.call_center
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometNativeScan parquet spark_catalog.default.catalog_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometNativeScan parquet spark_catalog.default.catalog_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.call_center
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_iceberg_compat/extended.txt
index 0f23bf5d4..c4efd6fb1 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.native_iceberg_compat/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.call_center
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.call_center
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_datafusion/extended.txt
index d05989eb4..6dbeba7e9 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_datafusion/extended.txt
@@ -5,74 +5,68 @@ BroadcastNestedLoopJoin
: : : :- BroadcastNestedLoopJoin [COMET: BroadcastNestedLoopJoin is not supported]
: : : : :- CometNativeColumnarToRow
: : : : : +- CometHashAggregate
-: : : : : +- CometColumnarExchange
-: : : : : +- HashAggregate
-: : : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : : +- CometNativeColumnarToRow
-: : : : : +- CometExchange
-: : : : : +- CometHashAggregate
-: : : : : +- CometProject
-: : : : : +- CometFilter
-: : : : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometHashAggregate
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometProject
+: : : : : +- CometFilter
+: : : : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : : : +- BroadcastExchange
: : : : +- CometNativeColumnarToRow
: : : : +- CometHashAggregate
-: : : : +- CometColumnarExchange
-: : : : +- HashAggregate
-: : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : +- CometNativeColumnarToRow
-: : : : +- CometExchange
-: : : : +- CometHashAggregate
-: : : : +- CometProject
-: : : : +- CometFilter
-: : : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometHashAggregate
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometProject
+: : : : +- CometFilter
+: : : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : : +- BroadcastExchange
: : : +- CometNativeColumnarToRow
: : : +- CometHashAggregate
-: : : +- CometColumnarExchange
-: : : +- HashAggregate
-: : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : +- CometNativeColumnarToRow
-: : : +- CometExchange
-: : : +- CometHashAggregate
-: : : +- CometProject
-: : : +- CometFilter
-: : : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometHashAggregate
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometProject
+: : : +- CometFilter
+: : : +- CometNativeScan parquet spark_catalog.default.store_sales
: : +- BroadcastExchange
: : +- CometNativeColumnarToRow
: : +- CometHashAggregate
-: : +- CometColumnarExchange
-: : +- HashAggregate
-: : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : +- CometNativeColumnarToRow
-: : +- CometExchange
-: : +- CometHashAggregate
-: : +- CometProject
-: : +- CometFilter
-: : +- CometNativeScan parquet spark_catalog.default.store_sales
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometHashAggregate
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometProject
+: : +- CometFilter
+: : +- CometNativeScan parquet spark_catalog.default.store_sales
: +- BroadcastExchange
: +- CometNativeColumnarToRow
: +- CometHashAggregate
-: +- CometColumnarExchange
-: +- HashAggregate
-: +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: +- CometNativeColumnarToRow
-: +- CometExchange
-: +- CometHashAggregate
-: +- CometProject
-: +- CometFilter
-: +- CometNativeScan parquet spark_catalog.default.store_sales
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometHashAggregate
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometProject
+: +- CometFilter
+: +- CometNativeScan parquet spark_catalog.default.store_sales
+- BroadcastExchange
+- CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometExchange
- +- CometHashAggregate
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.store_sales
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.store_sales
-Comet accelerated 42 out of 64 eligible operators (65%). Final plan contains 12 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 54 out of 64 eligible operators (84%). Final plan contains 6 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_iceberg_compat/extended.txt
index a778e244c..35dac9a90 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.native_iceberg_compat/extended.txt
@@ -5,74 +5,68 @@ BroadcastNestedLoopJoin
: : : :- BroadcastNestedLoopJoin [COMET: BroadcastNestedLoopJoin is not supported]
: : : : :- CometNativeColumnarToRow
: : : : : +- CometHashAggregate
-: : : : : +- CometColumnarExchange
-: : : : : +- HashAggregate
-: : : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : : +- CometNativeColumnarToRow
-: : : : : +- CometExchange
-: : : : : +- CometHashAggregate
-: : : : : +- CometProject
-: : : : : +- CometFilter
-: : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometHashAggregate
+: : : : : +- CometExchange
+: : : : : +- CometHashAggregate
+: : : : : +- CometProject
+: : : : : +- CometFilter
+: : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : : : +- BroadcastExchange
: : : : +- CometNativeColumnarToRow
: : : : +- CometHashAggregate
-: : : : +- CometColumnarExchange
-: : : : +- HashAggregate
-: : : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : : +- CometNativeColumnarToRow
-: : : : +- CometExchange
-: : : : +- CometHashAggregate
-: : : : +- CometProject
-: : : : +- CometFilter
-: : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometHashAggregate
+: : : : +- CometExchange
+: : : : +- CometHashAggregate
+: : : : +- CometProject
+: : : : +- CometFilter
+: : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : : +- BroadcastExchange
: : : +- CometNativeColumnarToRow
: : : +- CometHashAggregate
-: : : +- CometColumnarExchange
-: : : +- HashAggregate
-: : : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : : +- CometNativeColumnarToRow
-: : : +- CometExchange
-: : : +- CometHashAggregate
-: : : +- CometProject
-: : : +- CometFilter
-: : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometHashAggregate
+: : : +- CometExchange
+: : : +- CometHashAggregate
+: : : +- CometProject
+: : : +- CometFilter
+: : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: : +- BroadcastExchange
: : +- CometNativeColumnarToRow
: : +- CometHashAggregate
-: : +- CometColumnarExchange
-: : +- HashAggregate
-: : +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: : +- CometNativeColumnarToRow
-: : +- CometExchange
-: : +- CometHashAggregate
-: : +- CometProject
-: : +- CometFilter
-: : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometHashAggregate
+: : +- CometExchange
+: : +- CometHashAggregate
+: : +- CometProject
+: : +- CometFilter
+: : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
: +- BroadcastExchange
: +- CometNativeColumnarToRow
: +- CometHashAggregate
-: +- CometColumnarExchange
-: +- HashAggregate
-: +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-: +- CometNativeColumnarToRow
-: +- CometExchange
-: +- CometHashAggregate
-: +- CometProject
-: +- CometFilter
-: +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometHashAggregate
+: +- CometExchange
+: +- CometHashAggregate
+: +- CometProject
+: +- CometFilter
+: +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+- BroadcastExchange
+- CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometExchange
- +- CometHashAggregate
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store_sales
-Comet accelerated 42 out of 64 eligible operators (65%). Final plan contains 12 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 54 out of 64 eligible operators (84%). Final plan contains 6 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_datafusion/extended.txt
index c91f0410e..11de43f62 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_datafusion/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometNativeScan parquet spark_catalog.default.web_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometNativeScan parquet spark_catalog.default.web_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.web_site
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_iceberg_compat/extended.txt
index ff3a1a06a..a89558eb8 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.native_iceberg_compat/extended.txt
@@ -1,43 +1,42 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
-Comet accelerated 37 out of 39 eligible operators (94%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 39 out of 39 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_datafusion/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_datafusion/extended.txt
index 94e8b8d08..45ec12b82 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_datafusion/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_datafusion/extended.txt
@@ -1,65 +1,64 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_returns
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometFilter
- : : : +- CometNativeScan parquet spark_catalog.default.web_sales
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometNativeScan parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometNativeScan parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometNativeScan parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_returns
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometFilter
+ : : : +- CometNativeScan parquet spark_catalog.default.web_sales
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometNativeScan parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometNativeScan parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometNativeScan parquet spark_catalog.default.web_site
-Comet accelerated 59 out of 61 eligible operators (96%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 61 out of 61 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_iceberg_compat/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_iceberg_compat/extended.txt
index bcb3e1d44..9a692023a 100644
--- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_iceberg_compat/extended.txt
+++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.native_iceberg_compat/extended.txt
@@ -1,65 +1,64 @@
CometNativeColumnarToRow
+- CometHashAggregate
- +- CometColumnarExchange
- +- HashAggregate
- +- HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
- +- CometNativeColumnarToRow
- +- CometHashAggregate
- +- CometProject
- +- CometBroadcastHashJoin
- :- CometProject
- : +- CometBroadcastHashJoin
- : :- CometProject
- : : +- CometBroadcastHashJoin
- : : :- CometSortMergeJoin
- : : : :- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometProject
- : : : : +- CometSortMergeJoin
- : : : : :- CometSort
- : : : : : +- CometExchange
- : : : : : +- CometProject
- : : : : : +- CometFilter
- : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : : +- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
- : : : +- CometProject
- : : : +- CometSortMergeJoin
- : : : :- CometSort
- : : : : +- CometExchange
- : : : : +- CometProject
- : : : : +- CometFilter
- : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : : +- CometSort
- : : : +- CometExchange
- : : : +- CometProject
- : : : +- CometFilter
- : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
- : : +- CometBroadcastExchange
- : : +- CometProject
- : : +- CometFilter
- : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
- : +- CometBroadcastExchange
- : +- CometProject
- : +- CometFilter
- : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
- +- CometBroadcastExchange
- +- CometProject
- +- CometFilter
- +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
+ +- CometExchange
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometHashAggregate
+ +- CometProject
+ +- CometBroadcastHashJoin
+ :- CometProject
+ : +- CometBroadcastHashJoin
+ : :- CometProject
+ : : +- CometBroadcastHashJoin
+ : : :- CometSortMergeJoin
+ : : : :- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometProject
+ : : : : +- CometSortMergeJoin
+ : : : : :- CometSort
+ : : : : : +- CometExchange
+ : : : : : +- CometProject
+ : : : : : +- CometFilter
+ : : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : : +- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_returns
+ : : : +- CometProject
+ : : : +- CometSortMergeJoin
+ : : : :- CometSort
+ : : : : +- CometExchange
+ : : : : +- CometProject
+ : : : : +- CometFilter
+ : : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : : +- CometSort
+ : : : +- CometExchange
+ : : : +- CometProject
+ : : : +- CometFilter
+ : : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
+ : : +- CometBroadcastExchange
+ : : +- CometProject
+ : : +- CometFilter
+ : : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
+ : +- CometBroadcastExchange
+ : +- CometProject
+ : +- CometFilter
+ : +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
+ +- CometBroadcastExchange
+ +- CometProject
+ +- CometFilter
+ +- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_site
-Comet accelerated 59 out of 61 eligible operators (96%). Final plan contains 2 transitions between Spark and Comet.
\ No newline at end of file
+Comet accelerated 61 out of 61 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet.
\ No newline at end of file
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
index 95f3774e0..f3639c8fe 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -638,10 +638,10 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
dictionaryEnabled) {
withView("v") {
sql("CREATE TEMP VIEW v AS SELECT _1, _2 FROM tbl ORDER BY _1")
- checkSparkAnswerAndFallbackReason(
+ checkSparkAnswerAndOperator(
"SELECT _2, SUM(_1), SUM(DISTINCT _1), MIN(_1), MAX(_1), COUNT(_1)," +
- " COUNT(DISTINCT _1), AVG(_1), FIRST(_1), LAST(_1) FROM v GROUP BY _2",
- "Unsupported aggregation mode PartialMerge")
+ " COUNT(DISTINCT _1), AVG(_1)" +
+ " FROM v GROUP BY _2 ORDER BY _2")
}
}
}
@@ -650,6 +650,89 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ // FIRST/LAST are order-dependent aggregates whose merge result depends on hash table
+ // processing order. In PartialMerge mode, DataFusion's hash table may process rows
+ // in a different order than Spark's, so we fall back to Spark for correctness.
+ // https://github.com/apache/datafusion-comet/issues/4131
+ test("partialMerge - FIRST/LAST with distinct aggregates falls back") {
+ val numValues = 10000
+ Seq(100).foreach { numGroups =>
+ Seq(128).foreach { batchSize =>
+ withSQLConf(
+ SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> batchSize.toString) {
+ withParquetTable(
+ (0 until numValues).map(i => (i, Random.nextInt() % numGroups)),
+ "tbl",
+ false) {
+ withView("v") {
+ sql("CREATE TEMP VIEW v AS SELECT _1, _2 FROM tbl ORDER BY _1")
+ checkSparkAnswerAndFallbackReason(
+ "SELECT _2, FIRST(_1), LAST(_1), COUNT(DISTINCT _1)" +
+ " FROM v GROUP BY _2 ORDER BY _2",
+ "PartialMerge not supported for aggregates: first, last")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("partialMerge - cnt distinct + sum") {
+ withTempDir(dir => {
+ withSQLConf("spark.comet.enabled" -> "false") {
+ withView("t") {
+ sql("""
+ CREATE OR REPLACE TEMP VIEW t (v, v1, i) AS
+ VALUES
+ ('c', 'a', 1),
+ ('c1', 'a1', 1),
+ ('c2', 'a2', 2),
+ ('c3', 'a3', 2),
+ ('c4', 'a4', 2),
+ ('c', 'a', 1),
+ ('c1', 'a1', 1),
+ ('c2', 'a2', 2),
+ ('c3', 'a3', 2),
+ ('c4', 'a4', 2),
+ ('c', 'a', 1),
+ ('c1', 'a1', 1),
+ ('c2', 'a2', 2),
+ ('c3', 'a3', 2),
+ ('c4', 'a4', 2)
+ """)
+ sql("select * from t")
+ .repartition(3)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getAbsolutePath)
+ }
+ }
+
+ withSQLConf(
+ CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
+ "spark.comet.exec.shuffle.fallbackToColumnar" -> "false",
+ "spark.comet.enabled" -> "true") {
+ spark.read.parquet(dir.getAbsolutePath).createOrReplaceTempView("t2")
+ checkSparkAnswerAndOperator("SELECT i, sum(v1), count(distinct v) FROM t2 group by i")
+ }
+ })
+ }
+
+ test("partialMerge - distinct + non-distinct aggregates (Expand pattern)") {
+ withParquetTable((1 to 100).map(i => (i, i.toString)), "tbl", false) {
+ checkSparkAnswerAndOperator("SELECT avg(_1), sum(_1), count(distinct _1) FROM tbl")
+
+ checkSparkAnswerAndOperator(
+ "SELECT max(_1), count(distinct _1), sum(distinct _1), sum(1) FROM tbl")
+
+ // Exercises the Expand plan for a non-COUNT distinct (AVG(DISTINCT)) mixed with a
+ // non-distinct aggregate, so the PartialMerge stage has to survive for a float-producing
+ // distinct aggregate.
+ checkSparkAnswerAndOperator("SELECT avg(distinct _1), sum(_1) FROM tbl")
+ }
+ }
+
test("multiple group-by columns + single aggregate column (first/last), with nulls") {
val numValues = 10000
@@ -1108,54 +1191,40 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
- // TODO enable once https://github.com/apache/datafusion-comet/issues/1267 is implemented
- ignore("distinct") {
+ test("distinct") {
withSQLConf(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true") {
Seq("native", "jvm").foreach { cometShuffleMode =>
withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> cometShuffleMode) {
Seq(true, false).foreach { dictionary =>
withSQLConf("parquet.enable.dictionary" -> dictionary.toString) {
- val cometColumnShuffleEnabled = cometShuffleMode == "jvm"
val table = "test"
withTable(table) {
sql(s"create table $table(col1 int, col2 int, col3 int) using parquet")
sql(
s"insert into $table values(1, 1, 1), (1, 1, 1), (1, 3, 1), (1, 4, 2), (5, 3, 2)")
- var expectedNumOfCometAggregates = 2
+ checkSparkAnswerAndOperator(s"SELECT DISTINCT(col2) FROM $table")
- checkSparkAnswerAndNumOfAggregates(
- s"SELECT DISTINCT(col2) FROM $table",
- expectedNumOfCometAggregates)
+ // Keep one explicit aggregate-count assertion so a silent regression in the
+ // middle PartialMerge stage (which used to fall back before this PR) is caught:
+ // all four Spark aggregate stages for COUNT(DISTINCT ...) must remain native.
+ checkSparkAnswerAndNumOfAggregates(s"SELECT COUNT(distinct col2) FROM $table", 4)
- expectedNumOfCometAggregates = 4
+ checkSparkAnswerAndOperator(
+ s"SELECT COUNT(distinct col2), col1 FROM $table group by col1")
- checkSparkAnswerAndNumOfAggregates(
- s"SELECT COUNT(distinct col2) FROM $table",
- expectedNumOfCometAggregates)
-
- checkSparkAnswerAndNumOfAggregates(
- s"SELECT COUNT(distinct col2), col1 FROM $table group by col1",
- expectedNumOfCometAggregates)
-
- checkSparkAnswerAndNumOfAggregates(
- s"SELECT SUM(distinct col2) FROM $table",
- expectedNumOfCometAggregates)
+ checkSparkAnswerAndOperator(s"SELECT SUM(distinct col2) FROM $table")
- checkSparkAnswerAndNumOfAggregates(
- s"SELECT SUM(distinct col2), col1 FROM $table group by col1",
- expectedNumOfCometAggregates)
+ checkSparkAnswerAndOperator(
+ s"SELECT SUM(distinct col2), col1 FROM $table group by col1")
- checkSparkAnswerAndNumOfAggregates(
+ checkSparkAnswerAndOperator(
"SELECT COUNT(distinct col2), SUM(distinct col2), col1, COUNT(distinct col2)," +
- s" SUM(distinct col2) FROM $table group by col1",
- expectedNumOfCometAggregates)
+ s" SUM(distinct col2) FROM $table group by col1")
- expectedNumOfCometAggregates = if (cometColumnShuffleEnabled) 2 else 1
- checkSparkAnswerAndNumOfAggregates(
+ checkSparkAnswerAndOperator(
"SELECT COUNT(col2), MIN(col2), COUNT(DISTINCT col2), SUM(col2)," +
- s" SUM(DISTINCT col2), COUNT(DISTINCT col2), col1 FROM $table group by col1",
- expectedNumOfCometAggregates)
+ s" SUM(DISTINCT col2), COUNT(DISTINCT col2), col1 FROM $table group by col1")
}
}
}
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
index 1ac165975..1480e5f90 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -2092,9 +2092,11 @@ class CometExecSuite extends CometTestBase {
case s: CometHashAggregateExec => s
}.get
- assert(agg.mode.isDefined && agg.mode.get.isInstanceOf[AggregateMode])
+ assert(agg.modes.nonEmpty && agg.modes.headOption.get.isInstanceOf[AggregateMode])
val newAgg = agg.cleanBlock().asInstanceOf[CometHashAggregateExec]
- assert(newAgg.mode.isDefined && newAgg.mode.get.isInstanceOf[AggregateMode])
+ assert(
+ newAgg.modes.nonEmpty &&
+ newAgg.modes.headOption.get.isInstanceOf[AggregateMode])
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org