sunchao commented on code in PR #6599:
URL: https://github.com/apache/datafusion-comet/pull/6599#discussion_r4178402154
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -113,37 +113,24 @@ object CometArrayAppend extends
CometExpressionSerde[ArrayAppend] with ArraysBas
}
}
-object CometArrayContains
- extends CometExpressionSerde[ArrayContains]
- with CodegenDispatchFallback {
-
- private val floatingPointReason: String =
- "Spark compares array elements with ordering.equiv, so -0.0 matches +0.0
and all NaNs match " +
- "each other; Comet's native array_contains compares the raw Arrow values
bitwise"
-
- override def getIncompatibleReasons(): Seq[String] = Seq(floatingPointReason)
-
- override def getSupportLevel(expr: ArrayContains): SupportLevel =
expr.left.dataType match {
- // Native array_contains compares floating-point elements bitwise,
disagreeing with Spark for
- // -0.0/+0.0 and NaN. Report Incompatible (not Unsupported) for
float/double element types (at
- // any nesting level) so the expression routes through the JVM codegen
dispatcher (Spark's own
- // doGenCode) and stays native + Spark-exact under the default config,
while non-float arrays
- // keep the fast native kernel. Under allowIncompatible=true the native
kernel is used
- // as before.
- case ArrayType(elementType, _)
- if SupportLevel.containsType(elementType, classOf[FloatType],
classOf[DoubleType]) =>
- Incompatible(Some(floatingPointReason))
- case _ => Compatible()
- }
+object CometArrayContains extends CometExpressionSerde[ArrayContains] {
override def convert(
expr: ArrayContains,
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
val arrayExprProto = exprToProtoInternal(expr.children.head, inputs,
binding)
val keyExprProto = exprToProtoInternal(expr.children(1), inputs, binding)
-
- scalarFunctionExprToProto("array_contains", arrayExprProto, keyExprProto)
+ // datafusion-spark's array_contains compares floats by their bits.
spark_array_contains
+ // compares them as Spark does, with -0.0 equal to 0.0 and all NaNs equal,
at any depth.
+ val elementType = expr.left.dataType.asInstanceOf[ArrayType].elementType
+ val function =
+ if (SupportLevel.containsType(elementType, classOf[FloatType],
classOf[DoubleType])) {
Review Comment:
[P2] Preserve Spark dispatch for nested string collations. On Spark 4.x, let
`a` have type `ARRAY<STRUCT<d:DOUBLE,s:STRING COLLATE UTF8_LCASE>>` and `v` the
matching struct type. For `a=[(0D,'a')]` and `v=(0D,'A')`,
`array_contains(a,v)` must return `true`. This float-leaf gate now selects
`spark_array_contains`, whose shared comparator compares the string field
bytewise and returns `false`. The previous float fallback evaluated the whole
expression through Spark and preserved collation. Please retain
`CodegenDispatchFallback` and guard `hasNonDefaultStringCollation(elementType)`
before declaring this route compatible.
Evidence: Spark 4.1.3 Catalyst `ArrayContains` returned `true` through both
`eval` and `UnsafeProjection` for the collated struct pair. The exact-head Rust
UDF returned `[Some(false)]` for both scalar and column lookup arguments after
the existing serde’s conversion to Arrow types. Reproduction files:
`/tmp/6599-review/CatalystProbe.scala`, `/tmp/6599-review/review_6599.rs`
(`mixed_struct_string_comparison_probe`). Native command: `cargo test -p
datafusion-comet-spark-expr --test review_6599 --locked --offline --
--nocapture`.
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -113,37 +113,24 @@ object CometArrayAppend extends
CometExpressionSerde[ArrayAppend] with ArraysBas
}
}
-object CometArrayContains
- extends CometExpressionSerde[ArrayContains]
- with CodegenDispatchFallback {
-
- private val floatingPointReason: String =
- "Spark compares array elements with ordering.equiv, so -0.0 matches +0.0
and all NaNs match " +
- "each other; Comet's native array_contains compares the raw Arrow values
bitwise"
-
- override def getIncompatibleReasons(): Seq[String] = Seq(floatingPointReason)
-
- override def getSupportLevel(expr: ArrayContains): SupportLevel =
expr.left.dataType match {
- // Native array_contains compares floating-point elements bitwise,
disagreeing with Spark for
- // -0.0/+0.0 and NaN. Report Incompatible (not Unsupported) for
float/double element types (at
- // any nesting level) so the expression routes through the JVM codegen
dispatcher (Spark's own
- // doGenCode) and stays native + Spark-exact under the default config,
while non-float arrays
- // keep the fast native kernel. Under allowIncompatible=true the native
kernel is used
- // as before.
- case ArrayType(elementType, _)
- if SupportLevel.containsType(elementType, classOf[FloatType],
classOf[DoubleType]) =>
- Incompatible(Some(floatingPointReason))
- case _ => Compatible()
- }
+object CometArrayContains extends CometExpressionSerde[ArrayContains] {
override def convert(
expr: ArrayContains,
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
val arrayExprProto = exprToProtoInternal(expr.children.head, inputs,
binding)
val keyExprProto = exprToProtoInternal(expr.children(1), inputs, binding)
-
- scalarFunctionExprToProto("array_contains", arrayExprProto, keyExprProto)
+ // datafusion-spark's array_contains compares floats by their bits.
spark_array_contains
+ // compares them as Spark does, with -0.0 equal to 0.0 and all NaNs equal,
at any depth.
+ val elementType = expr.left.dataType.asInstanceOf[ArrayType].elementType
+ val function =
+ if (SupportLevel.containsType(elementType, classOf[FloatType],
classOf[DoubleType])) {
+ "spark_array_contains"
+ } else {
+ "array_contains"
+ }
+ scalarFunctionExprToProto(function, arrayExprProto, keyExprProto)
Review Comment:
[P2] Preserve null-array short-circuiting before evaluating a throwing
lookup. With ANSI enabled, consider `SELECT array_contains(a, CAST(s AS
DOUBLE)) FROM t` for `a ARRAY<DOUBLE>, s STRING` containing `(NULL,'bad')` and
`([1D],'1')`. Spark returns `NULL,true` because it skips the cast when `a` is
null. The new native route evaluates both children eagerly and raises
`CastInvalidValue` for `'bad'`, aborting a previously successful query. The old
float route dispatched the whole expression and preserved Spark’s evaluation
order. Please keep such expressions dispatched unless lookup evaluation is
proven safe, or implement masked evaluation of the lookup.
Evidence: Spark 4.1.3 interpreted and generated `ArrayContains` returned
`null` for the null-array/invalid-string row and `true` for the valid row. An
exact-head `ScalarFunctionExpr` using `SparkFloatArrayContains` and Comet’s
ANSI `Cast` on the same batch returned `Err(External(CastInvalidValue { value:
"bad", from_type: "STRING", to_type: "DOUBLE" }))`. Reproduction files:
`/tmp/6599-review/CatalystProbe.scala`, `/tmp/6599-review/review_6599.rs`
(`null_array_throwing_lookup_probe`). `ScalarFunctionExpr::evaluate` evaluates
every child before invoking the UDF.
--
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]
