fix: array to array cast

Author: manuzhang

.gitignore

@@ -18,3 +18,4 @@ apache-rat-*.jar
 venv
 dev/release/comet-rm/workdir
 spark/benchmarks
+/comet-event-trace.json

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

@@ -1076,8 +1076,22 @@ fn cast_array(
             cast_options,
         )?),
         (List(_), Utf8) => Ok(cast_array_to_string(array.as_list(), cast_options)?),
-        (List(_), List(_)) if can_cast_types(from_type, to_type) => {
-            Ok(cast_with_options(&array, to_type, &CAST_OPTIONS)?)
+        (List(from), List(to))
+            if can_cast_types(from_type, to_type)
+                || (matches!(from.data_type(), Decimal128(_, _))
+                    && matches!(to.data_type(), Boolean)) =>
+        {
+            let list_array = array.as_list::<i32>();
+            Ok(Arc::new(ListArray::new(
+                Arc::clone(to),
+                list_array.offsets().clone(),
+                cast_array(
+                    Arc::clone(list_array.values()),
+                    to.data_type(),
+                    cast_options,
+                )?,
+                list_array.nulls().cloned(),
+            )) as ArrayRef)
         }
         (UInt8 | UInt16 | UInt32 | UInt64, Int8 | Int16 | Int32 | Int64)
             if cast_options.allow_cast_unsigned_ints =>

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

@@ -1154,6 +1154,30 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("cast ArrayType to ArrayType") {
+    val types = Seq(
+      BooleanType,
+      StringType,
+      ByteType,
+      IntegerType,
+      LongType,
+      ShortType,
+      DecimalType(10, 2),
+      DecimalType(38, 18))
+    for (fromType <- types) {
+      for (toType <- types) {
+        if (fromType != toType &&
+          !tags
+            .get(s"cast $fromType to $toType")
+            .exists(s => s.contains("org.scalatest.Ignore")) &&
+          Cast.canCast(fromType, toType) &&
+          CometCast.isSupported(fromType, toType, None, CometEvalMode.LEGACY) == Compatible()) {
+          castTest(generateArrays(100, fromType), ArrayType(toType))
+        }
+      }
+    }
+  }
+
   private def generateFloats(): DataFrame = {
     withNulls(gen.generateFloats(dataSize)).toDF("a")
   }
@@ -1182,10 +1206,10 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     withNulls(gen.generateLongs(dataSize)).toDF("a")
   }
 
-  private def generateArrays(rowSize: Int, elementType: DataType): DataFrame = {
+  private def generateArrays(rowNum: Int, elementType: DataType): DataFrame = {
     import scala.collection.JavaConverters._
     val schema = StructType(Seq(StructField("a", ArrayType(elementType), true)))
-    spark.createDataFrame(gen.generateRows(rowSize, schema).asJava, schema)
+    spark.createDataFrame(gen.generateRows(rowNum, schema).asJava, schema)
   }
 
   // https://github.com/apache/datafusion-comet/issues/2038