andygrove commented on code in PR #5867:
URL: https://github.com/apache/datafusion-comet/pull/5867#discussion_r4065477544
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -51,7 +51,39 @@ object CometArrayRemove
}
}
-object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with
ArraysBase {
+/**
+ * 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 =
+ "a nondeterministic operand: the native NULL guard serializes the operand
twice, " +
+ "and the two copies of a stateful operand drift apart"
+
+ /** `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)
+
+ // 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.
+ override def getSupportLevel(expr: ArrayAppend): SupportLevel =
Review Comment:
I filed #6086 for the ANSI gap in your "What this does not cover" section,
and confirmed it on `main` at 5ca149928. With `spark.sql.ansi.enabled=true`,
`array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 1 / (_1 -
1))` raises `DIVIDE_BY_ZERO` in Spark and returns cleanly in Comet. A control
query in the same fixture confirms Comet's ANSI divide does raise on its own,
so the test is not vacuous.
Could you link that issue from this comment? `getSupportLevel` reports
`Compatible()` for the case, so without it nothing in the code or in the
generated compatibility guide records the divergence.
While I was there I checked whether `CometMapFromArrays` has the
mirror-image problem, since its guard is `IsNotNull(left) AND IsNotNull(right)`
while Spark's `nullSafeCodeGen` short-circuits the values array. It does not
reproduce. That one runs natively and matches Spark. The negative result is
written up in the issue so nobody re-derives it.
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -843,10 +883,12 @@ object CometArrayPosition extends
CometExpressionSerde[ArrayPosition] with Array
}
}
-object CometArraysZip extends CometExpressionSerde[ArraysZip] {
+object CometArraysZip extends CometExpressionSerde[ArraysZip] with
CodegenDispatchFallback {
override def getUnsupportedReasons(): Seq[String] = Seq(
- "Not all input data types are supported; falls back to Spark for
unsupported types")
+ "Not all input data types are supported; unsupported types run through the
JVM codegen " +
Review Comment:
`GenerateDocs` already prints "The following cases have no native
implementation and always run in the JVM using Spark's code-generated
implementation (inside the Comet pipeline)" above these bullets whenever the
serde mixes in `CodegenDispatchFallback`. So "unsupported types run through the
JVM codegen dispatcher" now repeats its own header, and the bullet still never
says which types. Could it name them instead, something like map, interval and
variant element types?
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -51,7 +51,39 @@ object CometArrayRemove
}
}
-object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with
ArraysBase {
+/**
+ * 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 =
Review Comment:
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 as `rand()`
or `monotonically_increasing_id()`, because the native NULL guard would
evaluate it twice" sit better there?
##########
spark/src/test/resources/sql-tests/expressions/array/arrays_zip_nondeterministic_child.sql:
##########
@@ -0,0 +1,51 @@
+-- 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.
+
+-- `CometArraysZip` reproduces Spark's NULL propagation with a `CASE WHEN`
guard over every
+-- child's `IS NOT NULL`, which serializes each child twice. A stateful child
advances each copy
+-- independently, so the guard and the `arrays_zip` see different rows and the
result silently
+-- drifts from Spark. The serde declines a nondeterministic child and routes
it through
+-- the JVM codegen dispatcher, which evaluates it once. A deterministic
nullable child keeps the
+-- native guard.
+
+statement
+CREATE TABLE test_arrays_zip_nondet(_1 int) USING parquet
+
+statement
+INSERT INTO test_arrays_zip_nondet VALUES
+ (0), (1), (2), (3), (4), (5), (6), (7), (8), (9), (10), (11), (12), (13),
(14), (15)
+
+-- Spark returns [{1, 2}] on every row whose first array is non-NULL and NULL
on the rest.
+query expect_dispatch(arrays_zip)
+SELECT _1, arrays_zip(IF(monotonically_increasing_id() % 2 = 0, array(1),
CAST(NULL AS ARRAY<INT>)), array(2)) AS z
+FROM test_arrays_zip_nondet
+
+-- The guard covers every child, so a stateful second child is declined the
same way.
+query expect_dispatch(arrays_zip)
+SELECT _1, arrays_zip(array(2), IF(monotonically_increasing_id() % 2 = 0,
array(1), CAST(NULL AS ARRAY<INT>))) AS z
+FROM test_arrays_zip_nondet
+
+-- A deterministic nullable child stays on the native guarded path.
Review Comment:
"A deterministic nullable child stays on the native guarded path" is sitting
above the `expect_dispatch` query for a non-nullable stateful child, and the
`expect_native` query below it has no comment at all. Looks like the two got
swapped.
Separately, `size` and `map_from_arrays` picked up column-fed dispatch cases
in this round but `arrays_zip` did not, and it is the one whose dispatched
kernel copies `ListVector`s into an `array<struct<>>`. Would you mind adding
one query here over an `array<int>` column?
--
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]
