LuciferYang commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4439286054
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.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 evaluated against each file's partition
values, 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 the same reason.
+ */
+ 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()
+
+ /**
+ * Plans the files whose partition values satisfy `expressions`, taken from
what `partitions`
+ * planned and bin-packed again for the surviving set.
+ *
+ * Filtering what was planned rather than listing the files again keeps one
listing per scan, and
+ * the files a subclass excluded in `partitions` stay excluded. How the
survivors group into
+ * partitions is decided here again, for their own size. Where each file was
cut into splits is
+ * not: `partitions` cut them for the whole set it listed, before these
filters existed.
+ *
+ * A partition value is fixed within a file, so every row of every partition
returned satisfies
+ * the expressions even though one partition can still hold files from
several partition
+ * directories.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ val fieldIndex = readPartitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizeName(name) -> i }.toMap
+ // Evaluating these is the scan's only chance to apply them, since the
attributes are declared
+ // fully pushed and Spark drops the post-scan filter. A reference outside
the read partition
+ // schema has no value to evaluate against, so fail loudly rather than
return wrong rows. Spark
+ // screens against `filterAttributes()` before it gets here, which is a
subset of that schema.
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a => fieldIndex.contains(normalizeName(a.name))))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"read partition columns
${readPartitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ val bound = expressions.reduce(And).transform {
+ case a: Attribute => BoundReference(fieldIndex(normalizeName(a.name)),
a.dataType, a.nullable)
+ }
+ val predicate = Predicate.createInterpreted(bound)
+ val keptFiles = plannedPartitions
Review Comment:
Done in bfc057d. A `FileScan` that does not extend a built-in scan keeps
master's behavior, so its `planInputPartitions()` override is never bypassed.
`DataSourceV2FileScanSubclassSuite` runs a DPP join and a scalar subquery on a
scan with its own partition type.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -43,6 +45,7 @@ trait FileScan extends Scan
with Batch
with SupportsReportStatistics
with SupportsMetadata
+ with SupportsRuntimeCatalystFiltering
Review Comment:
Done in bfc057d. `FileScan` no longer mixes in
`SupportsRuntimeCatalystFiltering`, so a subclass implementing
`SupportsRuntimeFiltering` is not a dual-interface scan.
`DataSourceV2FileScanSubclassSuite` runs a plain `SELECT`, a DPP join and a
scalar subquery on two such scans, and checks that their own `filter` still
prunes the DPP join.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.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 evaluated against each file's partition
values, 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 the same reason.
+ */
+ 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()
+
+ /**
+ * Plans the files whose partition values satisfy `expressions`, taken from
what `partitions`
+ * planned and bin-packed again for the surviving set.
+ *
+ * Filtering what was planned rather than listing the files again keeps one
listing per scan, and
+ * the files a subclass excluded in `partitions` stay excluded. How the
survivors group into
+ * partitions is decided here again, for their own size. Where each file was
cut into splits is
+ * not: `partitions` cut them for the whole set it listed, before these
filters existed.
+ *
+ * A partition value is fixed within a file, so every row of every partition
returned satisfies
+ * the expressions even though one partition can still hold files from
several partition
+ * directories.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ val fieldIndex = readPartitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizeName(name) -> i }.toMap
+ // Evaluating these is the scan's only chance to apply them, since the
attributes are declared
+ // fully pushed and Spark drops the post-scan filter. A reference outside
the read partition
+ // schema has no value to evaluate against, so fail loudly rather than
return wrong rows. Spark
+ // screens against `filterAttributes()` before it gets here, which is a
subset of that schema.
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a => fieldIndex.contains(normalizeName(a.name))))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"read partition columns
${readPartitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ val bound = expressions.reduce(And).transform {
+ case a: Attribute => BoundReference(fieldIndex(normalizeName(a.name)),
a.dataType, a.nullable)
+ }
+ val predicate = Predicate.createInterpreted(bound)
+ val keptFiles = plannedPartitions
+ .flatMap(_.files.filter(file => predicate.eval(file.partitionValues)))
+ .sortBy(_.length)(implicitly[Ordering[Long]].reverse)
+ // Pack the survivors instead of handing back the planned partitions with
files removed:
+ // `partitions` sized those bins for the whole file set, so a scan that
pruned most of its files
+ // would run the same few tasks over a fraction of the data. V1 packs the
pruned set the same
+ // way, from `dynamicallySelectedPartitions`.
+ val openCostInBytes = conf.filesOpenCostInBytes
+ val maxSplitBytes = FilePartition.maxSplitBytes(
+ sparkSession, keptFiles.map(_.length + openCostInBytes).sum)
+ FilePartition.getFilePartitions(sparkSession, keptFiles,
maxSplitBytes).toArray
Review Comment:
With per-format opt-in in bfc057d, only the built-in scans take this path,
and none of them reports a partitioning or an ordering. The new trait's
scaladoc says a scan that overrides `planInputPartitions()` or reports either
should not mix it in.
--
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]
