feat: support literal for ARRAY top level

Author: comphead

native/spark-expr/src/conversion_funcs/cast.rs

@@ -1021,7 +1021,7 @@ fn is_datafusion_spark_compatible(
         DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 => {
             // note that the cast from Int32/Int64 -> Decimal128 here is actually
             // not compatible with Spark (no overflow checks) but we have tests that
-            // rely on this cast working so we have to leave it here for now
+            // rely on this cast working, so we have to leave it here for now
             matches!(
                 to_type,
                 DataType::Boolean

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

@@ -2180,7 +2180,9 @@ object QueryPlanSerde extends Logging with CometExprShim {
     op match {
 
       // Fully native scan for V1
-      case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
+      case scan: CometScanExec
+          if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION
+            || scan.scanImpl == CometConf.SCAN_NATIVE_ICEBERG_COMPAT =>
         val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder()
         nativeScanBuilder.setSource(op.simpleStringWithNodeId())
 

spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala

@@ -24,6 +24,7 @@ import org.scalatest.Tag
 
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.functions.{array, col}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
 
@@ -253,18 +254,11 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
   }
 
   test("native reader - read a STRUCT subfield - field from second") {
-    withSQLConf(
-      CometConf.COMET_EXEC_ENABLED.key -> "true",
-      SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
-      CometConf.COMET_ENABLED.key -> "true",
-      CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
-      CometConf.COMET_NATIVE_SCAN_IMPL.key -> "native_datafusion") {
-      testSingleLineQuery(
-        """
+    testSingleLineQuery(
+      """
           |select 1 a, named_struct('a', 1, 'b', 'n') c0
           |""".stripMargin,
-        "select c0.b from tbl")
-    }
+      "select c0.b from tbl")
   }
 
   test("native reader - read a STRUCT subfield from ARRAY of STRUCTS - field from first") {
@@ -557,11 +551,10 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
       "select array(cast('a' as binary), cast('bc' as binary), cast('def' as binary), null) from tbl")
   }
 
-  test("native reader - array equality") {
-    testSingleLineQuery(
-      """
-        | select array(1) a union all select array(2)
-        |""".stripMargin,
-      "select * from tbl where a = array(1L)")
+  test("SPARK-18053: ARRAY equality is broken") {
+    withTable("array_tbl") {
+      spark.range(10).select(array(col("id")).as("arr")).write.saveAsTable("array_tbl")
+      assert(sql("SELECT * FROM array_tbl where arr = ARRAY(1L)").count == 1)
+    }
   }
 }