tlfua commented on code in PR #7457:
URL: https://github.com/apache/datafusion-comet/pull/7457#discussion_r4462332554
##########
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:
Thanks. I checked each plain-sort shape against Spark with ANSI on and filed
https://github.com/apache/datafusion-comet/issues/7470 .
Global `ORDER BY a, b + 1` and window ordering (`rank`, `row_number` with
`PARTITION BY`) throw in Spark too, because the range exchange or the window
pre-projection evaluates every key. The divergence shows only when nothing
beneath the sort evaluates all keys first, for example `SORT BY` or a sort over
a single-partition child such as `LIMIT`. I extracted the check into
`QueryPlanSerde.laterSortKeyThatCanFail(Seq[SortOrder])` so the plain sort can
reuse it in that follow-up.
##########
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:
Widened the list after reading the Spark source for 3.4.3, 3.5.8, 4.0.1 and
4.1.1. Added:
- String: `octet_length`, `bit_length`, `trim`/`ltrim`/`rtrim`, `replace`,
`initcap`, `concat_ws`, `concat` (string, at least one argument), `reverse`
(string)
- Datetime: `year`, `quarter`, `month`, `day`, `dayofweek`, `weekday`,
`dayofyear`, `hour`, `minute`, `second`
- Math: `floor`, `ceil`, `sqrt`, `exp`, `log`, `signum`, `greatest`,
`least`, `isnan`, `nanvl`, `IN`; `round`/`bround` for float/double always, and
for integral types only when ANSI is off
- Hash: `hash`, `xxhash64`, `md5`, `sha1`, `sha2`, `crc32`
- Cast: string to binary (inserted implicitly by `md5(s)` and friends)
I also checked each one's native implementation (the Comet kernels plus
DataFusion / datafusion-spark 55.2) for errors that depend on the data. The
only real ones were timestamp casts: they go through chrono, which rejects
years Spark accepts, and raise even for TRY (filed as
https://github.com/apache/datafusion-comet/issues/7471 ). So any cast to or
from `TIMESTAMP`/`TIMESTAMP_NTZ` now counts as able to fail. The rest error
only at planning time or beyond 2GB of string offsets per batch.
Left out on purpose: timestamp casts (above), decimal `round` (the
negative-scale path builds a `Decimal` that can fail the precision check),
integral `round` under ANSI (overflow), `lpad`/`rpad`/`repeat` (size limits),
`element_at`, `date_add`, `to_date`, and anything that parses input.
On upkeep: the list fails safe. A function missing from it only falls back
to Spark, so leaving one out costs performance, never correctness. Each entry
was checked against Spark 3.4.3 to 4.1.1 and the native kernel, and
`topk_later_key_ansi.sql` runs each one natively over NULL and boundary values.
##########
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:
Good catch. I reworded the comment: the list now states that it answers from
Spark semantics alone and is independent of the native `cannot_fail` in
`case_when.rs`. I left `cannotThrow` in `arrays.scala` as is, because switching
it to this list would change when `array_append` runs natively; I'd rather do
that in a separate PR with its own tests.
##########
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:
Added:
- Fallback: a fallible key nested under an allowed one (`coalesce(b + 1,
0)`, `IF(a > 3, b + 1, 0)`), and `round(b, -1)` under ANSI.
- Native: `try_cast`, and a `topk_types` table with NULL, NaN, -0.0,
int/long bounds, empty strings, and dates such as 1970-01-01, 2000-02-29 and
9999-12-31, run through the string, datetime, math and hash keys.
- ANSI off, running natively: a string-to-int cast in LEGACY mode, `round(b,
-1)`, `b / 0`, `b % 0`, `b DIV 0`, `l + 1` and `l * 2` as later keys.
##########
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:
Reworded. The section now says the check does not depend on ANSI: Comet
falls back whenever it cannot show that a later key never fails. It also notes
that a sort without a LIMIT and with no exchange beneath it, such as `SORT BY`,
still runs natively and can raise, and links
https://github.com/apache/datafusion-comet/issues/7470 .
--
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]
