fix: [iceberg] Keep deep copy for Iceberg Java integration scan path

CometScanWrapper unconditionally set isFfiSafe=true, which told native
ScanExec to skip deep copies for all scans. This is correct for
CometScanExec (native_iceberg_compat) which now uses immutable Arrow
readers, but incorrect for CometBatchScanExec (Iceberg Java integration
via SupportsComet) which still uses mutable buffers.

Make isFfiSafe conditional on the scan type: true for CometScanExec,
false for CometBatchScanExec. Also remove the stale
hasScanUsingMutableBuffers check for CometScanExec since PR apache#3411
replaced mutable buffers with immutable Arrow readers.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: andygrove

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.{CometBatchScanExec, CometCollectLimitExec, CometColumnarToRowExec, CometNativeColumnarToRowExec, CometNativeWriteExec, CometPlan, CometScanExec, CometSparkToColumnarExec}
+import org.apache.spark.sql.comet.{CometBatchScanExec, 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
@@ -155,8 +155,6 @@ case class EliminateRedundantTransitions(session: SparkSession) extends Rule[Spa
    * 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
    *   - CometBatchScanExec with CometParquetScan (V2 Parquet path) - uses BatchReader
    */
   private def hasScanUsingMutableBuffers(op: SparkPlan): Boolean = {
@@ -165,9 +163,6 @@ case class EliminateRedundantTransitions(session: SparkSession) extends Rule[Spa
       case c: ReusedExchangeExec => hasScanUsingMutableBuffers(c.child)
       case _ =>
         op.exists {
-          case scan: CometScanExec =>
-            scan.scanImpl == CometConf.SCAN_NATIVE_ICEBERG_COMPAT &&
-            scan.relation.partitionSchema.nonEmpty
           case scan: CometBatchScanExec => scan.scan.isInstanceOf[CometParquetScan]
           case _ => false
         }

spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala

@@ -37,7 +37,7 @@ import org.apache.comet.serde.QueryPlanSerde.{serializeDataType, supportedDataTy
 abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
 
   /** Whether the data produced by the Comet operator is FFI safe */
-  def isFfiSafe: Boolean = false
+  def isFfiSafe(op: T): Boolean = false
 
   override def enabledConfig: Option[ConfigEntry[Boolean]] = None
 
@@ -61,7 +61,7 @@ abstract class CometSink[T <: SparkPlan] extends CometOperatorSerde[T] {
     } else {
       scanBuilder.setSource(source)
     }
-    scanBuilder.setArrowFfiSafe(isFfiSafe)
+    scanBuilder.setArrowFfiSafe(isFfiSafe(op))
 
     val scanTypes = op.output.flatten { attr =>
       serializeDataType(attr.dataType)
@@ -93,7 +93,7 @@ object CometExchangeSink extends CometSink[SparkPlan] {
    *
    * Source of shuffle exchange batches is NativeBatchDecoderIterator.
    */
-  override def isFfiSafe: Boolean = true
+  override def isFfiSafe(op: SparkPlan): Boolean = true
 
   override def createExec(nativeOp: Operator, op: SparkPlan): CometNativeExec =
     CometSinkPlaceHolder(nativeOp, op, op)

spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala

@@ -271,7 +271,7 @@ object CometBroadcastExchangeExec extends CometSink[BroadcastExchangeExec] {
    *
    * Source of broadcast exchange batches is ArrowStreamReader.
    */
-  override def isFfiSafe: Boolean = true
+  override def isFfiSafe(op: BroadcastExchangeExec): Boolean = true
 
   override def enabledConfig: Option[ConfigEntry[Boolean]] = Some(
     CometConf.COMET_EXEC_BROADCAST_EXCHANGE_ENABLED)

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -2072,6 +2072,13 @@ case class CometSortMergeJoinExec(
 }
 
 object CometScanWrapper extends CometSink[SparkPlan] {
+  override def isFfiSafe(op: SparkPlan): Boolean = op match {
+    // CometScanExec (native_iceberg_compat) uses immutable Arrow readers
+    case _: CometScanExec => true
+    // CometBatchScanExec (Iceberg Java integration) still uses mutable buffers
+    case _ => false
+  }
+
   override def createExec(nativeOp: Operator, op: SparkPlan): CometNativeExec = {
     CometScanWrapper(nativeOp, op)
   }