perf: Enable native c2r for more queries [WIP]

spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala

@@ -22,7 +22,7 @@ package org.apache.comet.rules
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.util.sideBySide
-import org.apache.spark.sql.comet.{CometCollectLimitExec, CometColumnarToRowExec, CometNativeColumnarToRowExec, CometNativeWriteExec, CometPlan, CometScanExec, CometSparkToColumnarExec}
+import org.apache.spark.sql.comet.{CometCollectLimitExec, CometColumnarToRowExec, CometNativeColumnarToRowExec, CometNativeWriteExec, CometPlan, CometSparkToColumnarExec}
 import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometShuffleExchangeExec}
 import org.apache.spark.sql.execution.{ColumnarToRowExec, RowToColumnarExec, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.QueryStageExec
@@ -139,35 +139,12 @@ case class EliminateRedundantTransitions(session: SparkSession) extends Rule[Spa
   private def createColumnarToRowExec(child: SparkPlan): SparkPlan = {
     val schema = child.schema
     val useNative = CometConf.COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED.get() &&
-      CometNativeColumnarToRowExec.supportsSchema(schema) &&
-      !hasScanUsingMutableBuffers(child)
+      CometNativeColumnarToRowExec.supportsSchema(schema)
 
     if (useNative) {
       CometNativeColumnarToRowExec(child)
     } else {
       CometColumnarToRowExec(child)
     }
   }
-
-  /**
-   * Checks if the plan contains a scan that uses mutable buffers. Native C2R is not compatible
-   * with such scans because the buffers may be modified after C2R reads them.
-   *
-   * This includes:
-   *   - CometScanExec with native_iceberg_compat and partition columns - uses
-   *     ConstantColumnReader
-   */
-  private def hasScanUsingMutableBuffers(op: SparkPlan): Boolean = {
-    op match {
-      case c: QueryStageExec => hasScanUsingMutableBuffers(c.plan)
-      case c: ReusedExchangeExec => hasScanUsingMutableBuffers(c.child)
-      case _ =>
-        op.exists {
-          case scan: CometScanExec =>
-            scan.scanImpl == CometConf.SCAN_NATIVE_ICEBERG_COMPAT &&
-            scan.relation.partitionSchema.nonEmpty
-          case _ => false
-        }
-    }
-  }
 }

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q1.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometProject
       +- CometBroadcastHashJoin

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q10.native_iceberg_compat/extended.txt

@@ -1,6 +1,6 @@
 TakeOrderedAndProject
 +- HashAggregate
-   +- CometColumnarToRow
+   +- CometNativeColumnarToRow
       +- CometColumnarExchange
          +- HashAggregate
             +- Project
@@ -11,7 +11,7 @@ TakeOrderedAndProject
                   :     :  +- Filter
                   :     :     +- BroadcastHashJoin
                   :     :        :-  BroadcastHashJoin [COMET: Unsupported join type ExistenceJoin(exists#1)]
-                  :     :        :  :- CometColumnarToRow
+                  :     :        :  :- CometNativeColumnarToRow
                   :     :        :  :  +- CometBroadcastHashJoin
                   :     :        :  :     :- CometFilter
                   :     :        :  :     :  +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer
@@ -30,7 +30,7 @@ TakeOrderedAndProject
                   :     :        :  :                    +- CometFilter
                   :     :        :  :                       +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
                   :     :        :  +- BroadcastExchange
-                  :     :        :     +- CometColumnarToRow
+                  :     :        :     +- CometNativeColumnarToRow
                   :     :        :        +- CometProject
                   :     :        :           +- CometBroadcastHashJoin
                   :     :        :              :- CometScan [native_iceberg_compat] parquet spark_catalog.default.web_sales
@@ -40,7 +40,7 @@ TakeOrderedAndProject
                   :     :        :                    +- CometFilter
                   :     :        :                       +- CometScan [native_iceberg_compat] parquet spark_catalog.default.date_dim
                   :     :        +- BroadcastExchange
-                  :     :           +- CometColumnarToRow
+                  :     :           +- CometNativeColumnarToRow
                   :     :              +- CometProject
                   :     :                 +- CometBroadcastHashJoin
                   :     :                    :- CometScan [native_iceberg_compat] parquet spark_catalog.default.catalog_sales

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q11.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometProject
       +- CometBroadcastHashJoin

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q12.native_iceberg_compat/extended.txt

@@ -1,7 +1,7 @@
 TakeOrderedAndProject
 +- Project
    +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
-      +- CometColumnarToRow
+      +- CometNativeColumnarToRow
          +- CometSort
             +- CometExchange
                +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q13.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometHashAggregate
    +- CometExchange
       +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q14a.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange
@@ -8,7 +8,7 @@ CometColumnarToRow
                   :- CometProject
                   :  +- CometFilter
                   :     :  +- Subquery
-                  :     :     +- CometColumnarToRow
+                  :     :     +- CometNativeColumnarToRow
                   :     :        +- CometHashAggregate
                   :     :           +- CometExchange
                   :     :              +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q14b.native_iceberg_compat/extended.txt

@@ -1,9 +1,9 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometBroadcastHashJoin
       :- CometFilter
       :  :  +- Subquery
-      :  :     +- CometColumnarToRow
+      :  :     +- CometNativeColumnarToRow
       :  :        +- CometHashAggregate
       :  :           +- CometExchange
       :  :              +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q15.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q16.native_iceberg_compat/extended.txt

@@ -1,9 +1,9 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometHashAggregate
    +- CometColumnarExchange
       +- HashAggregate
          +-  HashAggregate [COMET: Unsupported aggregation mode PartialMerge]
-            +- CometColumnarToRow
+            +- CometNativeColumnarToRow
                +- CometHashAggregate
                   +- CometProject
                      +- CometBroadcastHashJoin

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q17.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q18.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q19.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q2.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometSort
    +- CometExchange
       +- CometProject

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q20.native_iceberg_compat/extended.txt

@@ -1,7 +1,7 @@
 TakeOrderedAndProject
 +- Project
    +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
-      +- CometColumnarToRow
+      +- CometNativeColumnarToRow
          +- CometSort
             +- CometExchange
                +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q21.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometFilter
       +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q22.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q23a.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometHashAggregate
    +- CometExchange
       +- CometHashAggregate
@@ -47,7 +47,7 @@ CometColumnarToRow
             :     :        +- CometProject
             :     :           +- CometFilter
             :     :              :  +- Subquery
-            :     :              :     +- CometColumnarToRow
+            :     :              :     +- CometNativeColumnarToRow
             :     :              :        +- CometHashAggregate
             :     :              :           +- CometExchange
             :     :              :              +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q23b.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometUnion
       :- CometHashAggregate
@@ -50,7 +50,7 @@ CometColumnarToRow
       :              :     :     +- CometProject
       :              :     :        +- CometFilter
       :              :     :           :  +- Subquery
-      :              :     :           :     +- CometColumnarToRow
+      :              :     :           :     +- CometNativeColumnarToRow
       :              :     :           :        +- CometHashAggregate
       :              :     :           :           +- CometExchange
       :              :     :           :              +- CometHashAggregate

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q24a.native_iceberg_compat/extended.txt

@@ -1,16 +1,16 @@
 Filter
 :  +- Subquery
 :     +- HashAggregate
-:        +- CometColumnarToRow
+:        +- CometNativeColumnarToRow
 :           +- CometColumnarExchange
 :              +- HashAggregate
 :                 +- HashAggregate
-:                    +- CometColumnarToRow
+:                    +- CometNativeColumnarToRow
 :                       +- CometColumnarExchange
 :                          +- HashAggregate
 :                             +- Project
 :                                +-  BroadcastHashJoin [COMET: Comet is not compatible with Spark for case conversion in locale-specific cases. Set spark.comet.caseConversion.enabled=true to enable it anyway.]
-:                                   :- CometColumnarToRow
+:                                   :- CometNativeColumnarToRow
 :                                   :  +- CometProject
 :                                   :     +- CometBroadcastHashJoin
 :                                   :        :- CometProject
@@ -47,16 +47,16 @@ Filter
 :                                            +- CometFilter
 :                                               +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
 +- HashAggregate
-   +- CometColumnarToRow
+   +- CometNativeColumnarToRow
       +- CometColumnarExchange
          +- HashAggregate
             +- HashAggregate
-               +- CometColumnarToRow
+               +- CometNativeColumnarToRow
                   +- CometColumnarExchange
                      +- HashAggregate
                         +- Project
                            +-  BroadcastHashJoin [COMET: Comet is not compatible with Spark for case conversion in locale-specific cases. Set spark.comet.caseConversion.enabled=true to enable it anyway.]
-                              :- CometColumnarToRow
+                              :- CometNativeColumnarToRow
                               :  +- CometProject
                               :     +- CometBroadcastHashJoin
                               :        :- CometProject

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q24b.native_iceberg_compat/extended.txt

@@ -1,16 +1,16 @@
 Filter
 :  +- Subquery
 :     +- HashAggregate
-:        +- CometColumnarToRow
+:        +- CometNativeColumnarToRow
 :           +- CometColumnarExchange
 :              +- HashAggregate
 :                 +- HashAggregate
-:                    +- CometColumnarToRow
+:                    +- CometNativeColumnarToRow
 :                       +- CometColumnarExchange
 :                          +- HashAggregate
 :                             +- Project
 :                                +-  BroadcastHashJoin [COMET: Comet is not compatible with Spark for case conversion in locale-specific cases. Set spark.comet.caseConversion.enabled=true to enable it anyway.]
-:                                   :- CometColumnarToRow
+:                                   :- CometNativeColumnarToRow
 :                                   :  +- CometProject
 :                                   :     +- CometBroadcastHashJoin
 :                                   :        :- CometProject
@@ -47,16 +47,16 @@ Filter
 :                                            +- CometFilter
 :                                               +- CometScan [native_iceberg_compat] parquet spark_catalog.default.customer_address
 +- HashAggregate
-   +- CometColumnarToRow
+   +- CometNativeColumnarToRow
       +- CometColumnarExchange
          +- HashAggregate
             +- HashAggregate
-               +- CometColumnarToRow
+               +- CometNativeColumnarToRow
                   +- CometColumnarExchange
                      +- HashAggregate
                         +- Project
                            +-  BroadcastHashJoin [COMET: Comet is not compatible with Spark for case conversion in locale-specific cases. Set spark.comet.caseConversion.enabled=true to enable it anyway.]
-                              :- CometColumnarToRow
+                              :- CometNativeColumnarToRow
                               :  +- CometProject
                               :     +- CometBroadcastHashJoin
                               :        :- CometProject

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q25.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q26.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q27.native_iceberg_compat/extended.txt

@@ -1,4 +1,4 @@
-CometColumnarToRow
+CometNativeColumnarToRow
 +- CometTakeOrderedAndProject
    +- CometHashAggregate
       +- CometExchange