HyukjinKwon commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4248263397
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -125,33 +106,21 @@ class ParquetStorageFilter private (
val maxSplicedRowGroupBytes: Long) extends Serializable {
/**
- * The predicate this filter evaluates, which the reader resolves once and
then runs per row. It
- * is handed out rather than wrapped, because resolving the `lazy val` costs
a volatile read that
- * a per-row loop would otherwise pay. Codegen-produced predicates can be
awkward to serialize
- * from driver to executor, so it is built on first use on the executor.
+ * The predicate this filter evaluates, built on first use on the executor,
since a generated one
+ * can be awkward to serialize.
*/
@transient lazy val preparedPredicate: BasePredicate = {
- // The field below is nulled the first time this runs, and the copy that
reaches every task is
- // the driver's. So a driver-side evaluation would leave each executor
building a predicate from
- // nothing, which is named here rather than left to an NPE from
`Predicate.create`.
+ // The expression is released below, so a driver-side evaluation would
ship a copy with nothing
+ // to build from. That is named here rather than left to an NPE.
if (boundExpression == null) {
throw SparkException.internalError("a storage filter must only be
evaluated on an " +
"executor, and this copy has released the expression it would build
its predicate from")
}
val created = Predicate.create(boundExpression)
- // A prepared bloom reaches the executor as a binary Literal inside this
expression, up to
- // `spark.sql.optimizer.runtime.bloomFilter.maxNumBits` of it, while a
code-generated `created`
- // holds the deserialized filter instead. Dropping the expression lets a
columnar scan's task
- // reclaim those bytes for the rest of its life. An interpreted `created`,
the fallback when
- // codegen fails or is off, keeps the expression itself, so the bytes
stay. A row-based scan
- // keeps them anyway, since its closure captures the scan node, whose
storage filters hold the
- // same bytes.
- // Safe because every task attempt deserializes its own copy from the
driver's bytes, and this
- // lazy val is the only reader of the field.
- //
- // It does mean the filter must only ever be evaluated on an executor.
Were the driver's copy
- // to run this, it is the copy serialized to every attempt, and it would
go out with nothing
- // left to build a predicate from.
+ // A bloom's bytes sit in the expression as a binary literal, while a
generated predicate holds
+ // the deserialized filter instead, so dropping the expression lets the
task reclaim them. An
Review Comment:
**Nit (P3):** This now reads as if every task with a generated predicate
gets the bloom's bytes back, but that is true only for a columnar scan.
`storageFiltersFor` offers storage filters whenever
`ParquetFileFormat.supportBatch` holds. That check does not look at whole-stage
codegen, while `FileSourceScanExec.supportsColumnar` does. So with
`spark.sql.codegen.wholeStage=false`, or a schema wider than
`wholeStageMaxNumFields`, the scan runs row-based. Its `doExecute` closure
calls `UnsafeProjection.create(schema)` and captures the scan node, whose
`storageFilters` still hold the same literal. Nulling `boundExpression` then
frees nothing for that task. The comment before this PR said so, and the trim
dropped that along with the columnar-scan qualifier.
One sentence after the interpreted-predicate case would keep the trade-off,
for example: "A row-based scan keeps them anyway, since its task closure
captures the scan node, whose storage filters hold the same bytes." Only the
comment is affected. The code is fine.
**Verification:**
- **Inspection:** The comment's stated conditions match the pinned code:
FileSourceScanExec.supportsColumnar needs whole-stage codegen while
ParquetFileFormat.supportBatch does not, and the row-based doExecute closure
references the scan node's schema and therefore its storageFilters.
- **Compile:** The edited Scala file still compiles and the changed comment
lines stay within the 100-character limit.
--
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]
