Fix match arms for old Iceberg integration.

Author: mbutrovich

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

@@ -196,11 +196,10 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
         val nativeOp = operator2Proto(scan).get
         CometNativeScan.createExec(nativeOp, scan)
 
-      // Fully native Iceberg scan for V2
+      // Fully native Iceberg scan for V2 (iceberg-rust path)
+      // Only handle scans with native metadata; SupportsComet scans fall through to isCometScan
       // Config checks (COMET_ICEBERG_NATIVE_ENABLED, COMET_EXEC_ENABLED) are done in CometScanRule
-      case scan: CometBatchScanExec
-          if scan.wrapped.scan.getClass.getName ==
-            "org.apache.iceberg.spark.source.SparkBatchQueryScan" =>
+      case scan: CometBatchScanExec if scan.nativeIcebergScanMetadata.isDefined =>
         operator2Proto(scan) match {
           case Some(nativeOp) =>
             CometIcebergNativeScan.createExec(nativeOp, scan)
@@ -820,10 +819,8 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
       case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
         CometNativeScan.convert(scan, builder, childOp: _*)
 
-      // Fully native Iceberg scan for V2
-      case scan: CometBatchScanExec
-          if scan.wrapped.scan.getClass.getName ==
-            "org.apache.iceberg.spark.source.SparkBatchQueryScan" =>
+      // Fully native Iceberg scan for V2 (iceberg-rust path)
+      case scan: CometBatchScanExec if scan.nativeIcebergScanMetadata.isDefined =>
         CometIcebergNativeScan.convert(scan, builder, childOp: _*)
 
       case op if isCometSink(op) =>

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

@@ -267,7 +267,7 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
         }
 
       // Iceberg scan - patched version implementing SupportsComet interface
-      case s: SupportsComet =>
+      case s: SupportsComet if !COMET_ICEBERG_NATIVE_ENABLED.get() =>
         val fallbackReasons = new ListBuffer[String]()
 
         if (!s.isCometEnabled) {