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:
**Possible correctness issue: the ordering claim may not hold.**
`SortAggregateExec` is an `OrderPreservingUnaryExecNode`, so
`CometExec.outputOrdering = originalPlan.outputOrdering` (operators.scala:399)
makes `CometSortAggregateExec` advertise grouping keys ASC. Native
`AggregateExec` only preserves that when DataFusion infers
`InputOrderMode::Sorted` from the **native** child's equivalence properties: DF
55 `physical-plan/src/aggregates/mod.rs` derives `input_order_mode` in
`try_new_with_schema`, and `compute_properties` does `if *input_order_mode ==
InputOrderMode::Linear { eq_properties.clear_orderings() }`. Comet's native
leaf advertises nothing: `native/core/src/execution/operators/scan.rs:78`
builds `EquivalenceProperties::new(schema)`.
So the claim only holds when the pre-aggregate `SortExec` lives in the same
native block. When the ordering comes from outside it, the aggregate runs
Linear/hash mode and emits in group-discovery order while still telling Spark
it is sorted.
Concrete shape I could not rule out: a bucketed + sorted source table.
`FileSourceScanExec.outputOrdering` is non-empty there, so `EnsureRequirements`
drops the sort under the Partial *and* under the Final, and also drops the sort
a downstream `SortMergeJoinExec` / `WindowExec` on the same keys would need.
The native plan becomes `AggregateExec(Final, AggregateExec(Partial,
ScanExec))` with no ordering anywhere, and the join silently loses rows.
Because Comet's rule runs after `EnsureRequirements`, lowering `outputOrdering`
to `Nil` here would not help — the sort is already gone.
Suggestion: only convert when the ordering is produced inside the native
plan (e.g. require the converted child to be a `CometNativeExec` whose
`outputOrdering` satisfies `op.requiredChildOrdering.head`), or teach the
native side about the sortedness Spark already guarantees. A test with a
bucketed+sorted input feeding an SMJ on the grouping key would pin it either
way.
I have not executed this, so please sanity-check the premise before acting
on it.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/first_last.sql:
##########
@@ -70,11 +70,16 @@ CREATE TABLE test_types(
grp string
) USING parquet
+-- first/last IGNORE NULLS are non-deterministic when a group has more than
one non-null value,
Review Comment:
Worth surfacing in the PR description as a compatibility note, not only as a
fixture change.
Before this PR these two queries fell back, so they returned exactly Spark's
answer. They now run natively and, as you found, can disagree with Spark
whenever a group has more than one non-null value. Spark documents
`First`/`Last` as non-deterministic, so this is defensible — but it is a
user-visible change in agreement-with-Spark for real queries, not just for this
test.
`test_types` is used only by the two multi-type queries (lines 200 and 346),
so the coverage loss is contained, and `test_ignore_nulls` /
`test_null_positions` / `test_decimal` / `test_date` still distinguish `first`
from `last` elsewhere in the file. The one case that goes untested is exactly
the interesting one: a **string** buffer (the column that forces
`SortAggregateExec`) with two non-null values in a group. Consider noting the
divergence in the compatibility docs alongside the fixture change.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2103,17 +2165,40 @@ case class CometHashAggregateExec(
override def hashCode(): Int =
Objects.hashCode(output, groupingExpressions, aggregateExpressions, input,
modes, child)
+}
- override lazy val metrics: Map[String, SQLMetric] = {
- val baseline = CometMetricNode.baselineMetrics(sparkContext)
- if (groupingExpressions.nonEmpty) {
- baseline ++ CometMetricNode.aggregateMetrics(sparkContext)
- } else {
- baseline
+case class CometSortAggregateExec(
+ 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)
+ extends CometBaseAggregateExec {
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ this.copy(child = newChild)
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometSortAggregateExec =>
Review Comment:
`equals`/`hashCode` here are character-identical to
`CometHashAggregateExec`'s apart from the type test. `CometBaseAggregateExec`
now exists, so both can live there:
```scala
override def equals(obj: Any): Boolean = obj match {
case other: CometBaseAggregateExec if other.getClass == getClass =>
output == other.output && groupingExpressions ==
other.groupingExpressions &&
aggregateExpressions == other.aggregateExpressions && input ==
other.input &&
modes == other.modes && child == other.child &&
serializedPlanOpt == other.serializedPlanOpt
case _ => false
}
override def hashCode(): Int =
Objects.hashCode(output, groupingExpressions, aggregateExpressions, input,
modes, child)
```
That drops ~28 duplicated lines and leaves both wrappers with only
`withNewChildInternal`.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/sort_aggregate.sql:
##########
@@ -0,0 +1,214 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Disabling ObjectHashAggregate forces Spark to plan SortAggregateExec for the
Review Comment:
**Most of this file already exists as `collect_set.sql`.** `sa_int`,
`sa_nulls`, `sa_empty`, `sa_single`, `sa_string`, `sa_decimal` and `sa_date`
carry the same rows as the corresponding `cs_src_*` tables, and the global /
GROUP BY / all-NULL / empty table / single row / mixed-aggregate /
multiple-`collect_set` / DISTINCT / HAVING / double-`expect_fallback` queries
are the same shapes. What is genuinely new: multiple grouping keys, an
expression grouping key, and `min`/`max`/`count(i)` in the mixed query.
`SqlFileTestParser` accepts several `ConfigMatrix` keys and
`CometSqlFileTestSuite.configMatrix` takes the cartesian product, so adding
```
-- ConfigMatrix: spark.sql.execution.useObjectHashAggregateExec=false,true
```
to `collect_set.sql` would run the whole existing fixture through both
`ObjectHashAggregateExec` and `SortAggregateExec`, and this file could shrink
to the three shapes above. Trade-off: it multiplies with the existing
`parquet.enable.dictionary` matrix, so `collect_set.sql` would run 4x.
Separately: nothing here or in the Scala tests covers the new Comet-shuffle
gate. A tiny fixture with `-- Config: spark.comet.exec.shuffle.enabled=false`
and `expect_fallback` would close that.
##########
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
Review Comment:
With `baseAggregateSupportLevel` lifted, the three serde objects now differ
only in `createExec`: `enabledConfig` is repeated 3x, `getSupportLevel(op) =
baseAggregateSupportLevel(op)` 3x, and this `convert` is identical to
`CometObjectHashAggregateExec`'s apart from the comment.
Parameterising the trait — `trait CometBaseAggregate[T <: BaseAggregateExec]
extends CometOperatorSerde[T]`, supplying `enabledConfig`, `getSupportLevel`
and `convert` behind a `requiresCometShuffle` hook — would leave each object
with just `createExec`. `baseAggregateSupportLevel` is already a half-step in
that direction, so this is finishing the same move rather than adding a new
abstraction.
Nit: `baseAggregateSupportLevel` keys off
`COMET_ENABLE_PARTIAL_HASH_AGGREGATE` / `COMET_ENABLE_FINAL_HASH_AGGREGATE`,
which now gate sort aggregates too. Worth a line in its scaladoc.
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -2408,6 +2409,46 @@ class CometAggregateSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ // useObjectHashAggregateExec=false forces Spark to plan SortAggregateExec
for
+ // TypedImperativeAggregate functions like collect_set. Comet converts those
just like
+ // ObjectHashAggregateExec via the shared CometBaseAggregate path. Broader
data-type and
+ // edge-case coverage lives in the SQL file test
+ //
spark/src/test/resources/sql-tests/expressions/aggregate/sort_aggregate.sql;
these Scala
+ // tests additionally assert that Spark actually planned a
SortAggregateExec, which the SQL
+ // framework cannot check.
+ private def assertSortAggregateRunsNatively(query: String): Unit = {
+ withSQLConf(
+ SQLConf.USE_OBJECT_HASH_AGG.key -> "false",
+ CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ withTempView("tbl") {
+ Seq((1, "a"), (2, "a"), (1, "a"), (3, "b"), (4, "b"), (4, "b"))
+ .toDF("v", "g")
+ .createOrReplaceTempView("tbl")
+ // Spark must actually plan a SortAggregateExec for this query;
otherwise the test
+ // would pass without exercising the new code path.
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val plan = stripAQEPlan(sql(query).queryExecution.executedPlan)
+ assert(
+ plan.find(_.isInstanceOf[SortAggregateExec]).isDefined,
+ s"Expected SortAggregateExec in Spark-only plan but got:\n$plan")
+ }
+ checkSparkAnswerAndOperator(sql(query))
+ }
+ }
+ }
+
+ test("SortAggregate with collect_set is converted to native") {
Review Comment:
These two differ only by query string and both go through
`assertSortAggregateRunsNatively`. One test iterating over the two queries
would read the same and keep the count honest:
```scala
test("SortAggregate with collect_set is converted to native") {
Seq(
"SELECT g, sort_array(collect_set(v)) FROM tbl GROUP BY g ORDER BY g",
// empty grouping is a distinct shape: no pre-aggregate sort, empty
output
// ordering, adjustOutputForNativeState with zero grouping columns
"SELECT sort_array(collect_set(v)) FROM
tbl").foreach(assertSortAggregateRunsNatively)
}
```
--
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]
