Remove redundant RuntimeFilterBound - data filters already pushed via addAllDataFilters

Author: Shekharrajak

native/proto/src/proto/operator.proto

@@ -22,7 +22,6 @@ syntax = "proto3";
 package spark.spark_operator;
 
 import "expr.proto";
-import "literal.proto";
 import "partitioning.proto";
 import "types.proto";
 
@@ -109,18 +108,6 @@ message NativeScan {
   // the map.
   map<string, string> object_store_options = 13;
   bool encryption_enabled = 14;
-  repeated RuntimeFilterBound runtime_filter_bounds = 15;
-}
-
-// Runtime filter bound for row-group pruning in native scan.
-// Extracted from Spark's data filters containing range or IN predicates.
-message RuntimeFilterBound {
-  string column_name = 1;
-  int32 column_index = 2;
-  optional spark.spark_expression.Literal min_value = 3;
-  optional spark.spark_expression.Literal max_value = 4;
-  repeated spark.spark_expression.Literal in_values = 5;
-  string filter_type = 6;  // "minmax", "in", "bloom"
 }
 
 message IcebergScan {

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

@@ -58,12 +58,10 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
       withInfo(scanExec, s"Full native scan disabled because ${COMET_EXEC_ENABLED.key} disabled")
     }
 
-    // Dynamic partition pruning (DPP) is now supported!
-    // The dynamicallySelectedPartitions in CometScanExec evaluates DPP filters
-    // and returns the filtered file list. Native scan receives these pre-filtered
-    // files, so partition-level pruning works correctly.
-    // Note: DPP filters are excluded from dataFilters to avoid pushing subqueries
-    // to native execution (see supportedDataFilters in CometScanExec).
+    // Native DataFusion doesn't support subqueries/dynamic pruning
+    if (scanExec.partitionFilters.exists(isDynamicPruningFilter)) {
+      withInfo(scanExec, "Native DataFusion scan does not support subqueries/dynamic pruning")
+    }
 
     if (SQLConf.get.ignoreCorruptFiles ||
       scanExec.relation.options
@@ -193,10 +191,6 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
         }
       }
 
-      // Add runtime filter bounds if available
-      // These are pushed down from join operators to enable I/O reduction
-      addRuntimeFilterBounds(scan, nativeScanBuilder)
-
       Some(builder.setNativeScan(nativeScanBuilder).build())
 
     } else {
@@ -256,70 +250,4 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
   override def createExec(nativeOp: Operator, op: CometScanExec): CometNativeExec = {
     CometNativeScanExec(nativeOp, op.wrapped, op.session)
   }
-
-  /**
-   * Add runtime filter bounds to the native scan for row-group pruning. Runtime filters are
-   * extracted from data filters that contain range predicates (GreaterThanOrEqual,
-   * LessThanOrEqual) or IN predicates.
-   */
-  private def addRuntimeFilterBounds(
-      scan: CometScanExec,
-      nativeScanBuilder: OperatorOuterClass.NativeScan.Builder): Unit = {
-    import org.apache.spark.sql.catalyst.expressions._
-
-    // Extract runtime filter bounds from data filters
-    scan.supportedDataFilters.foreach {
-      case GreaterThanOrEqual(attr: AttributeReference, Literal(value, dataType)) =>
-        val boundBuilder = OperatorOuterClass.RuntimeFilterBound.newBuilder()
-        boundBuilder.setColumnName(attr.name)
-        boundBuilder.setColumnIndex(scan.output.indexWhere(_.name == attr.name))
-        boundBuilder.setFilterType("minmax")
-        exprToProto(Literal(value, dataType), scan.output).foreach { minProto =>
-          boundBuilder.setMinValue(minProto.getLiteral)
-        }
-        nativeScanBuilder.addRuntimeFilterBounds(boundBuilder.build())
-
-      case LessThanOrEqual(attr: AttributeReference, Literal(value, dataType)) =>
-        val boundBuilder = OperatorOuterClass.RuntimeFilterBound.newBuilder()
-        boundBuilder.setColumnName(attr.name)
-        boundBuilder.setColumnIndex(scan.output.indexWhere(_.name == attr.name))
-        boundBuilder.setFilterType("minmax")
-        exprToProto(Literal(value, dataType), scan.output).foreach { maxProto =>
-          boundBuilder.setMaxValue(maxProto.getLiteral)
-        }
-        nativeScanBuilder.addRuntimeFilterBounds(boundBuilder.build())
-
-      case And(
-            GreaterThanOrEqual(attr1: AttributeReference, Literal(minVal, minType)),
-            LessThanOrEqual(attr2: AttributeReference, Literal(maxVal, maxType)))
-          if attr1.name == attr2.name =>
-        // Combined range filter: column >= min AND column <= max
-        val boundBuilder = OperatorOuterClass.RuntimeFilterBound.newBuilder()
-        boundBuilder.setColumnName(attr1.name)
-        boundBuilder.setColumnIndex(scan.output.indexWhere(_.name == attr1.name))
-        boundBuilder.setFilterType("minmax")
-        exprToProto(Literal(minVal, minType), scan.output).foreach { minProto =>
-          boundBuilder.setMinValue(minProto.getLiteral)
-        }
-        exprToProto(Literal(maxVal, maxType), scan.output).foreach { maxProto =>
-          boundBuilder.setMaxValue(maxProto.getLiteral)
-        }
-        nativeScanBuilder.addRuntimeFilterBounds(boundBuilder.build())
-
-      case InSet(attr: AttributeReference, values) if values.size <= 10 =>
-        // Small IN filter - pass individual values
-        val boundBuilder = OperatorOuterClass.RuntimeFilterBound.newBuilder()
-        boundBuilder.setColumnName(attr.name)
-        boundBuilder.setColumnIndex(scan.output.indexWhere(_.name == attr.name))
-        boundBuilder.setFilterType("in")
-        values.foreach { value =>
-          exprToProto(Literal(value, attr.dataType), scan.output).foreach { valProto =>
-            boundBuilder.addInValues(valProto.getLiteral)
-          }
-        }
-        nativeScanBuilder.addRuntimeFilterBounds(boundBuilder.build())
-
-      case _ => // Other filters are handled by data_filters
-    }
-  }
 }