[SPARK-58199][SQL] Add FileIndex fullyPushedFilter hint for physical plan

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.

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))}")

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] =>