LuciferYang commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4173545701
##########
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
Review Comment:
Done in 67e88d6. The memo is an `AtomicReference` set with `compareAndSet`,
so `partitions` runs outside the scan's monitor; the remaining `lazy val` only
creates the empty reference. Two threads asking first at the same time may both
list, and both get the stored result. `partitions runs without holding the
scan's monitor` records `Thread.holdsLock(this)` inside an overridden
`partitions` for both read paths, and fails against the old `lazy val`.
##########
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] = {
Review Comment:
Done in 2acbef9. For a `DataSourceV2ScanRelation`,
`CleanupDynamicPruningFilters` also reads equalities from the top-level
conjuncts of `pushedFilters`, so `part <> 3` or `part = 3 OR part = 4` keeps
the DPP. A new test runs the six SPARK-38148 conditions on V1 and V2 parquet
and expects the same DPP decision and rows on both, then checks those two extra
conditions on V2. This applies to any V2 source that fully pushes such an
equality: the result is the same, and the useless DPP is gone.
##########
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] = {
Review Comment:
Agreed. I'll add it to the user-facing section when I next update the
description.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -862,21 +883,17 @@ private class PartitioningBreakingScan(
afterFilter: Seq[InputPartition])
extends Scan with Batch with SupportsRuntimeCatalystFiltering {
- private var filtered = false
-
override def readSchema(): StructType = new StructType().add("part",
IntegerType)
override def toBatch: Batch = this
- override def planInputPartitions(): Array[InputPartition] =
- if (filtered) afterFilter.toArray else initialPartitions.toArray
+ override def planInputPartitions(): Array[InputPartition] =
initialPartitions.toArray
Review Comment:
Done in 2acbef9. It is now `PartitioningBreakingScan(afterFilter) extends
SupportsRuntimeCatalystFiltering`, without the `Batch` members and their
imports.
--
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]
