peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4258931995
##########
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
Review Comment:
I'd leave this one too. The reason sits at the one call site that depends on
it, and repeating it on the member is the kind of restatement this PR removes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -443,10 +428,9 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}.getOrElse(scan)
// The offered conjuncts are skipped on the rows the reader marks as
checked. An `If` rather
- // than an `Or`, so that a null mark leaves a conjunct as it was, an
`And` around it
- // included, and so that subexpression elimination cannot hoist what the
conjunct shares
- // with another one ahead of the skip. They are matched by
`ExpressionSet`, since
- // `storageFilters` carry the relation's column names.
+ // than an `Or`, so a null mark leaves the conjunct as it was, an `And`
around it included,
+ // and subexpression elimination cannot hoist a shared hash ahead of the
skip. They are
Review Comment:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04),
with the general wording back.
##########
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:
Thanks, done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04)
with a qualifier: "lets a columnar scan's task reclaim them".
--
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]
