dongjoon-hyun commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4048745686
##########
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:
This path bypasses `partitions` / `planInputPartitions()` entirely. Since
every `FileScan` now opts in by default (`filterAttributes()` and
`fullyPushedFilterAttributes()` return the read partition columns), an existing
3rd-party `FileScan` subclass that narrows the file set by overriding
`partitions` will silently get the excluded files back as soon as DPP or a
scalar-subquery filter fires. For the scalar-subquery case, the `FilterExec`
above the scan is also removed, so nothing downstream masks it.
This is different from the earlier answer in this thread ("existing sources
see no behavior change"), and the PR description's user-facing section doesn't
mention it either.
Could we honor the overrides instead? For example, like V1, prune the
`PartitionedFile`s already produced by `partitions` by evaluating the runtime
filters on their `partitionValues` (bound to `readPartitionSchema`, which
`filterAttributes()` already guarantees covers the referenced columns), then
repack. That would also remove the duplicate driver-side listing/packing
mentioned in the PR description. Alternatively, make the opt-in explicit per
built-in format rather than the default for every `FileScan`.
--
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]
