fix: Fall back on dynamicpruning expressions for CometIcebergNativeScan (apache#3335)

Author: mbutrovich

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

@@ -327,6 +327,10 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
       case _
           if scanExec.scan.getClass.getName ==
             "org.apache.iceberg.spark.source.SparkBatchQueryScan" =>
+        if (scanExec.runtimeFilters.exists(isDynamicPruningFilter)) {
+          return withInfo(scanExec, "Dynamic Partition Pruning is not supported")
+        }
+
         val fallbackReasons = new ListBuffer[String]()
 
         // Native Iceberg scan requires both configs to be enabled

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

@@ -700,10 +700,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     // If metadata is None, this is a programming error - metadata should have been extracted
     // in CometScanRule before creating CometBatchScanExec
     val metadata = scan.nativeIcebergScanMetadata.getOrElse {
-      logError(
+      throw new IllegalStateException(
         "Programming error: CometBatchScanExec.nativeIcebergScanMetadata is None. " +
           "Metadata should have been extracted in CometScanRule.")
-      return None
     }
 
     // Use pre-extracted metadata (no reflection needed)
@@ -979,11 +978,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       }
     } catch {
       case e: Exception =>
-        val msg =
-          "Iceberg reflection failure: Failed to extract FileScanTasks from Iceberg scan RDD: " +
-            s"${e.getMessage}"
-        logError(msg, e)
-        return None
+        // CometScanRule already validated this scan should use native execution.
+        // Failure here is a programming error, not a graceful fallback scenario.
+        throw new IllegalStateException(
+          s"Native Iceberg scan serialization failed unexpectedly: ${e.getMessage}",
+          e)
     }
 
     // Log deduplication summary

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

@@ -2294,4 +2294,70 @@ class CometIcebergNativeSuite extends CometTestBase with RESTCatalogHelper {
       deleteRecursively(dir)
     }
   }
+
+  test("runtime filtering - join with dynamic partition pruning") {
+    assume(icebergAvailable, "Iceberg not available")
+    withTempIcebergDir { warehouseDir =>
+      val dimDir = new File(warehouseDir, "dim_parquet")
+      withSQLConf(
+        "spark.sql.catalog.runtime_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+        "spark.sql.catalog.runtime_cat.type" -> "hadoop",
+        "spark.sql.catalog.runtime_cat.warehouse" -> warehouseDir.getAbsolutePath,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+        // Create partitioned Iceberg table (fact table)
+        spark.sql("""
+          CREATE TABLE runtime_cat.db.fact_table (
+            id BIGINT,
+            data STRING,
+            date DATE
+          ) USING iceberg
+          PARTITIONED BY (date)
+        """)
+
+        // Insert data across multiple partitions
+        spark.sql("""
+          INSERT INTO runtime_cat.db.fact_table VALUES
+          (1, 'a', DATE '1970-01-01'),
+          (2, 'b', DATE '1970-01-02'),
+          (3, 'c', DATE '1970-01-02'),
+          (4, 'd', DATE '1970-01-03')
+        """)
+
+        // Create dimension table (Parquet) in temp directory
+        spark
+          .createDataFrame(Seq((1L, java.sql.Date.valueOf("1970-01-02"))))
+          .toDF("id", "date")
+          .write
+          .parquet(dimDir.getAbsolutePath)
+        spark.read.parquet(dimDir.getAbsolutePath).createOrReplaceTempView("dim")
+
+        // This join should trigger dynamic partition pruning
+        val query =
+          """SELECT f.* FROM runtime_cat.db.fact_table f
+            |JOIN dim d ON f.date = d.date AND d.id = 1
+            |ORDER BY f.id""".stripMargin
+
+        // Verify the initial plan contains dynamic pruning expression
+        val df = spark.sql(query)
+        val initialPlan = df.queryExecution.executedPlan
+        val planStr = initialPlan.toString
+        assert(
+          planStr.contains("dynamicpruning"),
+          s"Expected dynamic pruning in plan but got:\n$planStr")
+
+        // Check results match Spark
+        // Note: AQE re-plans after subquery executes, converting dynamicpruningexpression(...)
+        // to dynamicpruningexpression(true), which allows native Iceberg scan to proceed.
+        // This is correct behavior - no actual subquery to wait for after AQE re-planning.
+        // However, the rest of the still contains non-native operators because CometExecRule
+        // doesn't run again.
+        checkSparkAnswer(df)
+
+        spark.sql("DROP TABLE runtime_cat.db.fact_table")
+      }
+    }
+  }
 }