rich7420 commented on code in PR #6073:
URL: https://github.com/apache/datafusion-comet/pull/6073#discussion_r4060561099
##########
native/core/src/execution/planner.rs:
##########
@@ -762,6 +762,14 @@ impl PhysicalPlanner {
.map(|x| self.create_expr(x, Arc::clone(&input_schema)))
.collect::<Result<Vec<_>, _>>()?;
+ // Normalize both sides before DataFusion hashes or compares
nested values.
+ let value = NormalizeNestedFloats::wrap_if_needed(value,
input_schema.as_ref())?;
Review Comment:
With the default optimizer, `a IN (array(double('0.0')))` still returns
`false` for `a = [-0.0]` (Spark: `true`). `OptimizeIn` rewrites it to
`EqualTo`, bypassing this branch. Please normalize both `Eq` and `NotEq`, which
also covers `NOT IN`.
The patch below adds regression coverage with constant folding enabled. The
regression fails before the fix and all five focused tests pass after it on
Spark 4.1.3.
<details>
<summary>Patch (save as nested-in.patch and run git apply
nested-in.patch)</summary>
```diff
diff --git a/native/core/src/execution/planner/macros.rs
b/native/core/src/execution/planner/macros.rs
index 0ec60c0f7..c36e843f9 100644
--- a/native/core/src/execution/planner/macros.rs
+++ b/native/core/src/execution/planner/macros.rs
@@ -93,6 +93,18 @@ macro_rules! binary_expr_builder {
&$operator,
&input_schema,
);
+ // OptimizeIn can rewrite singleton IN/NOT IN to equality
comparisons.
+ let (left, right) = match $operator {
+ datafusion::logical_expr::Operator::Eq
+ | datafusion::logical_expr::Operator::NotEq => {
+ use
datafusion_comet_spark_expr::NormalizeNestedFloats;
+ (
+ NormalizeNestedFloats::wrap_if_needed(left,
&input_schema)?,
+ NormalizeNestedFloats::wrap_if_needed(right,
&input_schema)?,
+ )
+ }
+ _ => (left, right),
+ };
Ok(std::sync::Arc::new(
datafusion::physical_expr::expressions::BinaryExpr::new(left, $operator, right),
))
diff --git
a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index 028dc0475..56692797f 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -52,7 +52,19 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
test("nested floating point membership uses native In and InSet") {
withTable("nested_in_plan") {
sql("CREATE TABLE nested_in_plan (a ARRAY<DOUBLE>) USING parquet")
- sql("INSERT INTO nested_in_plan VALUES (array(CAST('-0.0' AS
DOUBLE)))")
+ sql("""INSERT INTO nested_in_plan VALUES
+ |(array(CAST('-0.0' AS DOUBLE))), (array(CAST('0.0' AS DOUBLE))),
+ |(array(CAST('NaN' AS DOUBLE))), (array(CAST('1.0' AS DOUBLE))),
+ |(array(CAST(NULL AS DOUBLE))), (array()), (NULL)""".stripMargin)
+ withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> "") {
+ for (size <- Seq(1, 2)) {
+ val candidates = Seq.fill(size)("array(CAST('0.0' AS
DOUBLE))").mkString(", ")
+ val df = sql(s"SELECT a IN ($candidates), a NOT IN ($candidates)
FROM nested_in_plan")
+ val expressions =
df.queryExecution.optimizedPlan.flatMap(_.expressions)
+ assert(!expressions.exists(_.exists(_.isInstanceOf[In])))
+ checkSparkAnswerAndImpl(df, native = Seq("equalto"))
+ }
+ }
for (threshold <- Seq(100, 0)) {
withSQLConf("spark.sql.optimizer.inSetConversionThreshold" ->
threshold.toString) {
val df = sql("""SELECT a IN (array(CAST('0.0' AS DOUBLE)),
```
</details>
--
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]
