chore: Enable plan stability suite for native_datafusion scans (#2877)

Author: andygrove

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

@@ -160,6 +160,13 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
           scanImpl = selectScan(scanExec, r.partitionSchema, hadoopConf)
         }
 
+        // Native DataFusion doesn't support subqueries/dynamic pruning
+        if (scanImpl == SCAN_NATIVE_DATAFUSION &&
+          scanExec.partitionFilters.exists(isDynamicPruningFilter)) {
+          fallbackReasons += "Native DataFusion scan does not support subqueries/dynamic pruning"
+          return withInfos(scanExec, fallbackReasons.toSet)
+        }
+
         if (scanImpl == SCAN_NATIVE_DATAFUSION && !COMET_EXEC_ENABLED.get()) {
           fallbackReasons +=
             s"Full native scan disabled because ${COMET_EXEC_ENABLED.key} disabled"

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

@@ -65,7 +65,7 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
         CometConf.COMET_RESPECT_PARQUET_FILTER_PUSHDOWN.get(scan.conf)) {
 
         val dataFilters = new ListBuffer[Expr]()
-        for (filter <- scan.dataFilters) {
+        for (filter <- scan.supportedDataFilters) {
           exprToProto(filter, scan.output) match {
             case Some(proto) => dataFilters += proto
             case _ =>

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

@@ -155,16 +155,21 @@ case class CometScanExec(
     }
   }
 
-  @transient
-  private lazy val pushedDownFilters = {
-    val supportedFilters = if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION) {
-      // `native_datafusion` scan does not support subquery pushdown filters,
-      // see: https://github.com/apache/datafusion-comet/issues/2424
+  /**
+   * Returns the data filters that are supported for this scan implementation. For
+   * native_datafusion scans, this excludes dynamic pruning filters (subqueries)
+   */
+  lazy val supportedDataFilters: Seq[Expression] = {
+    if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION) {
       dataFilters.filterNot(isDynamicPruningFilter)
     } else {
       dataFilters
     }
-    getPushedDownFilters(relation, supportedFilters)
+  }
+
+  @transient
+  private lazy val pushedDownFilters = {
+    getPushedDownFilters(relation, supportedDataFilters)
   }
 
   override lazy val metadata: Map[String, String] =