fix: Scala 2.13 compilation errors and unused parameter warnings

- Fix type mismatch by adding .toSeq to ArrayBuffer in CometNativeCompaction
- Replace asScala on CloseableIterator with explicit while loop for Scala 2.13 compatibility
- Remove unused warehouseDir parameters and filesBefore/nativeFilesAfter variables
- Add @transient annotation to unused spark parameter

Author: Shekharrajak

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

@@ -69,8 +69,8 @@ class CometIcebergCompactionBenchmarkTest extends CometTestBase {
     CometConf.COMET_EXEC_ENABLED.key -> "true",
     CometConf.COMET_ICEBERG_COMPACTION_ENABLED.key -> "true")
 
+  // scalastyle:off parameter.number
   private def runTableBenchmark(
-      warehouseDir: File,
       sourceTable: String,
       schema: String,
       numFragments: Int,
@@ -91,8 +91,6 @@ class CometIcebergCompactionBenchmarkTest extends CometTestBase {
       """)
     }
 
-    val filesBefore = spark.sql(s"SELECT * FROM $tableName.files").count()
-
     // Benchmark 1: Spark default compaction
     val sparkStart = System.nanoTime()
     val sparkTable = Spark3Util.loadIcebergTable(spark, tableName)
@@ -147,7 +145,6 @@ class CometIcebergCompactionBenchmarkTest extends CometTestBase {
              l_returnflag STRING, l_linestatus STRING"""
         val (lSpark, lNative, lSpeedup) =
           runTableBenchmark(
-            warehouseDir,
             "lineitem",
             lineitemSchema,
             numFragments,
@@ -160,7 +157,7 @@ class CometIcebergCompactionBenchmarkTest extends CometTestBase {
              o_orderdate DATE, o_orderpriority STRING, o_clerk STRING, o_shippriority INT,
              o_comment STRING"""
         val (oSpark, oNative, oSpeedup) =
-          runTableBenchmark(warehouseDir, "orders", ordersSchema, numFragments, rowsPerFragment)
+          runTableBenchmark("orders", ordersSchema, numFragments, rowsPerFragment)
         println(f"${"orders"}%-15s $oSpark%12d $oNative%12d ${oSpeedup}%9.2fx")
 
         // Customer benchmark
@@ -169,7 +166,6 @@ class CometIcebergCompactionBenchmarkTest extends CometTestBase {
              c_phone STRING, c_acctbal DOUBLE, c_mktsegment STRING, c_comment STRING"""
         val (cSpark, cNative, cSpeedup) =
           runTableBenchmark(
-            warehouseDir,
             "customer",
             customerSchema,
             numFragments,

spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergTPCCompactionBenchmark.scala

@@ -127,7 +127,7 @@ object CometIcebergTPCCompactionBenchmark extends CometBenchmarkBase {
   private def runPartitionedTableBenchmark(dataLocation: String, numFragments: Int): Unit = {
     val tableFilePath = resolveTablePath(dataLocation, "lineitem")
 
-    withIcebergWarehouse { (warehouseDir, catalog) =>
+    withIcebergWarehouse { (_, catalog) =>
       val icebergTableName = s"$catalog.db.lineitem_partitioned"
 
       // Create fragmented partitioned table
@@ -193,7 +193,7 @@ object CometIcebergTPCCompactionBenchmark extends CometBenchmarkBase {
 
     val tableFilePath = resolveTablePath(dataLocation, tableName)
 
-    withIcebergWarehouse { (warehouseDir, catalog) =>
+    withIcebergWarehouse { (_, catalog) =>
       val icebergTableName = s"$catalog.db.${tableName}_iceberg"
 
       // Create fragmented table once to measure metadata
@@ -217,7 +217,6 @@ object CometIcebergTPCCompactionBenchmark extends CometBenchmarkBase {
       val nativeTable = Spark3Util.loadIcebergTable(spark, icebergTableName)
       new CometNativeCompaction(spark).rewriteDataFiles(nativeTable)
       val nativeTimeMs = (System.nanoTime() - nativeStart) / 1000000
-      val nativeFilesAfter = spark.sql(s"SELECT * FROM $icebergTableName.files").count()
 
       // Calculate speedup
       val speedup = if (nativeTimeMs > 0) sparkTimeMs.toDouble / nativeTimeMs.toDouble else 0.0

spark/src/test/scala/org/apache/spark/sql/comet/CometNativeCompaction.scala

@@ -85,7 +85,7 @@ case class NativeCompactionResult(
 /**
  * Native Iceberg compaction using Rust/DataFusion for scan+write, Java API for commit.
  */
-class CometNativeCompaction(spark: SparkSession) extends Logging {
+class CometNativeCompaction(@transient spark: SparkSession) extends Logging {
 
   private val objectMapper = new ObjectMapper().registerModule(DefaultScalaModule)
   private val native = new Native()
@@ -136,8 +136,7 @@ class CometNativeCompaction(spark: SparkSession) extends Logging {
         tableConfig,
         group,
         targetFileSizeBytes,
-        compression,
-        table.location())
+        compression)
       val result = executeNativeCompaction(compactionConfig)
 
       result match {
@@ -172,7 +171,8 @@ class CometNativeCompaction(spark: SparkSession) extends Logging {
         s"Committing compaction: ${allFilesToDelete.size} files to delete, " +
           s"${allFilesToAdd.size} files to add")
 
-      val commitSuccess = commitCompaction(table, allFilesToDelete, allFilesToAdd)
+      val commitSuccess =
+        commitCompaction(table, allFilesToDelete.toSeq, allFilesToAdd.toSeq)
 
       if (!commitSuccess) {
         throw new RuntimeException("Failed to commit compaction results")
@@ -264,8 +264,7 @@ class CometNativeCompaction(spark: SparkSession) extends Logging {
       tableConfig: IcebergTableConfig,
       tasks: Seq[FileScanTask],
       targetFileSizeBytes: Long,
-      compression: String,
-      tableLocation: String): CompactionTaskConfig = {
+      compression: String): CompactionTaskConfig = {
 
     val fileScanTaskConfigs = tasks.map { task =>
       val partitionPath = task.spec().partitionToPath(task.file().partition())
@@ -312,12 +311,12 @@ class CometNativeCompaction(spark: SparkSession) extends Logging {
     try {
       val specs = table.specs()
       val deleteFiles: java.util.Set[DataFile] = new java.util.HashSet[DataFile]()
-      val deletePathSet = filesToDelete.toSet
       val snapshot = table.currentSnapshot()
       if (snapshot != null) {
-        import scala.jdk.CollectionConverters._
-        val fileScanTasks = table.newScan().planFiles().iterator().asScala
-        fileScanTasks.foreach { task =>
+        val deletePathSet = filesToDelete.toSet
+        val fileScanTasks = table.newScan().planFiles().iterator()
+        while (fileScanTasks.hasNext) {
+          val task = fileScanTasks.next()
           val dataFile = task.file()
           if (deletePathSet.contains(dataFile.path().toString)) {
             deleteFiles.add(dataFile)