feat: add COMET_ICEBERG_COMPACTION_ENABLED config option

Author: Shekharrajak

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -150,6 +150,16 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_ICEBERG_COMPACTION_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.iceberg.compaction.enabled")
+      .category(CATEGORY_TESTING)
+      .doc(
+        "Whether to enable Comet-accelerated Iceberg compaction. When enabled, " +
+          "CALL rewrite_data_files() uses Comet's native scan for the read path, " +
+          "reducing JVM overhead during compaction. Experimental.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_CSV_V2_NATIVE_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.scan.csv.v2.enabled")
       .category(CATEGORY_TESTING)

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

@@ -315,9 +315,13 @@ case class CometScanRule(session: SparkSession)
         }
 
       // Iceberg scan - detected by class name (works with unpatched Iceberg)
+      // SparkBatchQueryScan: normal query scans
+      // SparkStagedScan: compaction scans via ScanTaskSetManager
       case _
-          if scanExec.scan.getClass.getName ==
-            "org.apache.iceberg.spark.source.SparkBatchQueryScan" =>
+          if Set(
+            "org.apache.iceberg.spark.source.SparkBatchQueryScan",
+            "org.apache.iceberg.spark.source.SparkStagedScan").contains(
+            scanExec.scan.getClass.getName) =>
         val fallbackReasons = new ListBuffer[String]()
 
         // Native Iceberg scan requires both configs to be enabled