Remove redundant RuntimeFilterBound - data filters already pushed

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/rules/CometScanRule.scala

@@ -601,8 +601,6 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
       partitionSchema: StructType,
       hadoopConf: Configuration): String = {
 
-    val cometExecEnabled = COMET_EXEC_ENABLED.get()
-
     val fallbackReasons = new ListBuffer[String]()
 
     // native_iceberg_compat only supports local filesystem and S3
@@ -623,6 +621,7 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] with Com
     val partitionSchemaSupported =
       typeChecker.isSchemaSupported(partitionSchema, fallbackReasons)
 
+    val cometExecEnabled = COMET_EXEC_ENABLED.get()
     if (!cometExecEnabled) {
       fallbackReasons += s"$SCAN_NATIVE_ICEBERG_COMPAT requires ${COMET_EXEC_ENABLED.key}=true"
     }

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

@@ -193,10 +193,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 +252,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
-    }
-  }
 }