feat: enable iceberg compat tests, more tests for complex types

Author: comphead

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -384,13 +384,13 @@ class CometSparkSessionExtensions
 
         // Comet JVM + native scan for V1 and V2
         case op if isCometScan(op) =>
-          val nativeOp = QueryPlanSerde.operator2Proto(op).get
-          CometScanWrapper(nativeOp, op)
+          val nativeOp = QueryPlanSerde.operator2Proto(op)
+          CometScanWrapper(nativeOp.get, op)
 
         case op if shouldApplySparkToColumnar(conf, op) =>
           val cometOp = CometSparkToColumnarExec(op)
-          val nativeOp = QueryPlanSerde.operator2Proto(cometOp).get
-          CometScanWrapper(nativeOp, cometOp)
+          val nativeOp = QueryPlanSerde.operator2Proto(cometOp)
+          CometScanWrapper(nativeOp.get, cometOp)
 
         case op: ProjectExec =>
           val newOp = transform1(op)
@@ -498,15 +498,15 @@ class CometSparkSessionExtensions
           val child = op.child
           val modes = aggExprs.map(_.mode).distinct
 
-          if (!modes.isEmpty && modes.size != 1) {
+          if (modes.nonEmpty && modes.size != 1) {
             // This shouldn't happen as all aggregation expressions should share the same mode.
             // Fallback to Spark nevertheless here.
             op
           } else {
             // For a final mode HashAggregate, we only need to transform the HashAggregate
             // if there is Comet partial aggregation.
             val sparkFinalMode = {
-              !modes.isEmpty && modes.head == Final && findCometPartialAgg(child).isEmpty
+              modes.nonEmpty && modes.head == Final && findCometPartialAgg(child).isEmpty
             }
 
             if (sparkFinalMode) {
@@ -520,7 +520,7 @@ class CometSparkSessionExtensions
                   // distinct aggregate functions or only have group by, the aggExprs is empty and
                   // modes is empty too. If aggExprs is not empty, we need to verify all the
                   // aggregates have the same mode.
-                  assert(modes.length == 1 || modes.length == 0)
+                  assert(modes.length == 1 || modes.isEmpty)
                   CometHashAggregateExec(
                     nativeOp,
                     op,
@@ -529,7 +529,7 @@ class CometSparkSessionExtensions
                     aggExprs,
                     resultExpressions,
                     child.output,
-                    if (modes.nonEmpty) Some(modes.head) else None,
+                    modes.headOption,
                     child,
                     SerializedPlan(None))
                 case None =>
@@ -540,7 +540,7 @@ class CometSparkSessionExtensions
 
         case op: ShuffledHashJoinExec
             if CometConf.COMET_EXEC_HASH_JOIN_ENABLED.get(conf) &&
-              op.children.forall(isCometNative(_)) =>
+              op.children.forall(isCometNative) =>
           val newOp = transform1(op)
           newOp match {
             case Some(nativeOp) =>
@@ -574,7 +574,7 @@ class CometSparkSessionExtensions
 
         case op: BroadcastHashJoinExec
             if CometConf.COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED.get(conf) &&
-              op.children.forall(isCometNative(_)) =>
+              op.children.forall(isCometNative) =>
           val newOp = transform1(op)
           newOp match {
             case Some(nativeOp) =>
@@ -1288,7 +1288,7 @@ object CometSparkSessionExtensions extends Logging {
     op.isInstanceOf[CometBatchScanExec] || op.isInstanceOf[CometScanExec]
   }
 
-  private def shouldApplySparkToColumnar(conf: SQLConf, op: SparkPlan): Boolean = {
+  def shouldApplySparkToColumnar(conf: SQLConf, op: SparkPlan): Boolean = {
     // Only consider converting leaf nodes to columnar currently, so that all the following
     // operators can have a chance to be converted to columnar. Leaf operators that output
     // columnar batches, such as Spark's vectorized readers, will also be converted to native

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

@@ -2721,7 +2721,10 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
 
       case op
           if isCometSink(op) && op.output.forall(a =>
-            supportedDataType(a.dataType, allowComplex = usingDataFusionParquetExec(conf))) =>
+            supportedDataType(
+              a.dataType,
+              allowComplex =
+                usingDataFusionParquetExec(conf) || op.isInstanceOf[CometSparkToColumnarExec])) =>
         // These operators are source of Comet native execution chain
         val scanBuilder = OperatorOuterClass.Scan.newBuilder()
         val source = op.simpleStringWithNodeId()