fix: dispatch map lookups with normalized keys and nondeterministic null-guarded children #5867
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,7 +51,63 @@ object CometArrayRemove | |
| } | ||
| } | ||
|
|
||
| /** | ||
| * Shared gate for serdes whose native NULL guard (`CASE WHEN child IS NOT NULL`) serializes the | ||
| * child twice: a stateful child drifts between the two copies, so it is declined and runs through | ||
| * the JVM codegen dispatcher, where Spark evaluates it once. Nullability is not consulted: a | ||
| * non-nullable stateful child only stays in step because DataFusion skips the filter when the | ||
| * guard matches every row, which is not a contract to lean on. | ||
| */ | ||
| private[serde] object NullGuardSupport { | ||
|
|
||
| val nondeterministicReason: String = | ||
| "Comet has no native path for a nondeterministic operand such as `rand()` or " + | ||
| "`monotonically_increasing_id()`, because the native `NULL` guard would evaluate it twice." | ||
|
|
||
| /** `Unsupported` when any of `children` is nondeterministic, otherwise `None`. */ | ||
| def nondeterministicChild(children: Seq[Expression]): Option[SupportLevel] = | ||
| children | ||
| .find(child => !child.deterministic) | ||
| .map(_ => Unsupported(Some(nondeterministicReason))) | ||
| } | ||
|
|
||
| object CometArrayAppend | ||
| extends CometExpressionSerde[ArrayAppend] | ||
| with ArraysBase | ||
| with CodegenDispatchFallback { | ||
|
|
||
| override def getUnsupportedReasons(): Seq[String] = | ||
| Seq(NullGuardSupport.nondeterministicReason) | ||
|
|
||
| private val ansiItemNote: String = | ||
| "With `spark.sql.ansi.enabled=true` and a nullable array, the native `NULL` guard skips " + | ||
| "the item on a row whose array is `NULL`, so an item that raises there (for example a " + | ||
| "division by zero) raises in Spark but not on the native path. Such an expression runs " + | ||
| "through the JVM codegen dispatcher by default; enabling the native path can swallow " + | ||
| "that error ([#6086](https://github.com/apache/datafusion-comet/issues/6086))." | ||
|
|
||
| override def getIncompatibleReasons(): Seq[String] = Seq(ansiItemNote) | ||
|
|
||
| // Only the ANSI nullable-array case is routed through the dispatcher; every other compatible | ||
| // instance runs natively by default. | ||
| override def hasConditionalNativeDefault: Boolean = true | ||
|
|
||
| // The item sits inside the guard's THEN branch, and DataFusion's CaseExpr evaluates that | ||
| // branch only on the rows the guard selects, while Spark's codegen evaluates the item on | ||
| // every row. A stateful item therefore drifts the same way a stateful array does, and is | ||
| // declined. The same shape lets an item that raises under ANSI mode go unevaluated on a row | ||
| // whose array is NULL, so Spark raises where the native path returns NULL. That case is | ||
| // reported as incompatible, which routes it through the JVM codegen dispatcher by default | ||
| // and reserves the native guard for allowIncompatible=true. A non-nullable array evaluates | ||
| // the item on every row on both paths, so it stays native. | ||
| override def getSupportLevel(expr: ArrayAppend): SupportLevel = | ||
|
Member
There was a problem hiding this comment. I filed #6086 for the ANSI gap in your "What this does not cover" section, and confirmed it on Could you link that issue from this comment? While I was there I checked whether
Contributor
Author
There was a problem hiding this comment.
Linked in 0d61cac, and the support level changed with it. With ANSI on and a nullable array, |
||
| NullGuardSupport.nondeterministicChild(expr.children).getOrElse { | ||
| if (SQLConf.get.ansiEnabled && expr.left.nullable) { | ||
| Incompatible(Some(ansiItemNote)) | ||
| } else { | ||
| Compatible() | ||
| } | ||
| } | ||
|
|
||
| override def convert( | ||
| expr: ArrayAppend, | ||
|
|
@@ -614,7 +670,7 @@ object CometArrayReverse extends CometExpressionSerde[Reverse] with ArraysBase { | |
|
|
||
| } | ||
|
|
||
| object CometElementAt extends CometExpressionSerde[ElementAt] with CodegenDispatchFallback { | ||
|
|
||
| /** | ||
| * Under ANSI, neither native shape reproduces Spark for a nullable nondeterministic operand. | ||
|
|
@@ -623,8 +679,9 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { | |
| * whole batch first, so a throwing index fires on rows whose operand is NULL. `convert` | ||
| * reproduces the short-circuit with a `CASE WHEN <operand> IS NOT NULL` guard, but that guard | ||
| * serializes the operand twice, which a stateful operand cannot survive: the two copies advance | ||
| * its state independently and silently move values and NULLs. Declining routes the lookup | ||
| * through the JVM codegen dispatcher, where Spark's own `doGenCode` evaluates the operand once. | ||
| * Lifting this needs a native lookup that evaluates the operand once and masks the index | ||
| * evaluation with the result, at which point the guard becomes unnecessary for every operand. | ||
| */ | ||
| private val eagerIndexReason: String = | ||
|
|
@@ -635,6 +692,8 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { | |
| private def needsNullGuard(expr: ElementAt): Boolean = | ||
| expr.failOnError && expr.left.nullable | ||
|
|
||
| override def getUnsupportedReasons(): Seq[String] = eagerIndexReason +: MapKeySupport.reasons | ||
|
|
||
| override def getSupportLevel(expr: ElementAt): SupportLevel = { | ||
| if (needsNullGuard(expr) && !expr.left.deterministic) { | ||
| Unsupported(Some(eagerIndexReason)) | ||
|
|
@@ -764,14 +823,19 @@ object CometArrayFilter extends CometExpressionSerde[ArrayFilter] { | |
| } | ||
| } | ||
|
|
||
| object CometSize extends CometExpressionSerde[Size] with CodegenDispatchFallback { | ||
|
|
||
| override def getUnsupportedReasons(): Seq[String] = | ||
| Seq(NullGuardSupport.nondeterministicReason) | ||
|
|
||
| override def getSupportLevel(expr: Size): SupportLevel = { | ||
| NullGuardSupport.nondeterministicChild(Seq(expr.child)).getOrElse { | ||
| expr.child.dataType match { | ||
| case _: ArrayType => Compatible() | ||
| case _: MapType => Compatible() | ||
| case other => | ||
| Unsupported(Some(s"Unsupported child data type: $other")) | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -843,10 +907,13 @@ object CometArrayPosition extends CometExpressionSerde[ArrayPosition] with Array | |
| } | ||
| } | ||
|
|
||
| object CometArraysZip extends CometExpressionSerde[ArraysZip] with CodegenDispatchFallback { | ||
|
|
||
| override def getUnsupportedReasons(): Seq[String] = Seq( | ||
| "An array whose element type is a map, a calendar, day-time or year-month interval, a " + | ||
| "variant, a `TIME` value or a user-defined type has no native `arrays_zip` kernel, and " + | ||
| "neither does a struct or inner array that holds one of those.", | ||
| NullGuardSupport.nondeterministicReason) | ||
|
|
||
| private def isTypeSupported(dt: DataType): Boolean = { | ||
| import DataTypes._ | ||
|
|
@@ -862,13 +929,13 @@ object CometArraysZip extends CometExpressionSerde[ArraysZip] { | |
| } | ||
|
|
||
| override def getSupportLevel(expr: ArraysZip): SupportLevel = { | ||
| NullGuardSupport.nondeterministicChild(expr.children).getOrElse { | ||
| expr.children | ||
| .map(_.dataType) | ||
| .collectFirst { case dt if !isTypeSupported(dt) => dt } | ||
| .map(dt => Unsupported(Some(s"Unsupported child data type: $dt"))) | ||
| .getOrElse(Compatible()) | ||
| } | ||
| } | ||
|
|
||
| override def convert( | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| -- 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. | ||
|
|
||
| -- `CometArrayAppend` reproduces Spark's NULL propagation with a `CASE WHEN array IS NOT NULL` | ||
| -- guard, and the item sits inside the guard's THEN branch. DataFusion evaluates that branch only | ||
| -- on the rows the guard selects, while Spark's codegen evaluates the item on every row. Under | ||
| -- ANSI mode an item that raises on a row whose array is NULL therefore raises in Spark and stays | ||
| -- silent on the native path. The serde reports a nullable array under ANSI as incompatible, so | ||
| -- by default the expression runs through the JVM codegen dispatcher, which raises like Spark. | ||
| -- A non-nullable array evaluates the item on every row on both paths, so it stays native. | ||
| -- | ||
| -- Spark 4.0 rewrites `array_append` to `array_insert(-1)` before serde, so `CometArrayAppend` is | ||
| -- only reachable on Spark 3.x. | ||
|
|
||
| -- MaxSparkVersion: 3.5 | ||
|
|
||
| -- Config: spark.sql.ansi.enabled=true | ||
|
|
||
| statement | ||
| CREATE TABLE test_array_append_ansi(_1 int, arr array<int>) USING parquet | ||
|
|
||
| statement | ||
| INSERT INTO test_array_append_ansi | ||
| SELECT id, IF(id = 1, NULL, array(id)) FROM range(0, 4) | ||
|
|
||
| -- The item divides by zero exactly on the row whose array is NULL, so Spark raises and the | ||
| -- dispatcher raises with it. | ||
| query expect_error(DIVIDE_BY_ZERO) | ||
| SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a | ||
| FROM test_array_append_ansi | ||
|
|
||
| -- The divide raises on its own, so the error above is not an artifact of the fixture. | ||
| query expect_error(DIVIDE_BY_ZERO) | ||
| SELECT _1, 1 / (_1 - 1) AS d | ||
| FROM test_array_append_ansi | ||
|
|
||
| -- A nullable array under ANSI is routed through the dispatcher even when the item cannot raise. | ||
| query expect_dispatch(array_append) | ||
| SELECT _1, array_append(arr, _1) AS a | ||
| FROM test_array_append_ansi | ||
|
|
||
| -- A non-nullable array literal cannot hit the gap and keeps the native guarded path. | ||
| query expect_native(array_append) | ||
| SELECT _1, array_append(array(1), _1) AS a | ||
| FROM test_array_append_ansi |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
This renders in the generated compatibility guide right next to
MapKeySupport's reasons, which are full sentences about what Spark does and what the native path does instead. This one is a lowercase fragment about how the serde is built, which reads oddly as a user-facing bullet. Would something like "Comet has no native path for a nondeterministic operand such asrand()ormonotonically_increasing_id(), because the native NULL guard would evaluate it twice" sit better there?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.
Yes. That is the reason now, word for word apart from the code formatting, and since the string is shared it changes the bullet for all four serdes in 0d61cac.