comphead opened a new issue, #6001:
URL: https://github.com/apache/datafusion-comet/issues/6001
### What is the problem the feature request solves?
`CometBaseAggregate.doConvert` refuses any `HashAggregateExec` whose
`groupingExpressions` **and** `aggregateExpressions` are both empty:
```scala
// spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
if (groupingExpressions.isEmpty && aggregateExpressions.isEmpty) {
withFallbackReason(aggregate, "No group by or aggregation")
return None
}
```
Spark plans exactly that operator whenever a **global** aggregate's output
is unused, which is not an exotic shape:
1. `ColumnPruning` rewrites `Project(_, Aggregate(Nil, [count(1) AS c],
child))` into `Aggregate(Nil, Nil, child)` (`Optimizer.scala`, `case p @
Project(_, a: Aggregate)`), and `RemoveNoopOperators` drops the now-empty
`Project`.
2. `RemoveRedundantAggregates` cannot remove the aggregate: it explicitly
refuses when the lower aggregate is global (`lowerIsGlobalAgg`).
3. `OptimizeOneRowPlan` cannot remove it either, not even for a single-row
input, because `Aggregate.groupOnly` requires non-empty `groupingExpressions`.
4. `AggUtils.planAggregateWithoutDistinct` still emits a partial/final pair,
and the final requires `AllTuples`, so an `Exchange SinglePartition` lands
between the two.
Both halves of that pair fall back, and a transition is forced immediately
above the scan.
**Minimal repro.** Comet `1.1.0-SNAPSHOT` (main @ 58ab5f618e), Spark 4.0.0,
the documented `spark-shell` setup plus
`spark.comet.exec.localTableScan.enabled=true` and
`spark.sql.adaptive.enabled=false`:
```sql
SELECT 1 FROM (SELECT count(*) FROM VALUES (1) AS v(a))
```
```
HashAggregate [COMET: No group by or aggregation]
+- Exchange
+- HashAggregate [COMET: No group by or aggregation]
+- CometColumnarToRow
+- CometLocalTableScan
```
> Comet accelerated 1 out of 4 eligible operators (25%). Final plan contains
1 transitions between Spark and Comet.
Not specific to local relations. The same fallback happens over Parquet,
where the scan is pruned to no columns:
```
HashAggregate [COMET: No group by or aggregation]
+- Exchange
+- HashAggregate [COMET: No group by or aggregation]
+- CometColumnarToRow
+- CometNativeScan parquet [] ... ReadSchema: struct<>
```
The likeliest way to hit this in real code is calling `.count()` on an
aggregated DataFrame. `Dataset.count()` is `groupBy().count()`, and that outer
count references nothing from the inner aggregate, so the inner one gets pruned
to the empty form:
```scala
Seq(1, 2).toDF("a").agg(sum($"a")).count()
```
```
HashAggregate
+- HashAggregate
+- HashAggregate [COMET: No group by or aggregation]
+- Exchange
+- HashAggregate [COMET: No group by or aggregation]
+- CometColumnarToRow
+- CometLocalTableScan
```
> Comet accelerated 1 out of 6 eligible operators (16%).
### Describe the potential solution
Serialize the operator rather than bailing out. Its contract is narrow and
needs none of the aggregate machinery:
> emit exactly one row with **no columns**, regardless of input cardinality.
The "regardless" is load-bearing, since a global aggregate emits a row even
for empty input. Verified: `SELECT 1 FROM (SELECT count(*) FROM VALUES (1) AS
v(a) WHERE a > 5)` plans to the same pair over `CometLocalTableScan <empty>`
and still returns one row. The two stages differ only in scope, the pre-shuffle
one emitting one empty row per partition and the post-shuffle one collapsing
those to a single row.
The awkward part is the zero-column output: the row count cannot ride in a
column, so this needs a batch with `num_rows = 1` and no arrays to survive the
JNI boundary and the native `ScanExec`. Whether that is representable end to
end today is a question for someone closer to the native side, and it decides
whether this is a small serde addition or needs real plumbing.
Note also that the predicate is duplicated.
`CometExecRule.canAggregateBeConverted` carries the same
`groupingExpressions.isEmpty && aggregateExpressions.isEmpty` early return,
with a comment warning that the two sites must stay in sync, so both would need
the change.
If native support turns out not to be worth the plumbing, this can be closed
as documented behaviour, but `df.agg(...).count()` seems common enough to
measure first.
### Additional context
Environment: macOS arm64, Spark 4.0.0, Comet `1.1.0-SNAPSHOT` built from
main @ 58ab5f618e, `spark.comet.exec.localTableScan.enabled=true`, AQE disabled
so the tree is stable.
One reporting quirk noticed while narrowing this down: which of the two
nodes carries the `[COMET: ...]` annotation depends on whether Comet shuffle is
enabled. With `spark.comet.shuffle.enabled=false`, only the aggregate below the
exchange is annotated; the one above it is never attempted and so falls back
with no reason recorded. That may be relevant to #2787.
--
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]
