comphead commented on code in PR #4565:
URL: https://github.com/apache/datafusion-comet/pull/4565#discussion_r4011490169
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2051,29 +2043,72 @@ object CometObjectHashAggregateExec
}
}
-case class CometHashAggregateExec(
- override val nativeOp: Operator,
- override val originalPlan: SparkPlan,
- override val output: Seq[Attribute],
- groupingExpressions: Seq[NamedExpression],
- aggregateExpressions: Seq[AggregateExpression],
- resultExpressions: Seq[NamedExpression],
- input: Seq[Attribute],
- child: SparkPlan,
- override val serializedPlanOpt: SerializedPlan)
+object CometSortAggregateExec
+ extends CometOperatorSerde[SortAggregateExec]
+ with CometBaseAggregate {
+
+ override def enabledConfig: Option[ConfigEntry[Boolean]] = Some(
+ CometConf.COMET_EXEC_AGGREGATE_ENABLED)
+
+ override def getSupportLevel(op: SortAggregateExec): SupportLevel =
+ baseAggregateSupportLevel(op)
+
+ override def convert(
+ aggregate: SortAggregateExec,
+ builder: Operator.Builder,
+ childOp: OperatorOuterClass.Operator*):
Option[OperatorOuterClass.Operator] = {
+
+ // SortAggregate is planned for TypedImperativeAggregate functions whose
intermediate
+ // buffer formats differ between Spark and Comet (same risk as
ObjectHashAggregate).
+ // Require Comet shuffle so a Partial->Final pair never spans the
JVM/native boundary.
+ if (!isCometShuffleEnabled(aggregate.conf)) {
+ return None
+ }
+
+ doConvert(aggregate, builder, childOp: _*)
+ }
+
+ override def createExec(nativeOp: Operator, op: SortAggregateExec):
CometNativeExec = {
+ // The native AggregateExec auto-detects Sorted input mode from the
child's output ordering
Review Comment:
**Retracting this — I tested it and the ordering holds. Sorry for the
noise.**
What I got wrong: I assumed `InputOrderMode::Linear` means unordered output.
It does not. DataFusion's hash aggregate emits groups in *first-seen* order
(`GroupValuesPrimitive::emit` / `GroupValuesRows::emit` walk the group-index
vector for `EmitTo::All`), and there is no early emit in Linear mode. Since
Spark guarantees `SortAggregateExec`'s child satisfies `requiredChildOrdering`,
first-seen order is ascending grouping-key order, so the output is sorted
either way.
Verified on Spark 4.0 / Scala 2.13 with a bucketed, physically sorted
single-bucket table, so the plan was `CometHashAggregate[Final] <-
CometHashAggregate[Partial] <- CometNativeScan` with no `SortExec` anywhere
(i.e. Linear mode, the exact shape I was worried about):
```
partition 0: rows=97 ascending=true
partition 0: rows=400000 ascending=true (400k distinct 192-byte string
keys)
```
Two things I checked along the way that may still be worth a line of your
time:
1. **My "bucketed table" premise needed a non-default config.**
`spark.sql.legacy.bucketedTableScan.outputOrdering` defaults to `false`, so a
bucketed+sorted scan reports no ordering and the pre-aggregate `Sort` is *not*
elided. With it set to `true` the sorts do disappear, including the one above
the aggregate feeding a `SortMergeJoin`. But a **cached sorted relation reaches
the same shape with entirely default configs** — `InMemoryTableScanExec`
reports the ordering, and `SELECT k, collect_list(v) FROM cached_sorted GROUP
BY k` plans with zero `SortExec`. So "sort aggregate whose ordering is not
produced inside the native block" is a real, default-reachable plan shape; it
just happens to still come out sorted.
2. **The comment here describes a mechanism that is not what actually
holds.** "The native AggregateExec auto-detects Sorted input mode from the
child's output ordering" is only true when the pre-aggregate sort is in the
same native block; Comet's native leaf reports
`EquivalenceProperties::new(schema)` (`operators/scan.rs:78`), so in the shapes
above DataFusion picks Linear. The invariant that actually carries the
correctness here is the weaker and undocumented one: *DataFusion's hash
aggregate emits in first-seen group order, and Spark guarantees the input is
sorted on the grouping keys.* Worth stating that way instead, since it is
load-bearing and a future change to DataFusion's emission order (or a
`PartialReduce`-style mode, which is already forced to Linear precisely because
it "can't promise" the output ordering) would break `outputOrdering` silently.
One sub-case I could not exercise: spilling. I could not get the aggregate
to spill (`spill_count` stayed 0 down to a ~24 MB pool). Reading
`grouped_hash_stream.rs::spill()`, it sorts the emitted state on all output
fields with `SortOptions::default()` (ascending, nulls first — matching Spark's
`SortOrder(_, Ascending)`) and streaming-merges on that ordering, so it should
enforce the ordering rather than break it. Source-only, not tested.
--
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]
