fix: [branch-1.0] honour Spark's legacy null IN () behavior (#5127)

docs/source/user-guide/latest/compatibility/index.md

@@ -145,9 +145,6 @@ so users hunting an unexpected value have a single place to check:
 - Native `RANGE` window frames with an explicit `PRECEDING` / `FOLLOWING` offset diverge from
   Spark when the boundary arithmetic overflows for `DATE` or `DECIMAL` `ORDER BY` columns
   ([#5022](https://github.com/apache/datafusion-comet/issues/5022)).
-- For an empty `IN` list, Comet always returns `false` for a `NULL` operand. Spark returns `NULL`
-  when `spark.sql.legacy.nullInEmptyListBehavior=true` (Spark 4.0+)
-  ([#4786](https://github.com/apache/datafusion-comet/issues/4786)).
 
 ## Object store cache
 

spark/src/main/scala/org/apache/comet/serde/predicates.scala

@@ -22,9 +22,11 @@ package org.apache.comet.serde
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.catalyst.expressions.{And, Attribute, BinaryExpression, EqualNullSafe, EqualTo, Expression, GreaterThan, GreaterThanOrEqual, In, InSet, IsNaN, IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Not, Or}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.BooleanType
 
-import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, withFallbackReason}
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde._
 
@@ -34,34 +36,39 @@ object CometNot extends CometExpressionSerde[Not] {
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
 
-    import ComparisonUtils.hasCollatedOperand
+    // The fused fast paths below build a single native node from `Not` and its child, bypassing
+    // `exprToProtoInternal`, and with it the child serde's `getSupportLevel` gate. Consult that
+    // gate here so the fast path is only taken when the child has a native path, rather than
+    // re-checking the child's conditions by hand.
+    def hasNativePath[T <: Expression](serde: CometExpressionSerde[T], child: T): Boolean =
+      serde.getSupportLevel(child).isInstanceOf[Compatible]
 
     expr.child match {
-      case inner: EqualTo if !hasCollatedOperand(inner.left, inner.right) =>
+      case inner: EqualTo if hasNativePath(CometEqualTo, inner) =>
         createBinaryExpr(
           inner,
           inner.left,
           inner.right,
           inputs,
           binding,
           (builder, binaryExpr) => builder.setNeq(binaryExpr))
-      case inner: EqualNullSafe if !hasCollatedOperand(inner.left, inner.right) =>
+      case inner: EqualNullSafe if hasNativePath(CometEqualNullSafe, inner) =>
         createBinaryExpr(
           inner,
           inner.left,
           inner.right,
           inputs,
           binding,
           (builder, binaryExpr) => builder.setNeqNullSafe(binaryExpr))
-      case inner: In if !hasCollatedOperand((inner.value +: inner.list): _*) =>
+      case inner: In if hasNativePath(CometIn, inner) =>
         ComparisonUtils.in(inner, inner.value, inner.list, inputs, binding, negate = true)
       case _ =>
-        // Includes the collated variants of EqualTo / EqualNullSafe / In above: fall through so
-        // the child expression's own serde is consulted, which now returns `Unsupported` for
-        // non-UTF8_BINARY operands (see `CometEqualTo.getSupportLevel` and siblings). That makes
-        // `exprToProtoInternal` return None for the child, which cascades this Not to None and
-        // falls the enclosing operator back to Spark — the only way to honour collation-aware
-        // (in)equality without a native path.
+        // Includes the cases the child serdes above declare as having no native path, such as
+        // non-UTF8_BINARY collated operands and the legacy `null IN ()` behavior: fall through so
+        // the child expression's own serde is consulted. `exprToProtoInternal` then either routes
+        // the child through the JVM codegen dispatcher (Spark's own `doGenCode`), keeping this Not
+        // native, or returns None, which cascades this Not to None and falls the enclosing
+        // operator back to Spark.
         createUnaryExpr(
           expr,
           expr.child,
@@ -244,15 +251,9 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] {
 object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback {
 
   override def getSupportLevel(expr: In): SupportLevel =
-    ComparisonUtils.collationSupportLevel("In", (expr.value +: expr.list): _*)
+    ComparisonUtils.inSupportLevel("In", expr.list, (expr.value +: expr.list): _*)
 
-  override def getUnsupportedReasons(): Seq[String] =
-    Seq(ComparisonUtils.nonDefaultCollationDocReason)
-
-  override def getCompatibleNotes(): Seq[String] = Seq(
-    "For an empty `IN` list, Comet always returns `false` for a `NULL` operand. Spark returns" +
-      " `NULL` when `spark.sql.legacy.nullInEmptyListBehavior=true` (Spark 4.0+)" +
-      " ([#4786](https://github.com/apache/datafusion-comet/issues/4786)).")
+  override def getUnsupportedReasons(): Seq[String] = ComparisonUtils.inUnsupportedReasons
 
   override def convert(
       expr: In,
@@ -265,10 +266,9 @@ object CometIn extends CometExpressionSerde[In] with CodegenDispatchFallback {
 object CometInSet extends CometExpressionSerde[InSet] with CodegenDispatchFallback {
 
   override def getSupportLevel(expr: InSet): SupportLevel =
-    ComparisonUtils.collationSupportLevel("InSet", expr.child)
+    ComparisonUtils.inSupportLevel("InSet", expr.hset, expr.child)
 
-  override def getUnsupportedReasons(): Seq[String] =
-    Seq(ComparisonUtils.nonDefaultCollationDocReason)
+  override def getUnsupportedReasons(): Seq[String] = ComparisonUtils.inUnsupportedReasons
 
   override def convert(
       expr: InSet,
@@ -330,6 +330,41 @@ object ComparisonUtils {
       Compatible()
     }
 
+  // Spark's `In` / `InSet` return `false` for any operand against an empty list, except under the
+  // legacy `null IN ()` behavior, where a `NULL` operand returns `NULL` (SPARK-44550). Comet's
+  // native `in` kernel only implements the non-legacy behavior, so the legacy case is marked
+  // `Unsupported` and, because both serdes mix in `CodegenDispatchFallback`, routed through the
+  // JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet pipeline).
+  private val legacyNullInEmptyListConfig = "spark.sql.legacy.nullInEmptyListBehavior"
+
+  private val legacyNullInEmptyListReason: String =
+    "An empty `IN` list has no native path when Spark's legacy `null IN ()` behavior is in " +
+      s"effect (`$legacyNullInEmptyListConfig`), because a `NULL` operand then evaluates to " +
+      "`NULL` rather than `false`."
+
+  // Spark 3.4 has no config and always uses the legacy behavior, Spark 3.5 defaults the config to
+  // true, and Spark 4.0+ makes it optional with a default of `!spark.sql.ansi.enabled`. Read by
+  // string key so this compiles against every supported Spark version.
+  private def legacyNullInEmptyListBehavior: Boolean =
+    !isSpark35Plus || {
+      val default = !isSpark40Plus || !SQLConf.get.ansiEnabled
+      CometConf.getBooleanConf(legacyNullInEmptyListConfig, default, SQLConf.get)
+    }
+
+  /**
+   * Support level shared by the `In` and `InSet` serdes: `list` is the set of values being tested
+   * against, and `operands` are the expressions whose collation must be checked.
+   */
+  def inSupportLevel(exprName: String, list: Iterable[_], operands: Expression*): SupportLevel =
+    if (list.isEmpty && legacyNullInEmptyListBehavior) {
+      Unsupported(Some(legacyNullInEmptyListReason))
+    } else {
+      collationSupportLevel(exprName, operands: _*)
+    }
+
+  val inUnsupportedReasons: Seq[String] =
+    Seq(nonDefaultCollationDocReason, legacyNullInEmptyListReason)
+
   def in(
       expr: Expression,
       value: Expression,

spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala

@@ -26,7 +26,7 @@ import scala.util.Random
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.{Column, CometTestBase, DataFrame, Row}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, FromUnixTime, Literal, StructsToJson, TruncDate, TruncTimestamp}
-import org.apache.spark.sql.catalyst.optimizer.SimplifyExtractValueOps
+import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, OptimizeIn, SimplifyExtractValueOps}
 import org.apache.spark.sql.comet.CometProjectExec
 import org.apache.spark.sql.execution.{ProjectExec, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -1715,6 +1715,32 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("null IN empty list honours legacy null-in-empty behavior") {
+    // Spark returns NULL for a NULL operand against an empty IN list when the legacy behavior is
+    // in effect (SPARK-44550): always on Spark 3.4, on by default on Spark 3.5, and whenever ANSI
+    // mode is disabled on Spark 4.0+. Comet's native `in` kernel always returns false, so the
+    // legacy case must leave the native path. Empty IN lists are not expressible in SQL and
+    // `OptimizeIn` folds them away, so build the expression via the DataFrame API with that rule
+    // (and `ConvertToLocalRelation`) excluded.
+    withSQLConf(
+      SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+        Seq(ConvertToLocalRelation.ruleName, OptimizeIn.ruleName).mkString(",")) {
+      val data: Seq[(Integer, Int)] = Seq((1, 1), (null, 2))
+      withParquetTable(data, "tbl") {
+        // An unset config exercises the version-dependent default, which follows ANSI mode on
+        // Spark 4.0+ and is always the legacy behavior on Spark 3.x.
+        for (legacy <- Seq(Some("true"), Some("false"), None); ansi <- Seq("true", "false")) {
+          val legacyConf = legacy.map("spark.sql.legacy.nullInEmptyListBehavior" -> _).toSeq
+          withSQLConf(Seq(SQLConf.ANSI_ENABLED.key -> ansi) ++ legacyConf: _*) {
+            val df = sql("SELECT _1 AS a FROM tbl")
+              .select(col("a"), col("a").isin(), !col("a").isin())
+            checkSparkAnswer(df)
+          }
+        }
+      }
+    }
+  }
+
   test("not") {
     Seq(false, true).foreach { dictionary =>
       withSQLConf("parquet.enable.dictionary" -> dictionary.toString) {