Global aggregate with no grouping keys and no aggregate functions falls back to Spark ("No group by or aggregation")

[comphead]

What is the problem the feature request solves?

CometBaseAggregate.doConvert refuses any HashAggregateExec whose groupingExpressions and aggregateExpressions are both empty:

// 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 @ 58ab5f6), Spark 4.0.0, the documented spark-shell setup plus spark.comet.exec.localTableScan.enabled=true and spark.sql.adaptive.enabled=false:

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:

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 @ 58ab5f6, 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.

[hsiang-c]

take