dongjoon-hyun commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4247335315
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,19 @@ object SQLConf {
val PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED =
buildConf("spark.sql.parquet.storageFilterPushdown.enabled")
.doc("If true, the vectorized reader of the V1 Parquet file source may
apply a runtime " +
- "storage filter, such as a bloom filter from join runtime filtering,
while it reads. " +
- "It reads the columns the " +
- "filter needs first, evaluates the filter per row, and then reads the
remaining columns " +
- "only for the rows that survived. This is a planning-time decision. " +
- "A filter that is attached also stays in the post-scan filter, the way
a pushed data " +
- "filter does, so the reader is free to stop applying it wherever doing
so would cost " +
- "more than it saves, and the answer does not change. In a scan that
returns columnar " +
- "batches, the post-scan filter skips it on the rows the reader applied
it to, which the " +
- "reader marks in a column named _tmp_storage_filter_checked. A table
with a column of " +
- "that name is not offered storage filters, and a file holding one
outside the table's " +
- "schema is not supported. A row group where the reader stops applying
the filter reads " +
- "its key columns twice, which is slower than not pushing the filter. "
+
- "Under spark.sql.files.ignoreCorruptFiles the answer can change,
because this reader " +
- "reads different pages in a different order than a plain read. Which
rows survive a " +
- "corrupt page can then differ from a plain read, in either direction.
" +
- "Returning only the surviving rows of a row group relies on the
Parquet page index, so a " +
- "file whose page index is wrong can pair a row's key with another
row's values, or " +
- "return rows the filter rejects. Setting " +
- "parquet.filter.columnindex.enabled to false makes the reader fall
back to skipping " +
- "whole row groups in which the filter rejects every row. A row group
with a surviving " +
- "row then reads its key columns twice, and the post-scan filter
evaluates the filter " +
- "on each of its rows. A file without a page index falls back the same
" +
- "way. It gains only where a whole row group has no surviving row, a
bloom's false " +
- "positives included, and every other row group reads its key columns
twice. A read " +
- "with pushed data filters already relies on the page index, and that
conf turns it off " +
- "there too. A read with no pushed data filter relies on it only with
this feature. " +
- "Note that the surviving key values of a whole row group are buffered
before that row " +
- "group's first batch is produced. Each reader holds them and their row
ranges for one " +
- "row group at a time, up to " +
- s"${PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key}.
Like the reader's " +
- "own column vectors, they are not tracked by Spark's memory manager.")
+ "bloom filter from join runtime filtering while it reads, and skip the
pages and row " +
+ "groups of the other columns that hold no matching rows. The filter
also stays in the " +
+ "post-scan filter, so the answer does not change, with two exceptions.
Under " +
+ "spark.sql.files.ignoreCorruptFiles the rows returned from a corrupt
file can differ. A " +
+ "file with a wrong Parquet page index can return wrong rows. Setting "
+
+ "parquet.filter.columnindex.enabled to false stops relying on the page
index. Without a " +
Review Comment:
**1. conf doc drops the side effect of `parquet.filter.columnindex.enabled`**
The doc now offers `parquet.filter.columnindex.enabled=false` as the way to
stop relying on the page index, but the previous text also said what else it
turns off:
> A read with pushed data filters already relies on the page index, and that
conf turns it off there too.
That still holds. The conf is parquet's own
(`readOptions.useColumnIndexFilter()`), so setting it for the session also
disables column-index filtering for every ordinary pushed data filter on every
Parquet scan, not only for this feature. A user who follows this doc to work
around one bad file can slow down unrelated queries without knowing why.
Could we keep one clause, e.g. "..., which also turns off page skipping for
pushed data filters"?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -30,30 +30,22 @@ import org.apache.spark.util.Utils
/**
* The SQL metrics the reader updates while applying a
[[ParquetStorageFilter]], created by
- * `ParquetFileFormat.storageFilterMetrics`. Every counter is scoped to what
the storage filter
- * saved against a read of the same projection without one. The one cost taken
off is a second
- * key read in a row group that gave splicing up, below. A row group whose
filter was given up
- * reads its key columns twice and counts nothing, so the counters can show a
saving on a scan
- * that read more than it would have with the feature off. Nor do they take
off the offset indexes
- * phase 2 reads for a narrowed row group when no data filter is pushed, which
a plain read does not
- * read. Bytes are the compressed bytes the reader did not ask parquet for.
The file system can
- * still fetch some of them, for example when a vectored read merges nearby
ranges.
+ * `ParquetFileFormat.storageFilterMetrics`. They count only what the filter
saved against a read of
Review Comment:
**2. metrics doc now claims a net saving it does not measure**
"They count only what the filter saved against a read of the same projection
without one" reads as a net saving, but the caveat that qualified it was
dropped:
> Nor do they take off the offset indexes phase 2 reads for a narrowed row
group when no data filter is pushed, which a plain read does not read.
With no pushed data filter, a plain read never touches the offset indexes,
while phase 2's narrowed `readFilteredRowGroup` does, and the counters don't
take that off. So `bytes avoided by storage filter (page filtering)` overstates
the saving on exactly the scans this feature targets. Since the metric is
user-visible in the SQL UI, I'd keep that sentence (it's one line), or at least
not say "only".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -601,25 +514,18 @@ protected boolean loadNextRowGroup() throws IOException {
}
fileReader.setRequestedSchema(spliceCurrentRowGroup ? nonKeyColumns :
requestedColumns);
- // The footer check above catches every file written without offset
indexes. This catch is
- // the backstop for an offset index the footer lists but parquet fails
to read, which empties
- // that block's store, and only that block's, since parquet keeps one
per block. The row group
- // is then read again over `pushedFilterRanges`, which that empty store
has already widened to
- // the whole block, so the retry cannot throw. The next row group is
narrowed as usual.
- //
- // `MissingOffsetIndexException` is parquet's internal API, as are the
`ColumnIndexStore` and
- // `OffsetIndex` that `ParquetBlockAccounting` reads. Revisit both on a
parquet upgrade. If
- // parquet stops throwing this one as it is, the damaged-index test in
- // `ParquetStorageFilterSuite` fails.
+ // An offset index the footer lists but parquet fails to read empties
that block's store.
+ // The row group is then read again over `pushedFilterRanges`, which the
empty store has
+ // widened to the whole block, so the retry cannot throw.
`MissingOffsetIndexException` is
+ // parquet's internal API, and the damaged-index test fails if parquet
stops throwing it.
Review Comment:
**5. parquet-upgrade note no longer covers `ColumnIndexStore` /
`OffsetIndex`**
The previous comment was the one place that said `ColumnIndexStore` and
`OffsetIndex`, which `ParquetBlockAccounting` imports from
`org.apache.parquet.internal.*`, are also parquet internal API, and to revisit
them on a parquet upgrade. Now only `MissingOffsetIndexException` is mentioned,
and `ParquetBlockAccounting` has no such note of its own.
The damaged-index test pins only the exception. Nothing would flag a change
to the store/offset-index API that silently miscounts the byte metrics. Could
we keep "as are the `ColumnIndexStore` and `OffsetIndex` that
`ParquetBlockAccounting` reads", here or in `ParquetBlockAccounting`'s class
javadoc?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -303,10 +303,9 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
// Filters on non-partition columns.
def dataFilters: Seq[Expression]
- // Filters the storage layer evaluates to prune value-column IO based on
key-column evaluation.
- // These may reference subqueries (e.g. a runtime bloom filter built from a
join build side),
- // which are materialized on the driver while the RDD is built, before the
reader is serialized.
- // Nil for a scan that does not support pushing them.
+ // Filters the storage layer evaluates to skip reading the columns they do
not reference, see
Review Comment:
**3. storage filters do not skip the unreferenced columns, only their
rejected rows**
"to skip reading the columns they do not reference" isn't accurate: phase 2
always reads those columns for the surviving rows. Only the rejected rows'
pages are skipped. The same wording is in
`FileSourceStrategy.storageFiltersFor` (line 155).
`FileFormat.buildReaderWithStorageFilters` in this PR says it correctly: "to
skip reading the other columns of the rows they reject".
Suggest using that wording in both places.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -523,11 +517,9 @@ protected final void installRowGroup(PageReadStore pages,
RowRanges rowRanges)
}
/**
- * Drops the column readers of the row group installed last, and the
row-index iterator, both of
- * which hold the row ranges that row group was read over. For a subclass
that does work of its
- * own between two row groups, so that work does not run with the previous
ranges still held.
- * Call it only from {@link #loadNextRowGroup}, once the row group is fully
emitted. A batch read
- * after it needs another row group installed first.
+ * Drops the column readers and the row-index iterator of the row group
installed last, for a
Review Comment:
**8. `releaseRowGroupReaders` lost its post-call precondition**
This protected method lost its post-call contract:
> A batch read after it needs another row group installed first.
After this call every leaf's column reader is null, so a subclass that calls
`nextBatch()`/`readBatch` before `installRowGroup` fails with an NPE. "Call it
only from `loadNextRowGroup`" says when to call it, not what must happen
afterwards. I'd keep that one sentence, since this is an extension point.
##########
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:
**10. format-agnostic comment now names a bloom-specific hash**
Nit: "cannot hoist a shared hash ahead of the skip" ties this
format-agnostic planner code to the bloom implementation. The previous "what
the conjunct shares with another one" holds for any storage filter. Since the
`If` guards against hoisting any shared subexpression, I'd keep the general
wording so a future non-bloom storage filter doesn't read this as bloom-only.
##########
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:
**9. `preparedPredicate` doc dropped why it is handed out, not wrapped**
The doc dropped why this is handed out rather than wrapped: reaching it
through the `lazy val` is a volatile read, so a per-row caller must resolve it
once. The reason now lives only at the one call site in `evaluateStorageFilter`
("Out of the loop, since resolving it is a volatile read").
That leaves nothing on this class to stop a convenience `eval(row)` wrapper,
or a new caller reading it per row. One clause here would keep the contract
with the member.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -854,25 +717,20 @@ private RowRanges evaluateStorageFilter(
if (currentKeyAccumulators == null) allocateKeyAccumulators();
splicedBytes += appendSurvivorRowToAccumulators(r);
} catch (RuntimeException e) {
- // The buffer is this reader's own and optional, so failing to
fill it gives splicing up
- // rather than the filter, and the memory goes back at once. What
fails here:
- // - a survivor vector that cannot grow;
- // - a key value the predicate did not read on this row, such as
`b` in
- // `coalesce(a, b)`, since the copy decodes every key;
- // - a bug of ours, which only the WARN then shows.
- // Phase 2 reads the key columns the way a plain read does, so a
value that cannot be
- // decoded fails the read only where a plain read fails.
+ // The buffer is optional, so failing to fill it gives splicing
up, not the filter. That
Review Comment:
**7. splicing catch dropped why absorbing a decode error is safe**
Thanks for restoring "a bug of ours" here. The other dropped sentence is the
one that makes absorbing a decode error safe:
> Phase 2 reads the key columns the way a plain read does, so a value that
cannot be decoded fails the read only where a plain read fails.
The catch swallows a `RuntimeException` from decoding a key the predicate
did not read (`b` in `coalesce(a, b)`). That is correct only because of this
property of phase 2, and the comment should name what it depends on.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -814,16 +807,11 @@ case class FileSourceScanExec(
readRDD
}
- // Materialize scalar subqueries inside storage filters to literals and bind
AttributeReferences
- // to BoundReferences targeting positions in `requiredSchema`. Subqueries
must have been prepared
- // by SparkPlan before this is forced (same contract as `pushedDownFilters`).
- //
- // No conf check here, since the conf decides at planning time whether a
scan is offered storage
- // filters at all, and re-reading it now could only make this scan drop work
it already has.
- //
+ // The storage filters with their scalar subqueries materialized, bound to
`requiredSchema`.
Review Comment:
**6. dropped why `preparedStorageFilters` does not re-check the conf**
This dropped a design invariant rather than narrative:
> No conf check here, since the conf decides at planning time whether a scan
is offered storage filters at all, and re-reading it now could only make this
scan drop work it already has.
The public doc also lost "This is a planning-time decision". Without either,
a missing conf check here looks like an oversight, and someone may add one.
That would make the executed scan differ from the plan, whose explain output
still shows `StorageFilters`. I'd keep the sentence.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,19 @@ object SQLConf {
val PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED =
buildConf("spark.sql.parquet.storageFilterPushdown.enabled")
.doc("If true, the vectorized reader of the V1 Parquet file source may
apply a runtime " +
- "storage filter, such as a bloom filter from join runtime filtering,
while it reads. " +
- "It reads the columns the " +
- "filter needs first, evaluates the filter per row, and then reads the
remaining columns " +
- "only for the rows that survived. This is a planning-time decision. " +
- "A filter that is attached also stays in the post-scan filter, the way
a pushed data " +
- "filter does, so the reader is free to stop applying it wherever doing
so would cost " +
- "more than it saves, and the answer does not change. In a scan that
returns columnar " +
- "batches, the post-scan filter skips it on the rows the reader applied
it to, which the " +
- "reader marks in a column named _tmp_storage_filter_checked. A table
with a column of " +
- "that name is not offered storage filters, and a file holding one
outside the table's " +
- "schema is not supported. A row group where the reader stops applying
the filter reads " +
- "its key columns twice, which is slower than not pushing the filter. "
+
- "Under spark.sql.files.ignoreCorruptFiles the answer can change,
because this reader " +
- "reads different pages in a different order than a plain read. Which
rows survive a " +
- "corrupt page can then differ from a plain read, in either direction.
" +
- "Returning only the surviving rows of a row group relies on the
Parquet page index, so a " +
- "file whose page index is wrong can pair a row's key with another
row's values, or " +
- "return rows the filter rejects. Setting " +
- "parquet.filter.columnindex.enabled to false makes the reader fall
back to skipping " +
- "whole row groups in which the filter rejects every row. A row group
with a surviving " +
- "row then reads its key columns twice, and the post-scan filter
evaluates the filter " +
- "on each of its rows. A file without a page index falls back the same
" +
- "way. It gains only where a whole row group has no surviving row, a
bloom's false " +
- "positives included, and every other row group reads its key columns
twice. A read " +
- "with pushed data filters already relies on the page index, and that
conf turns it off " +
- "there too. A read with no pushed data filter relies on it only with
this feature. " +
- "Note that the surviving key values of a whole row group are buffered
before that row " +
- "group's first batch is produced. Each reader holds them and their row
ranges for one " +
- "row group at a time, up to " +
- s"${PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key}.
Like the reader's " +
- "own column vectors, they are not tracked by Spark's memory manager.")
+ "bloom filter from join runtime filtering while it reads, and skip the
pages and row " +
+ "groups of the other columns that hold no matching rows. The filter
also stays in the " +
+ "post-scan filter, so the answer does not change, with two exceptions.
Under " +
Review Comment:
**4. 'with two exceptions', followed by an unsupported case**
"so the answer does not change, with two exceptions" is followed a few
sentences later by "a file holding one outside the table's schema is not
supported". A reader can't tell whether that is a third way for the answer to
change, or a failure.
For example, if the plain reader is used for such a file (say
spark.sql.parquet.enableVectorizedReader is flipped after planning), it decodes
the file's `_tmp_storage_filter_checked` values and the post-scan Filter trusts
them. Could the doc say what "not supported" means for the user (an error, or
rows not filtered), or drop the count ("with these exceptions")?
--
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]
