comphead commented on code in PR #5867:
URL: https://github.com/apache/datafusion-comet/pull/5867#discussion_r4329264331
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -60,7 +60,120 @@ 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 =
+ "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)))
+}
+
+/**
+ * Keeps a result holding a calendar interval, at any nesting depth, off the
JVM codegen
+ * dispatcher. The dispatcher writes the interval's microseconds to Arrow as
nanoseconds, which
+ * overflows a long beyond about 292 years, a range Spark represents
+ * ([#5279](https://github.com/apache/datafusion-comet/issues/5279)). Such an
expression falls
+ * back to Spark instead.
+ */
+private[serde] object CalendarIntervalOutput {
+
+ private val reason: String =
+ "the result holds a calendar interval, whose microseconds overflow the
dispatcher's Arrow " +
+ "nanoseconds beyond about 292 years"
+
+ def declineReason(expr: Expression): Option[String] =
+ if (SupportLevel.containsType(expr.dataType,
classOf[CalendarIntervalType])) Some(reason)
+ else None
+}
+
+/**
+ * Keeps an expression off the JVM codegen dispatcher when one of its nodes
could skip an argument
+ * holding a subexpression that the operator's other expressions share (see
+ * `QueryPlanSerde.skipsSharedSubexpression`). Spark evaluates such a
subexpression for every row,
+ * so the expression falls back to Spark instead.
+ */
+private[serde] object SharedSubexpressionArgument {
+
+ private val reason: String =
+ "an argument this expression can skip holds a subexpression that the
operator's other " +
+ "expressions share, which Spark evaluates for every row before this
expression's NULL check"
+
+ def declineReason(expr: Expression): Option[String] =
+ if (skipsSharedSubexpression(expr)) Some(reason) else None
+}
+
+/**
+ * Whether evaluating `expr` at a different point than Spark does, or not at
all, is unobservable.
+ *
+ * A literal or column read cannot throw, carry state or have a side effect,
so neither its
+ * evaluation order nor a skipped evaluation can be observed; anything else
may raise or drift
+ * where Spark does not. `foldable` is not usable here: ConstantFolding leaves
a throwing foldable
+ * expression unfolded in a conditional branch.
+ */
+private[serde] object EvaluationOrder {
+
+ def isUnobservable(expr: Expression): Boolean = expr match {
+ case _: Literal | _: Attribute | _: BoundReference => true
+ case _ => false
+ }
+}
+
+object CometArrayAppend
+ extends CometExpressionSerde[ArrayAppend]
+ with ArraysBase
+ with CodegenDispatchFallback {
+
+ override def dispatchDeclineReason(expr: Expression): Option[String] =
+ CalendarIntervalOutput
+ .declineReason(expr)
+ .orElse(SharedSubexpressionArgument.declineReason(expr))
Review Comment:
Five serdes carry this same override: `CometArrayAppend` here,
`CometElementAt` (line 950), `CometArraysZip` (line 1179), and
`CometMapExtract` and `CometMapFromArrays` in `maps.scala` (lines 125 and 192).
`CometSize` (line 1088) has it without the `CalendarIntervalOutput` half, which
cannot fire there because `Size` returns an int. Could one shared definition
hold the two gates, for example a small trait next to
`CodegenDispatchFallback`, so a new serde gets them by default instead of by
copy? `getUnsupportedReasons` has the same repetition, where four serdes return
`Seq(NullGuardSupport.nondeterministicReason)`.
##########
spark/src/test/resources/sql-tests/expressions/array/size_shared_subexpression.sql:
##########
@@ -0,0 +1,65 @@
+-- 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 projection evaluates a subexpression that several of its
expressions share for every
+-- row, before the expressions that use it. A `size` over a lookup whose key
holds a
+-- nondeterministic `rand(7)` goes through the dispatcher, whose decline check
walks the whole
+-- tree and finds the lookup skipping the shared `1.0D / x` on the row whose
map is NULL.
+
+-- Config: spark.sql.ansi.enabled=true
+-- Config: spark.sql.subexpressionElimination.enabled=true
+
+statement
+CREATE TABLE test_size_shared(m1 map<double, array<int>>, m2 map<double,
array<int>>, x double)
+USING parquet
+
+statement
+INSERT INTO test_size_shared VALUES
+ (CAST(NULL AS MAP<DOUBLE, ARRAY<INT>>), CAST(NULL AS MAP<DOUBLE,
ARRAY<INT>>), 0.0D),
+ (map(0.5D, array(1, 2)), map(0.5D, array(3)), 2.0D)
+
+-- Without a nondeterministic key `size` runs natively, so this query
exercises the decline of the
+-- dispatched `element_at`.
+query expect_error(DIVIDE_BY_ZERO)
+SELECT size(element_at(m1, 1.0D / x)), size(element_at(m2, 1.0D / x)) FROM
test_size_shared
+
+-- The `+ rand(7) * 0.0D` leaves the key's value unchanged.
+query expect_error(DIVIDE_BY_ZERO)
+SELECT size(element_at(m1, 1.0D / x + rand(7) * 0.0D)),
+ size(element_at(m2, 1.0D / x + rand(7) * 0.0D))
+FROM test_size_shared
+
+-- A single call shares nothing, so Spark skips the divide on the NULL map too.
+query expect_dispatch(size)
+SELECT size(element_at(m1, 1.0D / x + rand(7) * 0.0D)) FROM test_size_shared
+
+-- On rows where the divide cannot fail, the shared `1.0D / x` still sends the
calls back to
+-- Spark.
+statement
+CREATE TABLE test_size_shared_nonzero(
+ m1 map<double, array<int>>, m2 map<double, array<int>>, x double)
+USING parquet
+
+statement
+INSERT INTO test_size_shared_nonzero VALUES
+ (CAST(NULL AS MAP<DOUBLE, ARRAY<INT>>), CAST(NULL AS MAP<DOUBLE,
ARRAY<INT>>), 4.0D),
+ (map(0.5D, array(1, 2)), map(0.5D, array(3)), 2.0D)
+
+query expect_fallback(holds a subexpression that the operator's other
expressions share)
+SELECT size(element_at(m1, 1.0D / x + rand(7) * 0.0D)),
+ size(element_at(m2, 1.0D / x + rand(7) * 0.0D))
+FROM test_size_shared_nonzero
Review Comment:
The `_nonzero` copy of the table exists only so the `expect_fallback` query
can run on rows where the divide cannot fail.
`array_append_null_array_item.sql` (line 109) gets the same effect with `WHERE
_1 = 2` on the existing table. Could this and the matching `_nonzero` tables in
`array_append_shared_subexpression.sql`, `arrays_zip_shared_subexpression.sql`,
`map_from_arrays_shared_subexpression.sql`,
`map_lookup_shared_subexpression.sql` and
`map_lookup_shared_subexpression_operators.sql` drop the second `CREATE` and
`INSERT` by filtering the original table, or by using `1.0D / (x + 1.0D)`? I
did not check that a `WHERE` keeps the sharing in the `DISTRIBUTE BY`, sort and
`TakeOrderedAndProject` cases of the operators file, so the `x + 1.0D` form may
be the safer one there.
##########
spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala:
##########
@@ -388,30 +388,33 @@ class CometMapExpressionSuite extends CometTestBase {
// Finding E: a map nested inside a map value is handed to the JVM
dispatcher whole, so its double
// keys never revisit `CometLiteral`. The guard has to live on the lookup
instead. The outer key
- // type is INT, so inspecting only the outermost map would admit the query;
the fallback comes
- // from the inner `element_at`, whose child is `MapType(DoubleType,
IntegerType)`. Constant folding
- // is on here, which is the only configuration where the inner map becomes
such a literal, so this
- // regression cannot be expressed in a SQL fixture (the harness disables
folding). The direct
- // single-level double-key lookups live in `element_at_map.sql` /
`get_map_value.sql`.
- test("nested map lookup with floating-point keys falls back") {
+ // type is INT, so inspecting only the outermost map would admit the query;
the dispatch comes
+ // from the `element_at` whose child is `MapType(DoubleType, IntegerType)`,
which takes the whole
+ // nested lookup with it. Constant folding is on here, which is the only
configuration where the
+ // inner map becomes such a literal, so this regression cannot be expressed
in a SQL fixture (the
+ // harness disables folding). The direct single-level double-key lookups
live in
Review Comment:
This comment says the regression cannot be expressed in a SQL fixture
because the harness disables folding. `-- ConstantFolding: enabled` is
supported on `main` (`folded_array_map_literals.sql`, `cast_literal_ansi.sql`),
and `CometArrayExpressionSuite` already says so. Since this PR rewrites the "is
dispatched" tests in this suite anyway, could they move to a fixture with that
header and `expect_dispatch(element_at)`? If not, the sentence needs
correcting. I did not check that the nested multi-row forms reproduce in the
harness.
##########
spark/src/test/resources/sql-tests/expressions/array/array_append_ansi_null_array.sql:
##########
@@ -0,0 +1,84 @@
+-- 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 with an item
that is not a
+-- literal or column as incompatible, so by default the expression runs
through the JVM codegen
+-- dispatcher, which raises like Spark. A literal or column item cannot raise,
and a non-nullable
+-- array evaluates the item on every row on both paths, so both stay native.
+-- `array_append_null_array_item.sql` covers the same rules with ANSI off.
+--
+-- 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_class(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 column or literal item cannot raise, so a nullable array keeps the native
guarded path.
+query expect_native(array_append)
+SELECT _1, array_append(arr, _1) AS a
+FROM test_array_append_ansi
+
+query expect_native(array_append)
+SELECT _1, array_append(arr, 7) AS a
+FROM test_array_append_ansi
+
+query expect_native(array_append)
+SELECT _1, array_append(arr, CAST(NULL AS INT)) AS a
+FROM test_array_append_ansi
+
+-- Any other item over a nullable array is routed through the dispatcher, even
one that cannot
+-- raise on this data.
+query expect_dispatch(array_append)
+SELECT _1, array_append(arr, _1 + 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
+
+query expect_native(array_append)
+SELECT _1, array_append(array(1), _1 + 1) AS a
+FROM test_array_append_ansi
Review Comment:
The routing queries here (column item, literal item, `_1 + 1` dispatched,
non-nullable array native) are the same as the ones in
`array_append_null_array_item.sql`, and both headers say the gate does not
depend on ANSI. Only the `1 / (_1 - 1)` pair, the `CAST(NULL AS INT)` item and
the `array(1), _1` case are specific to this file or missing from the other
one. Could this file keep those and the matrix live once, or could both modes
run from one file with `element_at`, which raises in both?
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -1347,10 +1352,53 @@ object QueryPlanSerde extends Logging with
CometExprShim with CometTypeShim {
val later = expr.children.tail
// Subexpression elimination never shares a leaf, such as a column or a
literal
later.forall(_.isInstanceOf[LeafExpression]) ||
sharedSubexpressions.value.exists { s =>
- !later.exists(_.exists(e => s.common.exists(_.semanticEquals(e))))
+ !holdsSharedSubexpression(later, s)
}
}
+ /**
+ * Whether some node of `expr` with more than one child holds, in a child it
can skip, a
+ * subexpression that the operator's expressions share. A node can skip any
child after its
+ * first, a `DivModLike` or `Pmod` its dividend too, a `StringLocate`, which
evaluates its start
+ * first, its other two children, and an `ArrayJoin` with a nullable
replacement, which
+ * evaluates that first, its array and delimiter, while an `ArrayAppend`
skips none. Spark
+ * evaluates that subexpression for every row, while `expr` evaluated alone,
as the JVM codegen
+ * dispatcher does, can skip it on the rows where a child evaluated before
it is NULL, so an
+ * ANSI error Spark raises goes unraised. With
`spark.sql.subexpressionElimination.enabled` off
+ * only an aggregation shares subexpressions (see `sharedSubexpressions`). A
subexpression that
+ * `expr` itself repeats does not count while the setting is on: the
dispatcher then eliminates
+ * it too, and evaluates it for every row. False outside
`withSharedSubexpressions`, where what
+ * the operator shares is unknown.
+ */
+ def skipsSharedSubexpression(expr: Expression): Boolean =
+ sharedSubexpressions.value.exists { s =>
+ // The dispatcher eliminates subexpressions only while the setting is on
+ lazy val repeated =
+ if (SQLConf.get.subexpressionEliminationEnabled)
commonSubexpressions(Seq(expr))
+ else Nil
+ expr.exists { node =>
+ val skippable = node match {
+ // `DivModLike` and `Pmod` skip the dividend when the divisor,
evaluated first, is NULL
+ case _: DivModLike | _: Pmod => node.children
Review Comment:
The comment says `DivModLike` and `Pmod` skip the dividend, but the branch
returns `node.children`, which includes the divisor. In Spark 3.5.9
`DivModLike.doGenCode` evaluates `eval2` (the divisor) first and puts only
`eval1` (the dividend) inside the branch it skips, and `eval` says the same, so
the divisor is never skipped. The effect is only a decline where Spark would
not skip anything, for a shared subexpression that sits only in the divisor. Is
that intended? If not, `node.children.take(1)` matches the comment. I read
3.5.9 only.
##########
spark/src/test/resources/sql-tests/expressions/misc/codegen_dispatch_calendar_interval_output.sql:
##########
@@ -0,0 +1,83 @@
+-- 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.
+
+-- The JVM codegen dispatcher writes a calendar interval's microseconds to
Arrow as nanoseconds,
+-- which overflows a long beyond about 292 years, a range Spark represents
+-- (https://github.com/apache/datafusion-comet/issues/5279). An expression
that would only
+-- run through the dispatcher declines it when its result holds a calendar
interval at any
+-- nesting depth, so the projection falls back to Spark. The small values fall
back as well,
+-- since the decision is made on the type.
+
+statement
+CREATE TABLE test_interval_output_big(s decimal(18, 6)) USING parquet
+
+statement
+INSERT INTO test_interval_output_big VALUES (10000000000.000000),
(-10000000000.000000), (NULL)
+
+statement
+CREATE TABLE test_interval_output_small(s decimal(18, 6)) USING parquet
+
+statement
+INSERT INTO test_interval_output_small VALUES (1.000000), (-2.500001), (NULL)
+
+query expect_fallback(holds a calendar interval)
+SELECT arrays_zip(array(make_interval(0, 0, 0, 0, 0, 0, s))) FROM
test_interval_output_big
+
+-- An interval inside a struct inside the zipped array.
+query expect_fallback(holds a calendar interval)
+SELECT arrays_zip(array(named_struct('i', make_interval(0, 0, 0, 0, 0, 0,
s)))) FROM test_interval_output_big
+
+-- An interval as a map value inside the zipped array.
+query expect_fallback(holds a calendar interval)
+SELECT arrays_zip(array(map(1, make_interval(0, 0, 0, 0, 0, 0, s)))) FROM
test_interval_output_big
+
+-- A floating-point map key has no native lookup, so these would dispatch.
+query expect_fallback(holds a calendar interval)
+SELECT element_at(map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0,
0, 0, s))), 1.0D)
+FROM test_interval_output_big
+
+query expect_fallback(holds a calendar interval)
+SELECT map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0, 0, 0,
s)))[1.0D]
+FROM test_interval_output_big
+
+-- Native map construction compares floating-point keys by bits, so this would
dispatch.
+query expect_fallback(holds a calendar interval)
+SELECT map_from_arrays(array(1.0D), array(make_interval(0, 0, 0, 0, 0, 0, s)))
+FROM test_interval_output_big
+
+-- A nondeterministic child has no native path, so this would dispatch.
+query expect_fallback(holds a calendar interval)
+SELECT map_from_arrays(array(monotonically_increasing_id()),
array(make_interval(0, 0, 0, 0, 0, 0, s)))
+FROM test_interval_output_big
+
+query expect_fallback(holds a calendar interval)
+SELECT arrays_zip(array(make_interval(0, 0, 0, 0, 0, 0, s))) FROM
test_interval_output_small
+
+query expect_fallback(holds a calendar interval)
+SELECT arrays_zip(array(named_struct('i', make_interval(0, 0, 0, 0, 0, 0,
s)))) FROM test_interval_output_small
Review Comment:
The header (line 22) says the decision is made on the type, so these two
queries on `test_interval_output_small` repeat the first two on
`test_interval_output_big` and cannot catch anything those miss. The only use
of the small table that differs is the `expect_dispatch` control on line 75,
and `map(1, s)` has no interval, so it works on the big table too. Could the
small table and these two queries go?
--
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]
