Repository navigation
fix: fall back TopK whose later sort key can fail under ANSI #7457
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions._ | |
| import org.apache.spark.sql.catalyst.expressions.aggregate._ | ||
| import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke} | ||
| import org.apache.spark.sql.catalyst.expressions.xml.{XPathBoolean, XPathDouble, XPathFloat, XPathInt, XPathList, XPathLong, XPathShort, XPathString} | ||
| import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero | ||
| import org.apache.spark.sql.comet.DecimalPrecision | ||
| import org.apache.spark.sql.execution.{ScalarSubquery, SparkPlan} | ||
| import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils | ||
|
|
@@ -1394,6 +1395,72 @@ 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. It answers from Spark's semantics, independently of the native | ||
| * `cannot_fail` in `case_when.rs`, which judges the physical expressions it is given. It is | ||
| * conservative: an allowed expression whose children also cannot fail. Anything not listed is | ||
| * treated as able to fail. | ||
| */ | ||
|
Comment on lines
+1398
to
+1404
Contributor
There was a problem hiding this comment. The comment says this works like the native
Author
There was a problem hiding this comment. Good catch. I reworded the comment: the list now states that it answers from Spark semantics alone and is independent of the native
Author
|
||
| def cannotFail(expr: Expression): Boolean = { | ||
| def numeric: Boolean = expr.dataType.isInstanceOf[NumericType] | ||
| def nonAnsiNumeric(evalMode: EvalMode.Value): Boolean = evalMode != EvalMode.ANSI && numeric | ||
| // Floating-point arithmetic never overflows into an error, even in ANSI mode | ||
| def floating: Boolean = expr.dataType == FloatType || expr.dataType == DoubleType | ||
| val nodeCannotFail = expr match { | ||
| case _: Attribute | _: BoundReference | _: Literal => true | ||
| case _: GetStructField => true | ||
| case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true | ||
| case _: In | _: InSet | _: Greatest | _: Least => true | ||
| case _: If | _: CaseWhen | _: Coalesce | _: IsNaN | _: NaNvl => true | ||
| case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true | ||
| case _: Substring | _: Upper | _: Lower | _: Length | _: OctetLength | _: BitLength => true | ||
| case _: StringTrim | _: StringTrimLeft | _: StringTrimRight => true | ||
| case _: StringReplace | _: InitCap => true | ||
| // These fail only when the result would exceed 2GB, more than a row can hold | ||
| case _: ConcatWs => true | ||
| case c: Concat => c.children.nonEmpty && c.dataType == StringType | ||
| case _: Reverse => expr.dataType == StringType | ||
| case _: Year | _: Quarter | _: Month | _: DayOfMonth | _: DayOfWeek | _: WeekDay | | ||
| _: DayOfYear | _: Hour | _: Minute | _: Second => | ||
| true | ||
| case _: Murmur3Hash | _: XxHash64 | _: Md5 | _: Sha1 | _: Sha2 | _: Crc32 => true | ||
| // Invalid input gives NaN or NULL, and a floating-point floor or ceil saturates to a long | ||
| case _: Floor | _: Ceil | _: Sqrt | _: Exp | _: Log | _: Signum => true | ||
| // Rounding an integer fails on overflow in ANSI mode; a decimal is never treated as safe | ||
| case r: Round => roundCannotFail(r.dataType, r.ansiEnabled) | ||
| case r: BRound => roundCannotFail(r.dataType, r.ansiEnabled) | ||
| case c: Cast => CometCast.cannotFail(c) | ||
| case _: UnaryPositive => true | ||
| case u: UnaryMinus => floating || (!u.failOnError && numeric) | ||
| case a: Abs => floating || (!a.failOnError && numeric) | ||
| case a: Add => floating || nonAnsiNumeric(a.evalMode) | ||
| case s: Subtract => floating || nonAnsiNumeric(s.evalMode) | ||
| case m: Multiply => floating || 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 | ||
|
Comment on lines
+1411
to
+1444
Contributor
There was a problem hiding this comment. This set decides the fallback for every top-K with more than one key, with ANSI on or off. Later keys such as
Author
There was a problem hiding this comment. Widened the list after reading the Spark source for 3.4.3, 3.5.8, 4.0.1 and 4.1.1. Added:
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 #7471 ). So any cast to or from Left out on purpose: timestamp casts (above), decimal 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 |
||
| } | ||
| nodeCannotFail && expr.children.forall(cannotFail) | ||
| } | ||
|
|
||
| private def roundCannotFail(dataType: DataType, ansiEnabled: Boolean): Boolean = | ||
| dataType match { | ||
| case FloatType | DoubleType => true | ||
| case ByteType | ShortType | IntegerType | LongType => !ansiEnabled | ||
| case _ => false | ||
| } | ||
|
|
||
| /** | ||
| * The first sort key after the first that can fail. 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 such a key may raise on a row that Spark skips (#7246). | ||
| */ | ||
| def laterSortKeyThatCanFail(sortOrder: Seq[SortOrder]): Option[SortOrder] = | ||
| sortOrder.drop(1).find(s => !cannotFail(s.child)) | ||
|
|
||
| /** | ||
| * Has native execution evaluate the arguments of `scalarFunc`, the serialized `expr`, the way | ||
| * Spark evaluates those of a null-intolerant `BinaryExpression` or `TernaryExpression`: each | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
try_cast(ts_array AS STRING)bypasses this timestamp check. WithLong.MaxValuemicroseconds, the native cast errors while Spark formats it. Check nested timestamp types too, with a TopK regression.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks, good catch. Fixed in 5a5b423.
CometCast.cannotFailnow looks inside arrays, structs and maps: a timestamp anywhere in the source or target type makes the cast able to fail, and so does a date in the source combined with a string in the target. A nested cast applies the same chrono-based kernels to each element, field, key and value, so they raise under TRY just like the top-level cast ( #7471 ).topk_later_key_ansi.sqladds atopk_nestedtable withtimestamp_micros(9000000000000000000)anddate_from_unix_date(100000000), and checks that these later keys fall back:try_castofarray<timestamp>andarray<date>toSTRINGand toarray<string>,struct<timestamp>andstruct<date>toSTRING,struct<date>tostruct<string>andstruct<timestamp>,struct<timestamp>tostruct<date>,map<int, date>tomap<int, string>,map<timestamp, int>tomap<string, int>, andarray<struct<timestamp>>toSTRING.