From 5c4785517c9cd54b40e258c626ae97860c298c02 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Fri, 17 Jul 2026 14:42:28 -0700 Subject: [PATCH 1/2] [SPARK-58199][SQL] Add FileIndex fullyPushedFilters hint for physical plan Add an optional fullyPushedFilters(partitionKeyFilters, dataFilters) method to the FileIndex trait and use it in FileSourceStrategy to determine which filters can be removed from the row-level FilterExec above the scan. The default implementation returns partitionKeyFilters, preserving existing behavior. Co-authored-by: Isaac --- .../sql/execution/datasources/FileIndex.scala | 29 +++++++ .../datasources/FileSourceStrategy.scala | 6 +- .../datasources/FileSourceStrategySuite.scala | 82 +++++++++++++++++++ 3 files changed, 115 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala index 0291a5fd28a7..1417ee98fcec 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala @@ -133,6 +133,35 @@ trait FileIndex { def listFiles( partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): Seq[PartitionDirectory] + /** + * Given the `partitionKeyFilters` and remaining `dataFilters` Spark will use to prune + * files for this index, returns the filters that this index guarantees are fully applied for all + * returned rows. Only these are dropped from the row-level `FilterExec` above the + * scan; any filter not returned is still evaluated after the scan for correctness. By default + * only partition-key filters are considered fully pushed, preserving the standard file-source + * behavior. + * + * For example: + * - An index that fully applies additional predicates (e.g. one that prunes on non-partition + * columns) can override this to also return a subset of `dataFilters`. + * - Conversely, an index must omit a filter that it only partially applies. With partition + * evolution, not all data files are partitioned by a given partition column (e.g. older files + * written under a previous partition spec). Such files may contain both matching and + * non-matching rows for that column, so the partition filter is not fully pushed and must be + * omitted. + * + * Note these filters are the same predicates that [[listFiles]] prunes with, but not necessarily + * the exact expressions [[listFiles]] receives. The planner may derive the [[listFiles]] + * arguments from these (e.g. decomposing struct-equality predicates into per-field predicates + * for pushdown, substituting scalar-subquery or dynamic-partition values, or omitting + * dynamic-pruning filters from the static listing). The filters passed here are the original, + * un-derived forms so that the returned subset can be matched against those in the `FilterExec` + * above the scan. + */ + def fullyPushedFilters( + partitionKeyFilters: Seq[Expression], + dataFilters: Seq[Expression]): Seq[Expression] = partitionKeyFilters + /** * Returns the list of files that will be read when scanning this relation. This call may be * very expensive for large tables. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala index e2427222d8eb..ac21b5280cdf 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala @@ -212,8 +212,10 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging { .flatMap(DataSourceStrategy.translateFilter(_, supportNestedPredicatePushdown)) logInfo(log"Pushed Filters: ${MDC(PUSHED_FILTERS, pushedFilters.mkString(","))}") - // Predicates with both partition keys and attributes need to be evaluated after the scan. - val afterScanFilters = filterSet -- partitionKeyFilters.filter(_.references.nonEmpty) + // Give the FileIndex a chance to decide which filters can be dropped from the final + // FilterExec above the scan. By default only partition-key filters are dropped. + val afterScanFilters = filterSet -- fsRelation.location.fullyPushedFilters( + partitionKeyFilters.filter(_.references.nonEmpty).toSeq, dataFilters) val maxToStringFields = fsRelation.sparkSession.conf.get(SQLConf.MAX_TO_STRING_FIELDS) logInfo(log"Post-Scan Filters: ${MDC(POST_SCAN_FILTERS, afterScanFilters.simpleString(maxToStringFields))}") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index f38996adbd99..6b2ccf6f5b4c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -252,6 +252,34 @@ class FileSourceStrategySuite extends SharedSparkSession { assert(getPhysicalFilters(df2) contains resolve(df2, "(p1 + c2) = 2")) } + test("FileIndex fullyPushedFilters controls final FilterExec removal") { + val (table, testIndex) = + createTableWithFullyPushedFiltersIndex( + files = Seq( + "p1=1/file1" -> 10, + "p1=2/file2" -> 10)) + + // By default the index returns the partition-key filters, matching legacy behavior, so the + // partition-key filter is removed from the FilterExec above the scan. + val defaultPartitionFilter = table.where("p1 = 1") + assert(!getPhysicalFilters(defaultPartitionFilter).contains( + resolve(defaultPartitionFilter, "p1 = 1"))) + + // Returning nothing means no filter is fully pushed, so even the partition-key filter stays + // in the FilterExec above the scan. + testIndex.setFullyPushedFilters((_, _) => Nil) + val explicitNoFullyPushed = table.where("p1 = 1") + assert(getPhysicalFilters(explicitNoFullyPushed).contains( + resolve(explicitNoFullyPushed, "p1 = 1"))) + + // Returning a data filter as fully pushed removes it from the FilterExec above the scan. + val dataFilter = table.where("c1 = 1") + testIndex.setFullyPushedFilters((_, _) => Seq(resolve(dataFilter, "c1 = 1"))) + val explicitFullyPushed = table.where("c1 = 1") + assert(!getPhysicalFilters(explicitFullyPushed).contains( + resolve(explicitFullyPushed, "c1 = 1"))) + } + test("bucketed table") { val table = createTable( @@ -707,6 +735,60 @@ class FileSourceStrategySuite extends SharedSparkSession { } } + private class FullyPushedFiltersTestFileIndex( + sparkSession: SparkSession, + rootPathsSpecified: Seq[Path], + parameters: Map[String, String], + userSpecifiedSchema: Option[StructType]) + extends InMemoryFileIndex( + sparkSession, + rootPathsSpecified, + parameters, + userSpecifiedSchema) { + + private var pushedFilters: (Seq[Expression], Seq[Expression]) => Seq[Expression] = + (partitionKeyFilters, _) => partitionKeyFilters + + override def fullyPushedFilters( + partitionKeyFilters: Seq[Expression], + dataFilters: Seq[Expression]): Seq[Expression] = + pushedFilters(partitionKeyFilters, dataFilters) + + def setFullyPushedFilters( + f: (Seq[Expression], Seq[Expression]) => Seq[Expression]): Unit = { + pushedFilters = f + } + } + + private def createTableWithFullyPushedFiltersIndex( + files: Seq[(String, Int)]): (DataFrame, FullyPushedFiltersTestFileIndex) = { + val tempDir = Utils.createTempDir() + files.foreach { + case (name, size) => + val file = new File(tempDir, name) + assert(file.getParentFile.exists() || Utils.createDirectory(file.getParentFile)) + util.stringToFile(file, "*".repeat(size)) + } + + val dataSchema = StructType(Nil) + .add("c1", IntegerType) + .add("c2", IntegerType) + val fileIndex = new FullyPushedFiltersTestFileIndex( + spark, + Seq(new Path(tempDir.getCanonicalPath)), + Map.empty[String, String], + Some(dataSchema)) + val relation = HadoopFsRelation( + location = fileIndex, + partitionSchema = fileIndex.partitionSchema, + dataSchema = dataSchema, + bucketSpec = None, + fileFormat = TestFileFormat(), + options = Map.empty)(spark) + + Dataset.ofRows(spark, LogicalRelation(relation)) -> fileIndex + } + def getFileScanRDD(df: DataFrame): FileScanRDD = { df.queryExecution.executedPlan.collect { case scan: DataSourceScanExec if scan.inputRDDs().head.isInstanceOf[FileScanRDD] => From 95cc35378c0b5cadbe1530cf9d6e8ef079df515f Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 21 Jul 2026 16:48:08 -0700 Subject: [PATCH 2/2] Replace fullyPushedFilters with canFullPushDownPartitionFilter Simplify the FileIndex hint to return the subset of partition filters that are fully applied for all returned rows, instead of taking dataFilters and returning an arbitrary filter subset. Removes the javadoc about planner expansion/derived expressions since it no longer applies. --- .../sql/execution/datasources/FileIndex.scala | 35 +++++----------- .../datasources/FileSourceStrategy.scala | 8 ++-- .../datasources/FileSourceStrategySuite.scala | 42 +++++++------------ 3 files changed, 30 insertions(+), 55 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala index 1417ee98fcec..b4ed1c4a0a54 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala @@ -134,33 +134,18 @@ trait FileIndex { partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): Seq[PartitionDirectory] /** - * Given the `partitionKeyFilters` and remaining `dataFilters` Spark will use to prune - * files for this index, returns the filters that this index guarantees are fully applied for all - * returned rows. Only these are dropped from the row-level `FilterExec` above the - * scan; any filter not returned is still evaluated after the scan for correctness. By default - * only partition-key filters are considered fully pushed, preserving the standard file-source - * behavior. + * Given the `partitionFilters` Spark will use to prune files for this index, returns the + * subset that this index guarantees are fully applied for all returned rows. Only these are + * dropped from the row-level `FilterExec` above the scan; any filter not returned is still + * evaluated after the scan for correctness. * - * For example: - * - An index that fully applies additional predicates (e.g. one that prunes on non-partition - * columns) can override this to also return a subset of `dataFilters`. - * - Conversely, an index must omit a filter that it only partially applies. With partition - * evolution, not all data files are partitioned by a given partition column (e.g. older files - * written under a previous partition spec). Such files may contain both matching and - * non-matching rows for that column, so the partition filter is not fully pushed and must be - * omitted. - * - * Note these filters are the same predicates that [[listFiles]] prunes with, but not necessarily - * the exact expressions [[listFiles]] receives. The planner may derive the [[listFiles]] - * arguments from these (e.g. decomposing struct-equality predicates into per-field predicates - * for pushdown, substituting scalar-subquery or dynamic-partition values, or omitting - * dynamic-pruning filters from the static listing). The filters passed here are the original, - * un-derived forms so that the returned subset can be matched against those in the `FilterExec` - * above the scan. + * The default returns all `partitionFilters`, matching standard file-source behavior where + * partition filters are guaranteed by file listing. Override to omit filters that are only + * partially applied - for example with partition evolution, where not all data files are + * partitioned by a given partition column and may contain both matching and non-matching rows. */ - def fullyPushedFilters( - partitionKeyFilters: Seq[Expression], - dataFilters: Seq[Expression]): Seq[Expression] = partitionKeyFilters + def canFullPushDownPartitionFilter(partitionFilters: Seq[Expression]): Seq[Expression] = + partitionFilters /** * Returns the list of files that will be read when scanning this relation. This call may be diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala index ac21b5280cdf..1aa00a1a1482 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala @@ -212,10 +212,10 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging { .flatMap(DataSourceStrategy.translateFilter(_, supportNestedPredicatePushdown)) logInfo(log"Pushed Filters: ${MDC(PUSHED_FILTERS, pushedFilters.mkString(","))}") - // Give the FileIndex a chance to decide which filters can be dropped from the final - // FilterExec above the scan. By default only partition-key filters are dropped. - val afterScanFilters = filterSet -- fsRelation.location.fullyPushedFilters( - partitionKeyFilters.filter(_.references.nonEmpty).toSeq, dataFilters) + // Drop partition-key filters that the FileIndex guarantees are fully applied from the + // FilterExec above the scan. By default all partition-key filters are dropped. + val afterScanFilters = filterSet -- fsRelation.location.canFullPushDownPartitionFilter( + partitionKeyFilters.filter(_.references.nonEmpty).toSeq) val maxToStringFields = fsRelation.sparkSession.conf.get(SQLConf.MAX_TO_STRING_FIELDS) logInfo(log"Post-Scan Filters: ${MDC(POST_SCAN_FILTERS, afterScanFilters.simpleString(maxToStringFields))}") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index 6b2ccf6f5b4c..03d43b4a78ed 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -252,32 +252,25 @@ class FileSourceStrategySuite extends SharedSparkSession { assert(getPhysicalFilters(df2) contains resolve(df2, "(p1 + c2) = 2")) } - test("FileIndex fullyPushedFilters controls final FilterExec removal") { + test("FileIndex canFullPushDownPartitionFilter controls final FilterExec removal") { val (table, testIndex) = - createTableWithFullyPushedFiltersIndex( + createTableWithCanFullPushDownPartitionFilterIndex( files = Seq( "p1=1/file1" -> 10, "p1=2/file2" -> 10)) - // By default the index returns the partition-key filters, matching legacy behavior, so the + // By default the index returns the partition filters, matching legacy behavior, so the // partition-key filter is removed from the FilterExec above the scan. val defaultPartitionFilter = table.where("p1 = 1") assert(!getPhysicalFilters(defaultPartitionFilter).contains( resolve(defaultPartitionFilter, "p1 = 1"))) - // Returning nothing means no filter is fully pushed, so even the partition-key filter stays - // in the FilterExec above the scan. - testIndex.setFullyPushedFilters((_, _) => Nil) + // Returning nothing means no partition filter is fully pushed, so the partition-key + // filter stays in the FilterExec above the scan. + testIndex.setCanFullPushDownPartitionFilter(_ => Nil) val explicitNoFullyPushed = table.where("p1 = 1") assert(getPhysicalFilters(explicitNoFullyPushed).contains( resolve(explicitNoFullyPushed, "p1 = 1"))) - - // Returning a data filter as fully pushed removes it from the FilterExec above the scan. - val dataFilter = table.where("c1 = 1") - testIndex.setFullyPushedFilters((_, _) => Seq(resolve(dataFilter, "c1 = 1"))) - val explicitFullyPushed = table.where("c1 = 1") - assert(!getPhysicalFilters(explicitFullyPushed).contains( - resolve(explicitFullyPushed, "c1 = 1"))) } test("bucketed table") { @@ -735,7 +728,7 @@ class FileSourceStrategySuite extends SharedSparkSession { } } - private class FullyPushedFiltersTestFileIndex( + private class CanFullPushDownPartitionFilterTestFileIndex( sparkSession: SparkSession, rootPathsSpecified: Seq[Path], parameters: Map[String, String], @@ -746,22 +739,19 @@ class FileSourceStrategySuite extends SharedSparkSession { parameters, userSpecifiedSchema) { - private var pushedFilters: (Seq[Expression], Seq[Expression]) => Seq[Expression] = - (partitionKeyFilters, _) => partitionKeyFilters + private var canFullPushDown: Seq[Expression] => Seq[Expression] = identity - override def fullyPushedFilters( - partitionKeyFilters: Seq[Expression], - dataFilters: Seq[Expression]): Seq[Expression] = - pushedFilters(partitionKeyFilters, dataFilters) + override def canFullPushDownPartitionFilter( + partitionFilters: Seq[Expression]): Seq[Expression] = + canFullPushDown(partitionFilters) - def setFullyPushedFilters( - f: (Seq[Expression], Seq[Expression]) => Seq[Expression]): Unit = { - pushedFilters = f + def setCanFullPushDownPartitionFilter(f: Seq[Expression] => Seq[Expression]): Unit = { + canFullPushDown = f } } - private def createTableWithFullyPushedFiltersIndex( - files: Seq[(String, Int)]): (DataFrame, FullyPushedFiltersTestFileIndex) = { + private def createTableWithCanFullPushDownPartitionFilterIndex( + files: Seq[(String, Int)]): (DataFrame, CanFullPushDownPartitionFilterTestFileIndex) = { val tempDir = Utils.createTempDir() files.foreach { case (name, size) => @@ -773,7 +763,7 @@ class FileSourceStrategySuite extends SharedSparkSession { val dataSchema = StructType(Nil) .add("c1", IntegerType) .add("c2", IntegerType) - val fileIndex = new FullyPushedFiltersTestFileIndex( + val fileIndex = new CanFullPushDownPartitionFilterTestFileIndex( spark, Seq(new Path(tempDir.getCanonicalPath)), Map.empty[String, String],