comphead commented on code in PR #4565:
URL: https://github.com/apache/datafusion-comet/pull/4565#discussion_r4011423669
##########
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:
**Correction: my original comment here claimed a correctness bug. I tested
it and was wrong — see the reply below. The ordering is preserved.** What
survives is narrower: this comment names a mechanism that is not the one
actually doing the work.
"The native AggregateExec auto-detects Sorted input mode from the child's
output ordering" only holds when the pre-aggregate `SortExec` is inside this
native block. When it is not — and a cached sorted relation reaches that shape
with entirely default configs — the native leaf advertises no ordering
(`native/core/src/execution/operators/scan.rs`,
`EquivalenceProperties::new(schema)`) and DataFusion picks `Linear`. The output
is still correctly ordered, just for a different reason: the hash aggregate
emits groups in first-seen order, and Spark guarantees the child satisfies
`requiredChildOrdering`.
Worth writing down the invariant that actually holds, because it is
load-bearing and a change to DataFusion's emission order would break
`outputOrdering` silently — `AggregateMode::PartialReduce` is already pinned to
`Linear` for exactly that reason ("Input order mode is also used to advertise
plan output ordering ... partial reduce aggregation can't promise that").
Suggested replacement for the four comment lines here:
```scala
// CometExec.outputOrdering reports SortAggregateExec's grouping-key
ordering, and
// EnsureRequirements has already elided downstream sorts against it, so
native execution
// must honor it. It does under both input order modes: DataFusion picks
Sorted when the
// pre-aggregate sort is in this native block, and Linear otherwise (the
native ScanExec
// advertises no ordering), where the hash aggregate still emits groups
in first-seen
// order, which is grouping-key order because Spark guarantees the child
is sorted.
```
The `CometBaseAggregateExec` scaladoc carries the same "auto-detects sorted
input mode from the child ordering" phrasing and has the same problem; dropping
that clause there and keeping the explanation in one place would cover both.
--
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]
