LuciferYang commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4056395112
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -186,6 +209,47 @@ trait FileScan extends Scan
partitions.toArray
}
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is applied by selecting partition directories
in `buildPartitions`,
+ * the same treatment a compile-time `partitionFilters` entry gets, so the
scan evaluates it in
+ * full and Spark does not evaluate it again after the scan --
`FileScanBuilder.pushFilters`
+ * already keeps compile-time partition filters out of the post-scan filters
for that reason.
+ * Directory selection matches a predicate's references against
`fileIndex.partitionSchema` by
+ * name, so the names reported here have to be the ones that schema uses;
both sides derive from
+ * it today, and a divergence would silently drop the predicate instead of
failing.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizeName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizeName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
+
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ // Directory selection is the only thing that applies these, and it
silently ignores a predicate
+ // whose references are not all partition columns -- which, for attributes
declared fully
+ // pushed, would leave the predicate evaluated nowhere. Spark screens for
that before it gets
+ // here; fail loudly rather than return wrong rows if that ever stops
being true. Compare the
+ // way the file index does, by exact name against its own partition
schema, so this guard cannot
+ // pass something the index will then drop.
+ val partitionNames = fileIndex.partitionSchema.fieldNames.toSet
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a => partitionNames.contains(a.name)))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"partition columns
${fileIndex.partitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ buildPartitions(partitionFilters ++ expressions).toArray
Review Comment:
You are right, and my earlier answer in this thread was wrong in the case
that matters. It said an override of `partitions` keeps working because the
compile-time path still calls it, which only holds while no runtime filter
fires, and a runtime filter firing is the whole point of the feature.
Taken, in the shape you proposed. `planInputPartitionsWithRuntimeFilters`
now filters what `partitions` already planned instead of listing again: it
binds the expressions against `readPartitionSchema`, drops each
`PartitionedFile` whose `partitionValues` do not satisfy them, and drops the
partitions left empty.
```scala
val predicate = Predicate.createInterpreted(bound)
partitions.flatMap { part =>
val kept = part.files.filter(file => predicate.eval(file.partitionValues))
if (kept.isEmpty) None else Some(part.copy(files = kept))
}.toArray
```
Three things fall out of it. An override of `partitions` is honored, since
its files are the only ones reachable from here. `partitions` is memoized, so
both read paths share one listing and the duplicate driver-side listing and
packing is gone, which takes the first item in the description's list of things
this PR does not address with it. And the `buildPartitions` seam this PR had
added is no longer needed, so `partitions` goes back to what it was on master.
The guard stays, and now compares against `readPartitionSchema`, which is
what the binding uses. It matches names the way the rest of the scan does, so a
mixed-case reference binds rather than being rejected.
New test: `a subclass that narrows the file set in partitions keeps it under
a runtime filter` builds a `ParquetScan` subclass whose `partitions` keeps one
directory's files, then asks it for a filter selecting a different directory.
Reverting the method to the listing-based version turns it red by handing back
exactly the files the override excluded:
```
Array(FilePartition(0, Array(... /fact/part=7/part-00000-....parquet,
partition values: [7])),
FilePartition(1, Array(... /fact/part=7/part-00001-....parquet,
partition values: [7]))) was not empty
```
The same test asserts the scan asks for its partitions once across both
calls; dropping the memoization turns that into `2 did not equal 1`.
The description's user-facing section now also says that the scan is the
only evaluator of such a filter, and that a file whose partition value cannot
be evaluated fails the query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
