fix: handle NaN equality in array_position to match Spark semantics

Treat NaN == NaN in float/double comparisons, matching Spark's
ordering.equiv() behavior. This makes array_position Compatible
rather than Incompatible.

Author: andygrove

native/spark-expr/src/array_funcs/array_position.rs

@@ -86,6 +86,27 @@ macro_rules! find_position_primitive {
     }};
 }
 
+/// Float-aware comparison that treats NaN == NaN (matching Spark's ordering.equiv() semantics).
+macro_rules! find_position_float {
+    ($list_items:expr, $element:expr, $row_index:expr, $arrow_type:ty) => {{
+        let items = $list_items.as_primitive::<$arrow_type>();
+        let search = $element.as_primitive::<$arrow_type>();
+        let search_val = search.value($row_index);
+        let search_is_nan = search_val.is_nan();
+        let mut pos: i64 = 0;
+        for i in 0..items.len() {
+            if !items.is_null(i) {
+                let item_val = items.value(i);
+                if (search_is_nan && item_val.is_nan()) || item_val == search_val {
+                    pos = (i + 1) as i64;
+                    break;
+                }
+            }
+        }
+        pos
+    }};
+}
+
 fn find_position_in_row(
     list_items: &ArrayRef,
     element: &ArrayRef,
@@ -109,12 +130,8 @@ fn find_position_in_row(
         DataType::Int16 => find_position_primitive!(list_items, element, row_index, Int16Type),
         DataType::Int32 => find_position_primitive!(list_items, element, row_index, Int32Type),
         DataType::Int64 => find_position_primitive!(list_items, element, row_index, Int64Type),
-        DataType::Float32 => {
-            find_position_primitive!(list_items, element, row_index, Float32Type)
-        }
-        DataType::Float64 => {
-            find_position_primitive!(list_items, element, row_index, Float64Type)
-        }
+        DataType::Float32 => find_position_float!(list_items, element, row_index, Float32Type),
+        DataType::Float64 => find_position_float!(list_items, element, row_index, Float64Type),
         DataType::Decimal128(_, _) => {
             find_position_primitive!(list_items, element, row_index, Decimal128Type)
         }

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

@@ -664,10 +664,7 @@ object CometSize extends CometExpressionSerde[Size] {
 
 object CometArrayPosition extends CometExpressionSerde[ArrayPosition] with ArraysBase {
 
-  override def getSupportLevel(expr: ArrayPosition): SupportLevel =
-    Incompatible(Some(
-      "element comparison uses IEEE 754 equality where NaN != NaN, " +
-        "but Spark treats NaN as equal to NaN"))
+  override def getSupportLevel(expr: ArrayPosition): SupportLevel = Compatible()
 
   override def convert(
       expr: ArrayPosition,

spark/src/test/resources/sql-tests/expressions/array/array_position.sql

@@ -15,7 +15,6 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Config: spark.comet.expression.ArrayPosition.allowIncompatible=true
 -- ConfigMatrix: parquet.enable.dictionary=false,true
 
 statement
@@ -188,7 +187,7 @@ INSERT INTO test_ap_nan VALUES
   (array(cast(1.0 as float), cast('NaN' as float), cast(2.0 as float)), cast('NaN' as float)),
   (array(cast(1.0 as float), cast(2.0 as float)), cast('NaN' as float))
 
-query ignore(NaN equality: IEEE 754 says NaN != NaN but Spark treats NaN == NaN)
+query
 SELECT array_position(arr, val) FROM test_ap_nan
 
 -- decimal arrays

spark/src/test/scala/org/apache/spark/sql/benchmark/CometArrayExpressionBenchmark.scala

@@ -52,14 +52,10 @@ object CometArrayExpressionBenchmark extends CometBenchmarkBase {
                |  cast((value + 5) % 100 as int) as search_val
                |FROM $tbl""".stripMargin))
 
-        val extraConfigs =
-          Map("spark.comet.expression.ArrayPosition.allowIncompatible" -> "true")
-
         runExpressionBenchmark(
           "array_position - int array",
           values,
-          "SELECT array_position(int_arr, search_val) FROM parquetV1Table",
-          extraConfigs)
+          "SELECT array_position(int_arr, search_val) FROM parquetV1Table")
       }
     }
 
@@ -84,14 +80,10 @@ object CometArrayExpressionBenchmark extends CometBenchmarkBase {
                |  cast((value + 5) % 100 as string) as search_val
                |FROM $tbl""".stripMargin))
 
-        val extraConfigs =
-          Map("spark.comet.expression.ArrayPosition.allowIncompatible" -> "true")
-
         runExpressionBenchmark(
           "array_position - string array",
           values,
-          "SELECT array_position(str_arr, search_val) FROM parquetV1Table",
-          extraConfigs)
+          "SELECT array_position(str_arr, search_val) FROM parquetV1Table")
       }
     }
   }