andygrove opened a new pull request, #6899: URL: https://github.com/apache/datafusion-comet/pull/6899
## Which issue does this PR close? Closes #6001. ## Rationale for this change Spark plans a global aggregate with neither grouping keys nor aggregate functions whenever nothing reads the aggregate's output. ColumnPruning leaves `Aggregate(Nil, Nil, child)`, and `Dataset.count()` on an aggregated DataFrame, as in `df.agg(sum($"a")).count()`, is the most common way to get one. Comet fell back to Spark for both stages of the partial/final pair with "No group by or aggregation", which forced a transition right above the scan. #6235 by @hsiang-c removes that fallback and plans an `AggregateExec` with no aggregate expressions. DataFusion's `AggregateStream` builds the output row with `RecordBatch::try_new`, which rejects a batch with no columns (`must either specify a row count or at least one column`). So that approach needs apache/datafusion#25783 and a DataFusion release that includes it. DataFusion 55.2.0 still fails. This PR gets the same plans running natively without the upstream change. Once a DataFusion release includes the fix, the native planner can plan the empty aggregate directly, as #6235 does. ## What changes are included in this PR? - `CometBaseAggregate.doConvert` and `CometExecRule.canAggregateBeConverted` no longer reject an aggregate with no grouping keys and no aggregate functions. This is the Scala change from #6235. - The native planner plans that shape as `AggregateExec(Single, [], [count(1)])` under a `ProjectionExec` with no expressions. The count gives the aggregate a column to build its one row with, and the projection drops the column but keeps the row. Counting reads the whole input, as Spark does, so an error below the aggregate still surfaces. The partial and final stages both take this path. The aggregate is registered as an additional native plan, so its metrics roll up under the Spark operator. ## How are these changes tested? - A native planner test runs the planned operator in both modes, over zero-column input with rows and over no input, and checks that each emits exactly one row with no columns. With the plan from #6235 it fails with the DataFusion error above. - `global_empty_aggregate.sql` covers a multi-file table, a filter that removes every row, a table with no files, `count(*)` over an aggregate, `VALUES`, and an ANSI `DIVIDE_BY_ZERO` raised by a filter below the aggregate. - A `CometAggregateSuite` test covers `.select()` and `isEmpty` on an aggregated DataFrame. These are the shapes that hand the zero-column row from native code to the JVM, which a SQL query can't do because it always projects a column. - Locally on the Spark 4.1 profile, the full `CometAggregateSuite` and every aggregate and operator SQL fixture pass. I also ran these shapes with AQE off, JVM shuffle, Comet shuffle off, shuffle direct read off, and local table scan off, and every result matched Spark. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
