You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/14 02:40:55 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6899-8a4029ccf9a079385224bef8aca95d561993b221 created (now 33f4f888a0)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6899-8a4029ccf9a079385224bef8aca95d561993b221
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 33f4f888a0 feat: run global aggregates with no grouping keys or functions natively (#6899)
This branch includes the following new commits:
new 33f4f888a0 feat: run global aggregates with no grouping keys or functions natively (#6899)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: run global aggregates with no grouping keys or functions natively (#6899)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6899-8a4029ccf9a079385224bef8aca95d561993b221
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 33f4f888a0f49c1eb933149c1f0c502a7b192748
Author: Andy Grove <ag...@apache.org>
AuthorDate: Wed Oct 14 02:40:35 2026 +0000
feat: run global aggregates with no grouping keys or functions natively (#6899)
* feat: run global aggregates with no grouping keys or functions natively
Spark plans `Aggregate(Nil, Nil, child)` when nothing reads a global
aggregate's output, as in `df.agg(sum($"a")).count()`. Comet fell back
to Spark for it with "No group by or aggregation".
DataFusion's AggregateStream can't build a batch with no columns
(apache/datafusion#25782), so the native planner plans the shape as a
count(1) aggregate under an empty projection. That needs no DataFusion
change.
Closes #6001.
* test: cover a Spark partial aggregate under a native final one
Also write the fixture's multi-file table with an explicit partition count, so it doesn't depend on the harness's parallelism.
* test: say why the fixture's five files become five partitions
* refactor: build the count with the planner's aggregate helper
Also drop test assertions that another assertion already implies, and shorten the serde comment that repeated the native planner's.
* docs: say newer DataFusion rejects an aggregate with no keys or functions
apache/datafusion#25786 makes AggregateExec::try_new reject that shape, so the count plan is how Comet runs it from now on, not a stopgap.
* perf: pass a NULL literal to the throwaway count
The aggregate expands its argument to each input batch's length.
Int64(1) filled an Int64Array per batch only for the projection to drop
the count, while NULL comes back as a cached NullArray with no buffers.
Also shorten the planner comment.
* test: cover HAVING without GROUP BY over an empty global aggregate
The parser plans it as a global aggregate that outputs a literal. Also
note that CollapseProject already gives the SELECT 1 queries a result
projection.
---
native/core/src/execution/planner.rs | 109 +++++++++++++++++++++
.../org/apache/comet/rules/CometExecRule.scala | 2 -
.../org/apache/spark/sql/comet/operators.scala | 9 +-
.../aggregate/global_empty_aggregate.sql | 77 +++++++++++++++
.../apache/comet/exec/CometAggregateSuite.scala | 28 +++++-
5 files changed, 217 insertions(+), 8 deletions(-)
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index e5412477f7..b1b9a3bcc9 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -1750,6 +1750,45 @@ impl PhysicalPlanner {
let (scans, shuffle_scans, child) =
self.create_plan(&children[0], inputs, partition_count)?;
+ // For `df.agg(sum($"a")).count()`, Spark plans a global aggregate with neither
+ // keys nor functions that emits one row with no columns per partition. DataFusion
+ // can't run that (apache/datafusion#25786), so plan a throwaway `count(NULL)` and
+ // project it away. The count still drains the input, so errors below it surface.
+ // Both Spark stages take this path, since no count passes between them.
+ if agg.grouping_exprs.is_empty() && agg.agg_exprs.is_empty() {
+ let schema = child.schema();
+ let count = Self::create_aggr_func_expr(
+ "count",
+ Arc::clone(&schema),
+ vec![Arc::new(Literal::new(ScalarValue::Null))],
+ Arc::unwrap_or_clone(count_udaf()),
+ )?;
+ let aggregate: Arc<dyn ExecutionPlan> = Arc::new(
+ datafusion::physical_plan::aggregates::AggregateExec::try_new(
+ DFAggregateMode::Single,
+ PhysicalGroupBy::default(),
+ vec![Arc::new(count)],
+ vec![None],
+ Arc::clone(&child.native_plan),
+ schema,
+ )?,
+ );
+ let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ Vec::<(Arc<dyn PhysicalExpr>, String)>::new(),
+ Arc::clone(&aggregate),
+ )?);
+ return Ok((
+ scans,
+ shuffle_scans,
+ Arc::new(SparkPlan::new_with_additional(
+ spark_plan.plan_id,
+ projection,
+ vec![child],
+ vec![aggregate],
+ )),
+ ));
+ }
+
let group_exprs: PhyExprResult = agg
.grouping_exprs
.iter()
@@ -6351,6 +6390,76 @@ mod tests {
assert_eq!("ScanExec", projection_exec.children[0].native_plan.name());
}
+ #[tokio::test]
+ async fn keyless_aggregate_without_functions_emits_one_empty_row() {
+ use arrow::array::RecordBatchOptions;
+ use datafusion::common::tree_node::{Transformed, TreeNode};
+
+ let scan = Operator {
+ op_struct: Some(OpStruct::Scan(spark_operator::Scan {
+ fields: vec![],
+ source: String::new(),
+ })),
+ ..Default::default()
+ };
+ let schema = Arc::new(Schema::empty());
+ let rows = |num_rows| {
+ RecordBatch::try_new_with_options(
+ Arc::clone(&schema),
+ vec![],
+ &RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ .unwrap()
+ };
+ let ctx = SessionContext::new();
+ let task_ctx = ctx.task_ctx();
+ let planner = PhysicalPlanner::new(Arc::new(ctx), 0);
+
+ // Spark serializes the stage below its exchange as Partial and the one above as Final.
+ // Both emit one row with no columns per partition, even for an empty input.
+ for mode in [
+ spark_operator::AggregateMode::Partial,
+ spark_operator::AggregateMode::Final,
+ ] {
+ let aggregate = Operator {
+ plan_id: 1,
+ children: vec![scan.clone()],
+ op_struct: Some(OpStruct::HashAgg(spark_operator::HashAggregate {
+ mode: mode as i32,
+ ..Default::default()
+ })),
+ ..Default::default()
+ };
+ let (_, _, planned) = planner.create_plan(&aggregate, &mut vec![], 1).unwrap();
+ assert_eq!("ProjectionExec", planned.native_plan.name());
+ assert_eq!(0, planned.schema().fields().len());
+ assert_eq!(1, planned.additional_native_plans.len());
+ assert_eq!("AggregateExec", planned.additional_native_plans[0].name());
+ assert_eq!(1, planned.children.len());
+ assert_eq!("ScanExec", planned.children[0].native_plan.name());
+
+ for batches in [vec![rows(3), rows(2)], vec![]] {
+ let input: Arc<dyn ExecutionPlan> =
+ MemorySourceConfig::try_new_exec(&[batches], Arc::clone(&schema), None)
+ .unwrap();
+ let native_plan = Arc::clone(&planned.native_plan)
+ .transform_up(|node| {
+ Ok(if node.name() == "ScanExec" {
+ Transformed::yes(Arc::clone(&input))
+ } else {
+ Transformed::no(node)
+ })
+ })
+ .unwrap()
+ .data;
+ let results = collect(native_plan.execute(0, Arc::clone(&task_ctx)).unwrap())
+ .await
+ .unwrap();
+ assert_eq!(1, results.iter().map(RecordBatch::num_rows).sum::<usize>());
+ }
+ }
+ }
+
#[tokio::test]
async fn sort_aggregate_output_is_ordered_by_grouping_keys() {
use arrow::datatypes::Int32Type;
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 5a6e761d44..5a09275a16 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -1725,8 +1725,6 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
val aggregateExpressions = agg.aggregateExpressions
val groupingExpressions = agg.groupingExpressions
- if (groupingExpressions.isEmpty && aggregateExpressions.isEmpty) return false
-
if (groupingExpressions.exists(e =>
SupportLevel.containsType(e.dataType, classOf[MapType]))) {
return false
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 a1cc6f04c4..a8f5ae2704 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
@@ -2265,11 +2265,6 @@ trait CometBaseAggregate[T <: BaseAggregateExec] extends CometOperatorSerde[T] {
val resultExpressions = aggregate.resultExpressions
val child = aggregate.child
- if (groupingExpressions.isEmpty && aggregateExpressions.isEmpty) {
- withFallbackReason(aggregate, "No group by or aggregation")
- return None
- }
-
if (groupingExpressions.exists(expr =>
SupportLevel.containsType(expr.dataType, classOf[MapType]))) {
withFallbackReason(aggregate, "Grouping on map-containing types is not supported")
@@ -2314,6 +2309,10 @@ trait CometBaseAggregate[T <: BaseAggregateExec] extends CometOperatorSerde[T] {
// +- FileScan parquet spark_catalog.default.test[col1#6, col2#7] ......
// If the aggregateExpressions is empty, we only want to build groupingExpressions,
// and skip processing of aggregateExpressions.
+ //
+ // The groupingExpressions can be empty too, for the `Aggregate(Nil, Nil, child)` that
+ // ColumnPruning leaves when nothing reads a global aggregate's output. The native planner
+ // handles that shape.
if (aggregateExpressions.isEmpty) {
val hashAggBuilder = OperatorOuterClass.HashAggregate.newBuilder()
hashAggBuilder.addAllGroupingExprs(groupingExprs.map(_.get).asJava)
diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/global_empty_aggregate.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/global_empty_aggregate.sql
new file mode 100644
index 0000000000..4c45eebd03
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/global_empty_aggregate.sql
@@ -0,0 +1,77 @@
+-- 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.
+
+-- ColumnPruning leaves a global aggregate with neither keys nor functions when nothing reads its
+-- output. Each stage of the partial/final pair emits one row with no columns per partition, even
+-- for an empty input.
+-- Config: spark.comet.exec.localTableScan.enabled=true
+
+-- Five files. With the test session's parallelism of 5, the scan reads them as five partitions, so
+-- the final stage collapses five rows into one.
+statement
+CREATE TABLE empty_agg_multi(a int) USING parquet
+
+statement
+INSERT INTO empty_agg_multi SELECT CAST(id AS INT) FROM range(0, 10, 1, 5)
+
+-- No files at all.
+statement
+CREATE TABLE empty_agg_none(a int) USING parquet
+
+statement
+CREATE TABLE empty_agg_division(n int, d int) USING parquet
+
+statement
+INSERT INTO empty_agg_division SELECT 1, 0
+
+-- CollapseProject folds the outer SELECT into the aggregate, so the final stage projects the
+-- literal over the row with no columns.
+query
+SELECT 1 FROM (SELECT count(*) FROM empty_agg_multi)
+
+query
+SELECT 1 FROM (SELECT count(*) FROM empty_agg_multi WHERE a > 100)
+
+query
+SELECT 1 FROM (SELECT count(*) FROM empty_agg_none)
+
+-- Dataset.count() over an aggregated DataFrame plans this shape.
+query
+SELECT count(*) FROM (SELECT sum(a) FROM empty_agg_multi)
+
+query
+SELECT count(*) FROM (SELECT sum(a) FROM empty_agg_none)
+
+query
+SELECT 1 FROM (SELECT count(*) FROM VALUES (1) AS v(a))
+
+query
+SELECT 1 FROM (SELECT count(*) FROM VALUES (1) AS v(a) WHERE a > 5)
+
+-- The parser plans HAVING without GROUP BY as a global aggregate, here one that outputs a literal.
+query
+SELECT 1 AS x FROM empty_agg_multi HAVING true
+
+query
+SELECT 1 AS x FROM empty_agg_none HAVING true
+
+-- The aggregate still reads its whole input, so an error below it surfaces as it does in Spark.
+statement
+SET spark.sql.ansi.enabled=true
+
+query expect_error_class(DIVIDE_BY_ZERO)
+SELECT 1 FROM (SELECT count(*) FROM empty_agg_division WHERE n / d > 0)
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 faaf81a3e1..2886b56718 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
import org.apache.spark.sql.execution.LocalTableScanExec
import org.apache.spark.sql.execution.SQLExecution
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AdaptiveSparkPlanHelper, ShuffleQueryStageExec}
-import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
+import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExchangeExec}
import org.apache.spark.sql.functions.{avg, col, collect_list, collect_set, count_distinct, expr, sort_array, sum}
import org.apache.spark.sql.internal.SQLConf
@@ -4129,4 +4129,30 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ // global_empty_aggregate.sql covers the SQL shapes. A SQL query always projects a column, so
+ // only a DataFrame hands the zero-column row of `Aggregate(Nil, Nil, child)` to the JVM, where
+ // losing its row count would return no rows.
+ test("global aggregate with no keys and no functions returns one empty row") {
+ withParquetTable((0 until 10).map(Tuple1(_)), "tbl") {
+ Seq(spark.table("tbl"), spark.table("tbl").where($"_1" > 100)).foreach { input =>
+ checkSparkAnswerAndOperator(input.agg(sum($"_1")).select())
+ }
+ }
+ }
+
+ // Local table scans run in Spark by default, so the partial aggregate stays in Spark and the
+ // native final aggregate reads its rows through a JVM columnar shuffle.
+ test("global aggregate with no keys and no functions over a Spark partial aggregate") {
+ withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "false") {
+ val (_, cometPlan) = checkSparkAnswerAndOperator(
+ sql("SELECT * FROM VALUES (1), (2) AS v(a)").agg(sum($"a")).select(),
+ Seq(classOf[CometHashAggregateExec]),
+ classOf[HashAggregateExec],
+ classOf[LocalTableScanExec])
+ assert(
+ collect(cometPlan) { case agg: HashAggregateExec => agg }.nonEmpty,
+ s"Expected a Spark partial aggregate; plan:\n$cometPlan")
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org