peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4258929664
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04):
"..., and 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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04),
with that sentence back.
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04),
with the `FileFormat` wording here and in `storageFiltersFor`.
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04),
by moving the two unsupported cases ahead of "with two exceptions", so only
the exceptions follow it.
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04),
in the same comment.
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04).
I kept it to the code comment.
##########
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:
Done in
[668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04).
##########
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:
I'd leave this one. The first sentence already says it drops the column
readers, so a batch read needs new ones.
--
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]
