arrays_overlap inconsistent behaviour on two arrays with NULL values

[SparkApplicationMaster]

Describe the bug

Spark doc says:

arrays_overlap(a1, a2) - Returns true if a1 contains at least a non-null element present also in a2.
If the arrays have no common element and they are both non-empty and either of them contains a null element null is returned, false otherwise.
https://spark.apache.org/docs/latest/api/sql/index.html#arrays_overlap

But comet returns True instead of NULL if the arrays intersect only by NULL

Steps to reproduce

#change this to your path
comet_jar_path = "/home/x/comet-spark-spark3.5_2.12-0.9.0.jar"

from pyspark.sql import SparkSession

# uncomment to reproduce vanilla spark behavior:
#spark = SparkSession.builder.appName("test").getOrCreate()

# comment if upper is uncommented
spark = SparkSession.builder.appName("test")\
    .config("spark.jars", comet_jar_path)\
    .config("spark.driver.extraClassPath", comet_jar_path)\
    .config("spark.executor.extraClassPath", comet_jar_path)\
    .config("spark.plugins", "org.apache.spark.CometPlugin")\
    .config("spark.shuffle.manager", "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")\
    .config("spark.comet.explainFallback.enabled", "true")\
    .config("spark.comet.expression.allowIncompatible", "true")\
    .getOrCreate()

df = spark.createDataFrame([
    (["a", None], ["b", "c", None]),
    ], ['x', 'y']
)

df.write.mode("overwrite").save("/tmp/tmp_parquet")
spark.read.parquet("/tmp/tmp_parquet").createOrReplaceTempView("t1")

spark.sql("select * from t1").show()
spark.sql("SELECT arrays_overlap(x, y) from t1").show()
spark.sql("SELECT arrays_overlap(x, y) from t1").explain(True)

comet returns true if the arrays intersect only by NULL value:

+---------+------------+
|        x|           y|
+---------+------------+
|[a, NULL]|[b, c, NULL]|
+---------+------------+

+--------------------+
|arrays_overlap(x, y)|
+--------------------+
|                true|
+--------------------+

== Physical Plan ==
*(1) CometColumnarToRow
+- CometProject [arrays_overlap(x, y)#56], [arrays_overlap(x#34, y#35) AS arrays_overlap(x, y)#56]
   +- CometScan parquet [x#34,y#35] Batched: true, DataFilters: [], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/tmp_parquet], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<x:array<string>,y:array<string>>

Expected behavior

spark 3.5.6 returns NULL if the arrays intersect only by NULL value:

+---------+------------+
|        x|           y|
+---------+------------+
|[a, NULL]|[b, c, NULL]|
+---------+------------+

+--------------------+
|arrays_overlap(x, y)|
+--------------------+
|                NULL|
+--------------------+

Additional context

This is because the arrays_overlap call propagates to datafusion array_has_any call, which returns True in this case
Maybe it needs to be changed in datafusion
https://github.com/apache/datafusion/blob/8a3ea8739c36e5cafffff195ffc42ae1fc1e8dde/datafusion/functions-nested/src/array_has.rs#L652
Because if not, then there is no good solution to provide compartibility

[coderfender]

I can take this issue up if you arent already working on this @SparkApplicationMaster ?

[SparkApplicationMaster]

@coderfender thanks for response!

Actually, there is a whole bunch of such inconsistencies in different spark functions between datafusion output and what spark needs to return
So I would like to discuss what to do with all this with some who cares
Maybe add that to datafusion backlog if it is possible
To make datafusion more consistent with spark, so it won't be needed to overcome this type of differences in every function

But feel free to implement this in comet if you would like to
You can even start with my code closing linked issue in sail, because it passes spark-tests and is written in pure datafusion expressions without the need to go deeper to arrow buffers:
lakehq/sail@97b4622

[parthchandra]

@SparkApplicationMaster There is a datafusion-spark crate explicitly for the purpose of creating datafusion expressions that have Spark compatible behavior.
Original discussion: apache/datafusion#5600

[akupchinskiy]

Btw, this test is supposed to check this very behavior. Yet, it doesn't work because the constant expression seems to be folded into a boolean literal during a query planning phase.

[coderfender]

@SparkApplicationMaster , Thank you I tried replicating the issue but it seems like there isn't a regression between Spark and Comet ? Below is my code (perhaps I am missing something here? )

Test case (scala) :

test("check arrays_overlap_inconsistency") {
    val df = spark.range(1)
      .select(
        array(lit("a"), lit(null).cast("string")).as("x"),
        array(lit("b"), lit("c"), lit(null).cast("string")).as("y")
      )
    df.createOrReplaceTempView("t1")
    spark.sql("SELECT x , y ,arrays_overlap(x, y) from t1").show()
    checkSparkAnswerAndOperator(spark.sql("SELECT arrays_overlap(x, y) from t1"))
  }

[SparkApplicationMaster]

@coderfender I suppose createOrReplaceTempView support is not yet implemented to build optimized comet steps, and the query causes backoff to scala spark, and scala spark works well
You can check it by explain(true), if the Comet projection is here or not
It took some time for me to build a test case (in the topic start) which guarantees to use datafusion function, maybe it's not needed to save and read tmp parquets, but this was reproducible

[coderfender]

thank you for the clarification @SparkApplicationMaster . I tried below code and the check fails with a fallback to spark operator.

      spark.sql("CREATE TABLE tbl (x ARRAY<STRING>, y ARRAY<STRING>) USING parquet; ")
      spark.sql("insert into table tbl values (array('a', null), array('b','c', null))")
      val res = spark.sql("select x,y, arrays_overlap(x,y) from tbl")
      res.show(10)
      checkSparkAnswerAndOperator(res)