comphead commented on code in PR #7457:
URL: https://github.com/apache/datafusion-comet/pull/7457#discussion_r4460105716
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -1394,6 +1395,38 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
}
}
+ /**
+ * Whether `expr` cannot raise an error for any row, so native execution may
evaluate it for
+ * rows that Spark skips. Like the native `cannot_fail`, it is conservative:
columns, literals,
+ * and comparisons, boolean logic, null checks, conditionals, substrings,
case conversion and
+ * length, numeric arithmetic, negation and absolute value outside ANSI mode
and casts that
+ * cannot fail over them. Anything else is treated as able to fail.
+ */
+ def cannotFail(expr: Expression): Boolean = {
+ def numeric: Boolean = expr.dataType.isInstanceOf[NumericType]
+ def nonAnsiNumeric(evalMode: EvalMode.Value): Boolean = evalMode !=
EvalMode.ANSI && numeric
+ val nodeCannotFail = expr match {
+ case _: Attribute | _: BoundReference | _: Literal => true
+ case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _:
IsNotNull => true
+ case _: If | _: CaseWhen | _: Coalesce => true
+ case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true
+ case _: Substring | _: Upper | _: Lower | _: Length => true
+ case c: Cast => CometCast.cannotFail(c)
+ case _: UnaryPositive => true
+ case u: UnaryMinus => !u.failOnError && numeric
+ case a: Abs => !a.failOnError && numeric
+ case a: Add => nonAnsiNumeric(a.evalMode)
+ case s: Subtract => nonAnsiNumeric(s.evalMode)
+ case m: Multiply => nonAnsiNumeric(m.evalMode)
+ case d: Divide => nonAnsiNumeric(d.evalMode)
+ case d: IntegralDivide => nonAnsiNumeric(d.evalMode)
+ case r: Remainder => nonAnsiNumeric(r.evalMode)
+ case p: Pmod => nonAnsiNumeric(p.evalMode)
+ case _ => false
Review Comment:
This set decides the fallback for every top-K with more than one key, with
ANSI on or off. Later keys such as `year(d)`, `trim(s)`, `round(x, 2)` or
`concat` fall back even though they cannot fail, and `substring`, `upper`,
`lower` and `length` came in to fix the q79 plan change. Is that narrow set
intended? If not, the common infallible functions could be added with tests.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -1394,6 +1395,38 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
}
}
+ /**
+ * Whether `expr` cannot raise an error for any row, so native execution may
evaluate it for
+ * rows that Spark skips. Like the native `cannot_fail`, it is conservative:
columns, literals,
+ * and comparisons, boolean logic, null checks, conditionals, substrings,
case conversion and
+ * length, numeric arithmetic, negation and absolute value outside ANSI mode
and casts that
+ * cannot fail over them. Anything else is treated as able to fail.
+ */
Review Comment:
The comment says this works like the native `cannot_fail`, but the two sets
differ. The native one in `conditional_funcs/case_when.rs` (line 449) admits no
functions and no integer or decimal `Divide`, while this one admits
`Substring`, `Upper`, `Lower`, `Length`, `Coalesce`, all non-ANSI arithmetic
and every LEGACY atomic cast. The difference may be right, but the wording
should say they are independent. Also, `arrays.scala` already has a private
`cannotThrow` (line 86) answering the same question for `array_append`. Could
it call this helper, so there is one Scala notion of an expression that cannot
fail?
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala:
##########
@@ -63,6 +63,13 @@ object CometTakeOrderedAndProjectExec extends
CometSink[TakeOrderedAndProjectExe
if (!supportedSortType(op, op.sortOrder)) {
return Unsupported(Some("unsupported data type in sort order"))
}
+ // Spark's ordering evaluates a sort key only for the rows whose earlier
keys tie, but native
+ // execution evaluates every key over the whole batch, so a later key that
can fail may raise
+ // on a row that Spark skips (#7246).
+ op.sortOrder.drop(1).find(s => !cannotFail(s.child)).foreach { s =>
+ return Incompatible(
+ Some(s"sort key ${s.child.sql} can fail on rows that Spark's ordering
skips"))
+ }
Review Comment:
The description leaves the plain sort for a follow-up. Reading Spark, it has
the same difference: `SortExec.createSorter` builds its comparator with
`RowOrdering.create`, which goes through `GenerateOrdering`, so only the first
key is computed for every row, as the prefix. `CometSortExec` has no guard, so
I expect `ORDER BY a, b + 1` without a `LIMIT` to raise in Comet on the fixture
data. I have not run it. Window ordering goes the same way. Since a plain sort
is far more common than a top-K, could you file an issue for it and link it
here and in the docs? If the guard will be reused there, a helper over
`Seq[SortOrder]` would save copying `drop(1).find` into each operator.
##########
spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql:
##########
@@ -0,0 +1,80 @@
+-- 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.
+
+-- Spark's top-K ordering evaluates a later sort key only for rows whose
earlier keys tie, so a
+-- later key that fails under ANSI mode does not fail on a row Spark skips.
Comet evaluates every
+-- key over the whole batch, so a top-K whose later key can fail runs on Spark
(#7246).
+-- Each table is one batch.
+
+-- Config: spark.sql.ansi.enabled=true
+-- Config: spark.sql.adaptive.enabled=false
+
+statement
+CREATE TABLE topk(a int, b int) USING parquet
+
+statement
+INSERT INTO topk SELECT /*+ REPARTITION(1) */ * FROM VALUES
+ (1, 2147483647), (2, 5), (3, 2147483647), (4, 7), (5, 9) AS t(a, b)
+
+-- No two rows tie on a, so Spark never evaluates b + 1
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, b + 1 LIMIT 2
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a DESC, CAST(CAST(b AS STRING) AS INT) * 2
LIMIT 2
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, abs(b) LIMIT 2
+
+-- Later keys that cannot fail stay native
+query
+SELECT a, b FROM topk ORDER BY a, b LIMIT 2
+
+query
+SELECT a, b FROM topk ORDER BY a, b IS NULL, CAST(b AS BIGINT), try_add(b, 1)
LIMIT 2
+
+query
+SELECT a, b FROM topk ORDER BY a, substr(CAST(b AS STRING), 1, 3),
upper(CAST(b AS STRING)),
+ length(lower(CAST(b AS STRING))) LIMIT 2
Review Comment:
The native cases here cover only `int` columns without NULLs. Could you add
a fallible later key nested under an allowed parent (`coalesce(b + 1, 0)`,
`IF(a > 3, b + 1, 0)`), which is the only coverage of
`children.forall(cannotFail)`? Also `try_cast` and a LEGACY cast with ANSI off
for the two `CometCast.cannotFail` branches, later keys over `bigint`,
`decimal`, `double`, `string`, `date` and `timestamp` with NULLs, NaN, `-0.0`
and bound values, and a non-ANSI `/`, `%` and `DIV` by zero.
##########
docs/source/user-guide/latest/compatibility/operators.md:
##########
@@ -94,6 +94,14 @@ TopK, and ranking window order keys run natively under both
default and nondefau
searches boundaries over a nullable nested key still runs on Spark
([#6477](https://github.com/apache/datafusion-comet/issues/6477)).
+Spark's TopK (`TakeOrderedAndProject`, an `ORDER BY` with a `LIMIT`) evaluates
a sort key after
+the first only for the rows whose earlier keys tie. Comet evaluates every key
over the whole batch,
+so under `spark.sql.ansi.enabled=true` a later key such as `b + 1` could raise
+`ARITHMETIC_OVERFLOW` on a row that Spark never evaluates it for. A TopK whose
later sort key can
+fail therefore runs on Spark, unless
+`spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true`
+([#7246](https://github.com/apache/datafusion-comet/issues/7246)).
Review Comment:
The check does not read `spark.sql.ansi.enabled`. With ANSI off, it also
falls back for any later key outside the allowed set. Could the text say that a
later key Comet cannot show to be infallible runs on Spark in either mode, and
that `Sort` and window ordering still run natively with the same difference,
linking the issue?
--
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]
