You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/10/09 14:15:39 UTC
[PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
peter-toth opened a new pull request, #59324:
URL: https://github.com/apache/spark/pull/59324
### What changes were proposed in this pull request?
This trims the comments and conf docs of #58895 (SPARK-59620) and #59269 (SPARK-60034), as @viirya suggested in https://github.com/apache/spark/pull/58895#issuecomment-6064730845.
The comments keep only the invariants and the non-obvious trade-offs. The narrative goes: alternatives considered, past measurements, and the same point restated across `SQLConf`, `FileFormat`, `ParquetStorageFilter` and the reader.
It also rewrites the doc of `spark.sql.parquet.storageFilterPushdown.enabled` to what changes a user's decision or answer, based on @viirya's text. It now also says that only `ParquetFileFormat` itself is supported, not its subclasses. The doc of the internal `maxSplicedRowGroupBytes` conf is shorter too.
There are no code changes.
### Why are the changes needed?
About 1,000 of the feature's ~2,400 non-test lines were comments, which made the reader hard to maintain. The public conf doc described internals rather than what a user needs to know.
### Does this PR introduce _any_ user-facing change?
Only the doc of `spark.sql.parquet.storageFilterPushdown.enabled`. The feature is unreleased.
### How was this patch tested?
Comments and doc strings only. `ParquetStorageFilterSuite`, `ParquetColumnIndexSuite` and `ParquetVectorizedSuite` pass, and so do scalastyle and checkstyle.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6146121207
@dongjoon-hyun @HyukjinKwon @viirya thank you for the reviews. [668797d4093](https://github.com/apache/spark/pull/59324/commits/668797d4093aea461d2a0bf532ed6b03b583dd04) takes dongjoon's 1 to 7 and 10 and Hyukjin's point, and I replied on 8 and 9. I kept each fix to a qualifier, a reorder or one sentence, so the comments stay short.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6082716289
This is a draft until #59269 lands, since it is built on top of it and trims its comments too. Only the last commit, 76fc348daf0, belongs to this PR. I'll rebase it onto master then.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6196495282
@HyukjinKwon thanks, fixed in [99f8cd24e5b](https://github.com/apache/spark/pull/59324/commits/99f8cd24e5b4b0b7fa5dd50c7af2f32f2755b957): "An `InternalError` from the first or the last case goes out in an `UnknownError`, another fatal error."
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4283192631
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetBlockAccounting.java:
##########
@@ -38,14 +38,8 @@
/**
* What a partial read of a Parquet file transfers, and whether one is possible at all, answered
- * from the footer the {@link ParquetFileReader} already holds. Every method here is metadata
- * arithmetic and causes no IO of its own, which is what lets a reader report these numbers per row
- * group. For a strict subset of a block that holds only once the block's column index store is
- * built, see {@link #compressedBytesForRowRanges}. A walk before that would read the offset indexes
- * itself and cache a store over the reader's current columns.
- *
- * <p>One instance per reader, which reads one split of a file, because the warning below is
- * reported once per instance.
+ * from the metadata the {@link ParquetFileReader} already holds, with no IO of its own. One
Review Comment:
**2. (Non-blocking) The class javadoc still says "with no IO of its own" unconditionally**
The `compressedBytesForRowRanges` doc now says that a walk before the block's store exists "would build and keep one", and building it reads the offset indexes. [Round-2 comment 5](https://github.com/apache/spark/pull/59324#discussion_r4264399069) also pointed at this line, which is unchanged, so the class and method docs now disagree. `hasOffsetIndexes` and a whole-block walk do read only the footer.
Maybe "..., with no IO of its own once the block's store is built, see `{@link #compressedBytesForRowRanges}`."?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped when the executor would report the error as a kill,
+ * see [[throwIfKilled]], since the error may be the kill's interrupt surfacing through a UDF;
Review Comment:
**1. (Non-blocking) The kill bullet now says when a kill is wrapped, but not what happens otherwise**
Thanks for adding the condition ([round-2 comment 6](https://github.com/apache/spark/pull/59324#discussion_r4264399077)). But the case the condition now excludes does not fall back. Under a pending kill, `throwIfKilled(e)` rethrows an error the executor would not report as a kill unchanged (`case other => throw other`). Two kinds of error get there:
- a codec's `InternalError` while decoding the key pages, which `corrupt` lets past the first check;
- a `LinkageError`, e.g. a native codec's `UnsatisfiedLinkError`, which `Utils.isFatalError` does not count as fatal and which is not `NonFatal` either.
The javadoc does not list this case, so it falls under "Anything else falls back". The first one also falls under "an `InternalError` is corrupt data". Without a pending kill, both do fall back. `throwIfKilled`'s own doc already covers it ("a `surfaced` error of any other type goes out as it is"), so this is only about the summary here.
Maybe "a pending task kill goes out wrapped when the executor would report the error as a kill, and as it is otherwise, see [[throwIfKilled]], ..."?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6162821198
@dongjoon-hyun thank you for the review. [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3) takes 1 to 3, 5 to 7 and 9 to 15, and I replied on 4, 8 and the second half of 7.
I also went over every sentence the trim removed or rewrote, in all 13 files. That found two more claims the trim made false, both fixed in the same commit:
- the struct case in `initializeLateMaterialization` also covered a struct the file does not have, which counts as missing by default too;
- `giveUpOnError` said every fatal error is rethrown, while the decode catch absorbs a codec's `InternalError`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #59324: [SPARK-60107][SQL] Trim the storage-filter comments and conf docs
URL: https://github.com/apache/spark/pull/59324
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6196241944
Thanks, @peter-toth. I rechecked the reclaim comment in `ParquetStorageFilter.preparedPredicate` at 0c77f7ddd22: it now says dropping the expression "lets a columnar scan's task reclaim them", and it still notes that an interpreted predicate keeps the expression, so my earlier point about the row-based scan is addressed.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:6146121207","thread_id":"conversation:6146121207","verdict_sha256":"53906f0d378b7b67407aa0d8e2fcd0a464caf019d187fa4094c6b7e08e440891"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4237892654
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -523,75 +446,61 @@ private void initializeLateMaterialization() throws IOException {
}
/**
- * Loads the next row group with rows left to emit, through the three phases of the class javadoc.
- * Phase 2 reads the non-key columns over the survivors while splicing, the whole projection over
- * the survivors once splicing is given up, and the whole projection over
- * {@code pushedFilterRanges} once the filter is. Returns false when the row groups run out, which
- * ends a read the filter narrowed.
+ * Loads the next row group with rows left to emit, through the phases of the class javadoc.
+ * Returns false when the row groups run out.
*/
@Override
protected boolean loadNextRowGroup() throws IOException {
if (keyColumns == null) return super.loadNextRowGroup();
- // The previous row group is fully emitted by the time this is called, so its pages are done
- // with. Released here rather than at the next assignment, so a row group the filter empties,
- // or the end of the file, does not keep the previous one's pages alive. Its column readers go
- // too. They hold the row ranges it was read over, which the budget no longer counts, and the
- // decoders of their last page.
+ // The previous row group is fully emitted, so its pages and column readers go now rather than
+ // at the next assignment, which a row group the filter empties would delay.
closeDataPages();
releaseRowGroupReaders();
StorageFilterMetrics m = storageFilter.metrics();
List<BlockMetaData> blocks = fileReader.getRowGroups();
while (nextBlockIndex < blocks.size()) {
- // A plain read goes back to `FileScanRDD`, which checks for a task kill, after every batch.
- // This loop can pass any number of row groups the filter empties without emitting one.
+ // A plain read checks for a task kill after every batch, and this loop can pass any number
+ // of row groups without emitting one.
ParquetStorageFilter.throwIfKilled();
int blockIdx = nextBlockIndex++;
long blockRowCount = blocks.get(blockIdx).getRowCount();
- // Phase 0: rows the pushed data filter allows, by column index. The full projection goes back
- // on first, because `getRowRanges` builds an unseen block's store over the current paths.
+ // Phase 0. The full projection goes back on first, because `getRowRanges` builds an unseen
+ // block's store over the current paths.
fileReader.setRequestedSchema(requestedColumns);
RowRanges pushedFilterRanges = pushedFilterRangesFor(blockIdx, blockRowCount);
- // RowRanges.rowCount() walks every range, so resolve each range set's count once.
long baselineRows = pushedFilterRanges.rowCount();
if (baselineRows == 0) {
- // An empty block, or one the pushed data filter rejects whole. Not this filter's skip, so
- // no storage-filter metric counts it.
+ // Not this filter's skip, so no storage-filter metric counts it.
continue;
}
- // Whether phase 2 can read part of this row group. That needs an offset index for every
- // column of the block's column index store, and parquet empties the store if one column
- // lacks it. With a data filter pushed, the store is built at initialize over the whole
- // projection, so every projected column is checked. Files written before parquet-mr 1.11,
- // or by pyarrow's `write_table` by default, have no offset index at all. With no data filter
- // pushed the check is stricter than needed, since the store would cover only the non-key
- // columns. False means nothing is buffered and the filter can only skip the row group whole.
- // Phase 2 then reads the whole projection over `pushedFilterRanges`. Parquet narrows those
- // only by an index it could read, so that read cannot throw.
+ // Phase 2 can read part of the row group only with an offset index for every column of the
+ // block's store, since parquet empties the store when one column lacks it. Files written by
+ // parquet-mr before 1.11, or by pyarrow's `write_table` by default, have none. Otherwise the
Review Comment:
**Nit (P3):** 'Otherwise' comes right after the sentence naming the writers whose files have no offset index, so it reads as 'for files from any other writer, the filter can only skip the row group whole'. The code says the opposite. `canNarrowRowGroup` is `!pageIndexDisabled && blockAccounting.hasOffsetIndexes(...)`, so whole-row-group skipping is what a row group *without* an offset index gets. The previous comment avoided this by naming the condition ('False means nothing is buffered and ...'). Something like 'Without one, the filter can only skip the row group whole.' would fix the polarity and keep the trim.
**Verification:**
- **Inspection:** The reworded comment's condition matches canNarrowRowGroup = !pageIndexDisabled && hasOffsetIndexes(...): only a row group lacking the index is limited to whole-row-group skipping. Only comment text changes, and the edited lines stay within 100 characters.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4246401641
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -523,75 +446,61 @@ private void initializeLateMaterialization() throws IOException {
}
/**
- * Loads the next row group with rows left to emit, through the three phases of the class javadoc.
- * Phase 2 reads the non-key columns over the survivors while splicing, the whole projection over
- * the survivors once splicing is given up, and the whole projection over
- * {@code pushedFilterRanges} once the filter is. Returns false when the row groups run out, which
- * ends a read the filter narrowed.
+ * Loads the next row group with rows left to emit, through the phases of the class javadoc.
+ * Returns false when the row groups run out.
*/
@Override
protected boolean loadNextRowGroup() throws IOException {
if (keyColumns == null) return super.loadNextRowGroup();
- // The previous row group is fully emitted by the time this is called, so its pages are done
- // with. Released here rather than at the next assignment, so a row group the filter empties,
- // or the end of the file, does not keep the previous one's pages alive. Its column readers go
- // too. They hold the row ranges it was read over, which the budget no longer counts, and the
- // decoders of their last page.
+ // The previous row group is fully emitted, so its pages and column readers go now rather than
+ // at the next assignment, which a row group the filter empties would delay.
closeDataPages();
releaseRowGroupReaders();
StorageFilterMetrics m = storageFilter.metrics();
List<BlockMetaData> blocks = fileReader.getRowGroups();
while (nextBlockIndex < blocks.size()) {
- // A plain read goes back to `FileScanRDD`, which checks for a task kill, after every batch.
- // This loop can pass any number of row groups the filter empties without emitting one.
+ // A plain read checks for a task kill after every batch, and this loop can pass any number
+ // of row groups without emitting one.
ParquetStorageFilter.throwIfKilled();
int blockIdx = nextBlockIndex++;
long blockRowCount = blocks.get(blockIdx).getRowCount();
- // Phase 0: rows the pushed data filter allows, by column index. The full projection goes back
- // on first, because `getRowRanges` builds an unseen block's store over the current paths.
+ // Phase 0. The full projection goes back on first, because `getRowRanges` builds an unseen
+ // block's store over the current paths.
fileReader.setRequestedSchema(requestedColumns);
RowRanges pushedFilterRanges = pushedFilterRangesFor(blockIdx, blockRowCount);
- // RowRanges.rowCount() walks every range, so resolve each range set's count once.
long baselineRows = pushedFilterRanges.rowCount();
if (baselineRows == 0) {
- // An empty block, or one the pushed data filter rejects whole. Not this filter's skip, so
- // no storage-filter metric counts it.
+ // Not this filter's skip, so no storage-filter metric counts it.
continue;
}
- // Whether phase 2 can read part of this row group. That needs an offset index for every
- // column of the block's column index store, and parquet empties the store if one column
- // lacks it. With a data filter pushed, the store is built at initialize over the whole
- // projection, so every projected column is checked. Files written before parquet-mr 1.11,
- // or by pyarrow's `write_table` by default, have no offset index at all. With no data filter
- // pushed the check is stricter than needed, since the store would cover only the non-key
- // columns. False means nothing is buffered and the filter can only skip the row group whole.
- // Phase 2 then reads the whole projection over `pushedFilterRanges`. Parquet narrows those
- // only by an index it could read, so that read cannot throw.
+ // Phase 2 can read part of the row group only with an offset index for every column of the
+ // block's store, since parquet empties the store when one column lacks it. Files written by
+ // parquet-mr before 1.11, or by pyarrow's `write_table` by default, have none. Otherwise the
Review Comment:
Thanks, fixed in [0eb44e78ba1](https://github.com/apache/spark/pull/59324/commits/0eb44e78ba139cfaf181e82d29fda059f381af44): "Without one, the filter can only skip the row group whole."
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4267722238
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -190,11 +151,10 @@ public class LateMaterializationParquetRecordReader extends VectorizedParquetRec
/**
* Whether {@code parquet.filter.columnindex.enabled} is false, so this file's page index must not
- * be used. That turns off phase 0 and every partial read. A wrong index would pair a row's key
- * with another row's values, or return rows the filter rejects once splicing is given up. The
- * post-scan Filter catches neither, since it takes the reader's mark on those rows. Parquet's
- * partial read does not check the conf itself. Read off {@link #readOptions}, so it agrees with
- * the row count {@link #fileReader} reported.
+ * be used. That turns off phase 0 and every partial read, which parquet does not check itself.
+ * It is the way out of a wrong page index, which would pair a row's key with another row's
+ * values, and which the post-scan Filter cannot catch on the rows this reader marks checked.
+ * Read off {@link #readOptions}, so it agrees with {@link #fileReader}.
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3): `so it agrees with the row count {@link #fileReader} reported.` I left the splicing case out. The comment gives one example of what a wrong index does, and the conf doc covers wrong rows in general.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetBlockAccounting.java:
##########
@@ -38,14 +38,8 @@
/**
* What a partial read of a Parquet file transfers, and whether one is possible at all, answered
- * from the footer the {@link ParquetFileReader} already holds. Every method here is metadata
- * arithmetic and causes no IO of its own, which is what lets a reader report these numbers per row
- * group. For a strict subset of a block that holds only once the block's column index store is
- * built, see {@link #compressedBytesForRowRanges}. A walk before that would read the offset indexes
- * itself and cache a store over the reader's current columns.
- *
- * <p>One instance per reader, which reads one split of a file, because the warning below is
- * reported once per instance.
+ * from the metadata the {@link ParquetFileReader} already holds, with no IO of its own. One
Review Comment:
Done in [0c77f7ddd22](https://github.com/apache/spark/pull/59324/commits/0c77f7ddd22e4f70db4072d4db39d832ed53dbe7): `with no IO of its own once the block's store is built, see {@link #compressedBytesForRowRanges}`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4267720887
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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 " +
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3): "A file with a wrong Parquet page index can return wrong rows, even with no pushed data filter."
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
+ "turns off page skipping for pushed data filters. Without a page index, as in the files " +
+ "pyarrow's write_table writes by default, or with that conf set to false, the reader can " +
+ "only skip whole row groups, and it reads the key columns twice in the others. It can be " +
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3): "With a page index, it can be slower when ...".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
+ "turns off page skipping for pushed data filters. Without a page index, as in the files " +
+ "pyarrow's write_table writes by default, or with that conf set to false, the reader can " +
+ "only skip whole row groups, and it reads the key columns twice in the others. It can be " +
+ "slower when the matching rows are spread across most pages. The matching key values of " +
+ "a row group are buffered while it is read, outside Spark's memory manager.")
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3), with the bound as viirya's text had it: "..., up to spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes, outside Spark's memory manager."
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
Review Comment:
I'd leave this one. It is parquet's Hadoop key, and where such a key can be set is the same for every one of them, so it is not this conf's to explain.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetBlockAccounting.java:
##########
@@ -84,29 +77,14 @@ boolean hasOffsetIndexes(int blockIndex, Set<ColumnPath> paths) {
}
/**
- * Compressed bytes a reader transfers for the leaf columns named by {@code paths} when it reads
- * exactly {@code rowRanges} of the given block. Page headers and the dictionary page are
- * included, since both are read whenever any page of a chunk is read. {@code rowRangeCount} is
- * {@code rowRanges.rowCount()}, passed in because that walks every range and the caller has it.
- *
- * <p>Two sources, chosen so this never causes IO of its own:
- * <ul>
- * <li>{@code rowRanges} covers the whole block, and the answer is the sum of the chunks'
- * {@code getTotalSize()}, which is already in the footer. This is the case that matters.
- * Whenever nothing else has built the block's {@link ColumnIndexStore}, {@code rowRanges}
- * is necessarily the whole block, because a narrower range can only come from column-index
- * filtering, which builds the store as a side effect.
- * <li>{@code rowRanges} is a strict subset, and the walk goes over the offset index, as
- * parquet's own read path does, adding the dictionary page the way
- * {@code calculateOffsetRanges} does. The store is guaranteed to exist here, so the walk is
- * pure metadata arithmetic. For ranges a reader narrowed itself, which column-index
- * filtering had no hand in, that guarantee is an ordering one, since the reader's own read
- * of those ranges built the store first. A new caller has to keep that order.
- * </ul>
+ * Compressed bytes a reader transfers for the columns in {@code paths} when it reads exactly
+ * {@code rowRanges} of the block, page headers and the dictionary page included.
+ * {@code rowRangeCount} is {@code rowRanges.rowCount()}, which the caller has.
*
- * <p>Columns absent from this physical file (schema evolution) contribute nothing, which is
- * correct, since the reader transfers nothing for them. They fall out of the walk by themselves,
- * which goes over the block's chunks rather than over {@code paths}.
+ * <p>A whole block is the chunks' sizes from the footer. A strict subset is walked over the
+ * offset index, as parquet's read does. That reads the block's {@link ColumnIndexStore}, which
+ * costs no IO only once a read of those ranges has built it, so a new caller has to keep that
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3). It now says which reads build the store, and that a walk before them breaks a later read of other columns.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6095715524
Rebased onto master now that #59269 is in, so only the cleanup is left. @viirya, this is the comment and conf doc cleanup you suggested on #58895.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4267722003
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped, see [[throwIfKilled]], since the error may be the
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3): "... goes out wrapped when the executor would report the error as a kill, see [[throwIfKilled]] ...".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -190,11 +151,10 @@ public class LateMaterializationParquetRecordReader extends VectorizedParquetRec
/**
* Whether {@code parquet.filter.columnindex.enabled} is false, so this file's page index must not
- * be used. That turns off phase 0 and every partial read. A wrong index would pair a row's key
- * with another row's values, or return rows the filter rejects once splicing is given up. The
- * post-scan Filter catches neither, since it takes the reader's mark on those rows. Parquet's
- * partial read does not check the conf itself. Read off {@link #readOptions}, so it agrees with
- * the row count {@link #fileReader} reported.
+ * be used. That turns off phase 0 and every partial read, which parquet does not check itself.
+ * It is the way out of a wrong page index, which would pair a row's key with another row's
+ * values, and which the post-scan Filter cannot catch on the rows this reader marks checked.
+ * Read off {@link #readOptions}, so it agrees with {@link #fileReader}.
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3): "so it agrees with the row count {@link #fileReader} reported." I left the splicing case out. The comment gives one example of what a wrong index does, and the conf doc covers wrong rows in general.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -117,25 +98,17 @@ public class LateMaterializationParquetRecordReader extends VectorizedParquetRec
SparkLoggerFactory.getLogger(LateMaterializationParquetRecordReader.class);
/**
- * What one buffered variable-length value costs the vector that holds it, as a multiple of the
- * value's own bytes. Those bytes go in the byte child, with one null byte per element of that
- * child, and up to as much again because {@link WritableColumnVector#reserve} doubles the child
- * when it grows, by default. A factor rather than an exact figure, so with that default this
- * half of the charge over-counts rather than under-counts, since nothing above this buffer can
- * spill it. A {@code hugeVectorReserveRatio} above 2 makes it under-count for a child past
- * {@code hugeVectorThreshold}. That threshold is off by default. A fixed-width key needs no
- * factor, since its accumulator is allocated at capacity and never grows.
+ * What one buffered variable-length value costs, as a multiple of its own bytes. The byte child
+ * holds a null byte per byte, and {@link WritableColumnVector#reserve} doubles it when it grows.
+ * It over-counts on purpose, since nothing can spill this buffer. That holds at the default
+ * growth. A {@code hugeVectorReserveRatio} above 2 makes it under-count past
+ * {@code hugeVectorThreshold}, which is off by default.
*/
private static final long VARIABLE_LENGTH_BYTES_FACTOR = 4L;
/**
- * What one surviving row range costs to hold, which is the half of the budget a row group
- * buffering nothing still pays. Scattered survivors make one range each, and phase 2 needs the
- * whole set to select its pages. One set is held, which every column reader of the row group
- * walks with a cursor of its own. The figure is parquet's {@code RowRanges.Range}, two longs,
- * plus its object header and a list slot. That is what is held. At the end of phase 1 the
- * builder's list can have up to half as many slots again, and {@code build()} copies it, so the
- * peak is somewhat higher.
+ * What one surviving row range costs to hold, parquet's {@code RowRanges.Range} with its object
+ * header and list slot. Scattered survivors make one range each.
Review Comment:
I'd leave this one. It guards against a change to how the readers take the ranges, rather than explaining the current code.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -523,75 +450,61 @@ private void initializeLateMaterialization() throws IOException {
}
/**
- * Loads the next row group with rows left to emit, through the three phases of the class javadoc.
- * Phase 2 reads the non-key columns over the survivors while splicing, the whole projection over
- * the survivors once splicing is given up, and the whole projection over
- * {@code pushedFilterRanges} once the filter is. Returns false when the row groups run out, which
- * ends a read the filter narrowed.
+ * Loads the next row group with rows left to emit, through the phases of the class javadoc.
+ * Returns false when the row groups run out.
*/
@Override
protected boolean loadNextRowGroup() throws IOException {
if (keyColumns == null) return super.loadNextRowGroup();
- // The previous row group is fully emitted by the time this is called, so its pages are done
- // with. Released here rather than at the next assignment, so a row group the filter empties,
- // or the end of the file, does not keep the previous one's pages alive. Its column readers go
- // too. They hold the row ranges it was read over, which the budget no longer counts, and the
- // decoders of their last page.
+ // The previous row group is fully emitted, so its pages and column readers go now rather than
+ // at the next assignment, which a row group the filter empties would delay.
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3), with the reason that holds for every row group: "The next assignment comes only after the next row group's phase 1, whose budget does not count them."
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -57,56 +57,37 @@
/**
* A {@link VectorizedParquetRecordReader} that reads a row group in phases under a
* {@code ParquetStorageFilter}, so the projected columns of the rows the filter rejects are never
- * read. The filter is a deterministic conjunct of the post-scan {@code Filter}, which still holds
- * it, so any phase below may stop short and read the file the way a plain scan would.
+ * read. The post-scan {@code Filter} still holds the filter, so any phase may stop short and read
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3), with the same wording.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -477,14 +409,10 @@ private void initializeLateMaterialization() throws IOException {
return;
}
- // The non-key columns phase 2 reads. If this file has none of them, the filter saves nothing,
- // since phase 1 already reads every column the file has for every row. The reader then
- // declines it. The planner already declines a projection of key columns alone, so this catches
- // a file missing every projected non-key leaf, which also covers the row-index column. A struct
- // the file has, but none of whose requested fields it has, counts only under the legacy
- // `returnNullStructIfAllFieldsMissing`, since otherwise the clipped schema reads one of its
- // other fields to tell a null struct from one whose requested fields are all null. The checked
- // column is left out too, since a file column of its name is never decoded.
+ // The non-key columns phase 2 reads. A file with none of them gains nothing from the filter,
+ // so the reader declines it. A struct none of whose requested fields the file has counts as
+ // missing only under the legacy `returnNullStructIfAllFieldsMissing`, since otherwise the
+ // clipped schema reads one of its other fields. The checked column never counts.
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped, see [[throwIfKilled]], since the error may be the
+ * kill's interrupt surfacing through a UDF;
+ * - a fatal error deeper in the cause chain, as deep as the executor looks, goes out as itself.
*
- * An `InternalError` among those goes out in an `UnknownError`, a fatal error of another type.
- * `FileScanRDD` would take a bare `InternalError` from a reader for a corrupt file under
- * `ignoreCorruptFiles`. A `NonFatal` wrapper does not work either. With a kill pending, the
- * executor reports one as killed and keeps running. A plain plan raises it from the post-scan
- * Filter, where neither happens, and the executor stops. It stops on the `UnknownError` too, at
- * any `spark.executor.killOnFatalError.depth` of 1 or more, since that is at the top.
+ * An `InternalError` among those goes out in an `UnknownError`, another fatal error. A bare one
+ * would be taken for a corrupt file under `ignoreCorruptFiles`, and a `NonFatal` wrapper would
+ * let the executor run on under a pending kill.
*
* An error from decoding differs in three ways:
- * - a plain read raises it from the reader, where `FileScanRDD` wraps a `NonFatal` one in
- * FAILED_READ_FILE before the executor looks. So a fatal error deeper in the chain of a
- * `NonFatal` one has to sit one level shallower to count, while any other error is searched
- * as deep as the executor does. Under `ignoreCorruptFiles` a plain read may skip the rest of
- * the file on it instead.
+ * - `FileScanRDD` wraps a `NonFatal` one before the executor looks, so a fatal cause counts one
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3).
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -57,56 +57,37 @@
/**
* A {@link VectorizedParquetRecordReader} that reads a row group in phases under a
* {@code ParquetStorageFilter}, so the projected columns of the rows the filter rejects are never
- * read. The filter is a deterministic conjunct of the post-scan {@code Filter}, which still holds
- * it, so any phase below may stop short and read the file the way a plain scan would.
+ * read. The post-scan {@code Filter} still holds the filter, so any phase may stop short and read
+ * the rows the way a plain scan would.
*
- * <p>Per row group, all three phases driven by the one {@link #fileReader}:
+ * <p>Per row group:
* <ul>
- * <li>phase 0 asks the column index which rows the pushed data filter allows, metadata only;</li>
- * <li>phase 1 reads the key columns the filter references over those rows, evaluates the filter
- * per row, and buffers the surviving key values;</li>
- * <li>phase 2 reads the remaining projected columns over the surviving row ranges.</li>
+ * <li>phase 0 asks the column index which rows the pushed data filter allows;</li>
+ * <li>phase 1 reads the key columns over those rows, evaluates the filter, and buffers the
+ * surviving key values;</li>
+ * <li>phase 2 reads the other projected columns over the surviving row ranges.</li>
* </ul>
+ * The emit path splices the buffered key values into the batch, so the key columns are read once.
+ * A row group whose buffer passes {@code maxSplicedRowGroupBytes} gives splicing up, and phase 2
+ * reads its key columns again. If the row ranges alone pass it, the filter is given up too.
*
- * <p>The emit path then splices the buffered key vectors into the batch, so phase 2 never reads a
- * key column again. A row group whose buffer has passed
- * {@code spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes} gives splicing up, and
- * phase 2 reads its key columns again with the rest. If the survivors' row ranges alone pass it,
- * the filter is given up for that row group too.
- *
- * <p>A whole file is read without splicing in these cases:
+ * <p>Some files and row groups get less:
* <ul>
- * <li>every key column is missing from it under schema evolution. The predicate is then constant
- * over the file and is evaluated once, against the constants the reader materializes for
- * those columns. True keeps the file unfiltered, false skips it;</li>
- * <li>the file has none of the projected non-key columns, or none is projected. Phase 2 would
- * have nothing to read, so the reader declines the filter and reads the file plainly;</li>
- * <li>{@code parquet.filter.columnindex.enabled} is false, so its page index is not used. The
- * filter then only skips row groups it rejects whole.</li>
+ * <li>a file with none of the key columns has a constant predicate, evaluated once. False skips
+ * the file, and true reads it whole;</li>
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4264399042
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
+ "turns off page skipping for pushed data filters. Without a page index, as in the files " +
+ "pyarrow's write_table writes by default, or with that conf set to false, the reader can " +
+ "only skip whole row groups, and it reads the key columns twice in the others. It can be " +
Review Comment:
**2. After the reorder, "It can be slower" binds to the no-page-index case**
Moving the two unsupported cases up for round-1 comment 4 put this sentence right after the no-page-index one:
> ... the reader can only skip whole row groups, and it reads the key columns twice in the others. It can be slower when the matching rows are spread across most pages.
"It" now reads as the reader without a page index. There, how the matches spread across pages does not matter, since the first match in a row group gives the filter up (`if (!canNarrowRowGroup) return null`). The caveat is about the page-indexed case: when the matches touch most pages, phase 2 skips almost nothing, while phase 1 still pays the per-row evaluation and the splicing. A user whose files have a page index (parquet-mr 1.11 and later) would read the only general performance caveat as not applying to them.
Maybe "Even with a page index, it can be slower when ..."? Moving the sentence instead would leave the buffering sentence right after the no-page-index one, a case where nothing is buffered.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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 " +
Review Comment:
**1. The conf doc again drops that bloom-only scans rely on the page index only with this feature**
Round-1 comment 1 quoted only the first half of this pair from the base, and that half is back. The second half is still gone:
> A read with no pushed data filter relies on it only with this feature.
That sentence was the agreed doc fix in #58895, twice: for [the OffsetIndex thread](https://github.com/apache/spark/pull/58895#discussion_r4108708237), and again in [round 7](https://github.com/apache/spark/pull/58895#discussion_r4148902141), where it was restored after the round-6 rewrite had dropped it.
It still holds. A plain read with no pushed data filter, or only an inferred `IS NOT NULL` that prunes no page, never consults the offset index, since `readNextFilteredRowGroup` falls back to `readNextRowGroup`. With this feature, phase 2 reads the survivors through the offset index (`readFilteredRowGroup(blockIdx, finalRanges)`) and marks them checked. So a file whose page index is wrong can start returning wrong rows from a query that has always been right. Right after "also turns off page skipping for pushed data filters", the hazard reads as one that the user's existing reads already carry, so that user has no reason to apply the workaround.
Could we add it back, e.g. "A read with no pushed data filter relies on the page index only with this feature."?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
+ "turns off page skipping for pushed data filters. Without a page index, as in the files " +
+ "pyarrow's write_table writes by default, or with that conf set to false, the reader can " +
+ "only skip whole row groups, and it reads the key columns twice in the others. It can be " +
+ "slower when the matching rows are spread across most pages. The matching key values of " +
+ "a row group are buffered while it is read, outside Spark's memory manager.")
Review Comment:
**3. The conf doc no longer bounds the buffer it says is outside the memory manager**
This doc is based on [viirya's text](https://github.com/apache/spark/pull/58895#issuecomment-6064730845), which ends "Matching key values are buffered per row group, up to spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes, outside Spark's memory manager." The bound was dropped here, and so was the base's "before that row group's first batch is produced".
`evaluateStorageFilter` buffers every survivor of a row group, and its row ranges, before `installRowGroup`, and gives splicing up only past `splicedBytes + rangeBytes > cap`, 64MB by default. That conf is internal, so `SET -v` and the generated docs show neither the knob nor its default. "buffered while it is read" sounds like a batch-sized buffer, but it can be 64MB per running task, which comes out of `spark.executor.memoryOverhead` when off heap.
Could we keep the bound, e.g. "The matching key values of a whole row group are buffered before its first batch, up to 64MB per reader by default, outside Spark's memory manager."? Naming the internal conf, as the base did, would work too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1976,37 +1969,20 @@ 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. Only ParquetFileFormat itself " +
+ "is supported, not its subclasses. A table with a column named " +
+ "_tmp_storage_filter_checked is not offered storage filters, and a file holding one " +
+ "outside the table's schema is not supported. 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, and also " +
Review Comment:
**4. The doc does not say where `parquet.filter.columnindex.enabled` has to be set**
This is parquet's Hadoop key, not a Spark conf. The configuration docs say runtime SQL confs can be set in the config file or with `--conf`, but both drop a key that does not start with `spark.` (`SparkSubmitArguments.ignoreNonSparkProperties`, with only a WARN). A user who sets it there keeps reading the bad page index.
It works with `SET` or `spark.conf.set`, as a read option, which also keeps the side effect on pushed data filters to one scan, or at launch as `spark.hadoop.parquet.filter.columnindex.enabled`. Other public docs say where such keys go, e.g. `parquet.compression` "in the table-specific options/properties". Maybe one clause, e.g. "..., in the session or as a read option, ..."?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped, see [[throwIfKilled]], since the error may be the
Review Comment:
**6. The kill bullet lost its condition**
> a pending task kill goes out wrapped, see [[throwIfKilled]], ...
and below, "An `InternalError` among those goes out in an `UnknownError`". Neither holds for an error the executor's kill case does not take. With `fromRead = true`, a codec's `InternalError` under a pending kill skips L249 as `corrupt`, and `throwIfKilled(e)` rethrows it bare at L291 (`case other => throw other`), as a plain read would. The base said "goes out wrapped when the error is one the executor's kill case takes", and `throwIfKilled`'s doc gave a codec's `InternalError` and a native codec's `UnsatisfiedLinkError` as examples. Aligning the code with the current text, wrapping always, would let the executor report a fatal error as a kill and run on, which L228-229 rule out.
Maybe "a pending task kill goes out wrapped when the executor's kill case takes the error, see [[throwIfKilled]], ..."?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -117,25 +98,17 @@ public class LateMaterializationParquetRecordReader extends VectorizedParquetRec
SparkLoggerFactory.getLogger(LateMaterializationParquetRecordReader.class);
/**
- * What one buffered variable-length value costs the vector that holds it, as a multiple of the
- * value's own bytes. Those bytes go in the byte child, with one null byte per element of that
- * child, and up to as much again because {@link WritableColumnVector#reserve} doubles the child
- * when it grows, by default. A factor rather than an exact figure, so with that default this
- * half of the charge over-counts rather than under-counts, since nothing above this buffer can
- * spill it. A {@code hugeVectorReserveRatio} above 2 makes it under-count for a child past
- * {@code hugeVectorThreshold}. That threshold is off by default. A fixed-width key needs no
- * factor, since its accumulator is allocated at capacity and never grows.
+ * What one buffered variable-length value costs, as a multiple of its own bytes. The byte child
+ * holds a null byte per byte, and {@link WritableColumnVector#reserve} doubles it when it grows.
+ * It over-counts on purpose, since nothing can spill this buffer. That holds at the default
+ * growth. A {@code hugeVectorReserveRatio} above 2 makes it under-count past
+ * {@code hugeVectorThreshold}, which is off by default.
*/
private static final long VARIABLE_LENGTH_BYTES_FACTOR = 4L;
/**
- * What one surviving row range costs to hold, which is the half of the budget a row group
- * buffering nothing still pays. Scattered survivors make one range each, and phase 2 needs the
- * whole set to select its pages. One set is held, which every column reader of the row group
- * walks with a cursor of its own. The figure is parquet's {@code RowRanges.Range}, two longs,
- * plus its object header and a list slot. That is what is held. At the end of phase 1 the
- * builder's list can have up to half as many slots again, and {@code build()} copies it, so the
- * peak is somewhat higher.
+ * What one surviving row range costs to hold, parquet's {@code RowRanges.Range} with its object
+ * header and list slot. Scattered survivors make one range each.
Review Comment:
**8. The range charge depends on one shared range list, which no comment says anymore**
`ROW_RANGE_BYTES` is charged once per range (L715). That is right only because every column reader of the row group walks the one `RowRanges` list: `RangeListState` takes `rowRanges.getRanges()`, which parquet returns without copying. The base said so in three places, and all three are gone:
- here: "One set is held, which every column reader of the row group walks with a cursor of its own."
- at L558: "rather than left to rebuild them from the store's one row index per row"
- on `RangeListState`: "the list itself is built once for the whole row group"
`ParquetReadState.forRead` now also presents null ranges as a neutral alternative. Passing null at L559, which reads the same rows, or copying the list in `RangeListState` would give each leaf reader a list of its own, so memory grows with ranges times leaf columns, against a budget that counts the ranges once. Could we keep one clause here, e.g. "It is charged once, since every column reader walks that one list."?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -477,14 +409,10 @@ private void initializeLateMaterialization() throws IOException {
return;
}
- // The non-key columns phase 2 reads. If this file has none of them, the filter saves nothing,
- // since phase 1 already reads every column the file has for every row. The reader then
- // declines it. The planner already declines a projection of key columns alone, so this catches
- // a file missing every projected non-key leaf, which also covers the row-index column. A struct
- // the file has, but none of whose requested fields it has, counts only under the legacy
- // `returnNullStructIfAllFieldsMissing`, since otherwise the clipped schema reads one of its
- // other fields to tell a null struct from one whose requested fields are all null. The checked
- // column is left out too, since a file column of its name is never decoded.
+ // The non-key columns phase 2 reads. A file with none of them gains nothing from the filter,
+ // so the reader declines it. A struct none of whose requested fields the file has counts as
+ // missing only under the legacy `returnNullStructIfAllFieldsMissing`, since otherwise the
+ // clipped schema reads one of its other fields. The checked column never counts.
Review Comment:
**11. "The checked column never counts" reads as "never counts as missing"**
Right after "counts as missing only under the legacy `returnNullStructIfAllFieldsMissing`", it reads as if the checked column were always present. The code does the opposite: L419 leaves it out of `nonKey`, so a file column of that name never makes the file look as if it had a non-key column. The base said "The checked column is left out too, since a file column of its name is never decoded."
Maybe "The checked column is left out."?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -57,56 +57,37 @@
/**
* A {@link VectorizedParquetRecordReader} that reads a row group in phases under a
* {@code ParquetStorageFilter}, so the projected columns of the rows the filter rejects are never
- * read. The filter is a deterministic conjunct of the post-scan {@code Filter}, which still holds
- * it, so any phase below may stop short and read the file the way a plain scan would.
+ * read. The post-scan {@code Filter} still holds the filter, so any phase may stop short and read
+ * the rows the way a plain scan would.
*
- * <p>Per row group, all three phases driven by the one {@link #fileReader}:
+ * <p>Per row group:
* <ul>
- * <li>phase 0 asks the column index which rows the pushed data filter allows, metadata only;</li>
- * <li>phase 1 reads the key columns the filter references over those rows, evaluates the filter
- * per row, and buffers the surviving key values;</li>
- * <li>phase 2 reads the remaining projected columns over the surviving row ranges.</li>
+ * <li>phase 0 asks the column index which rows the pushed data filter allows;</li>
+ * <li>phase 1 reads the key columns over those rows, evaluates the filter, and buffers the
+ * surviving key values;</li>
+ * <li>phase 2 reads the other projected columns over the surviving row ranges.</li>
* </ul>
+ * The emit path splices the buffered key values into the batch, so the key columns are read once.
+ * A row group whose buffer passes {@code maxSplicedRowGroupBytes} gives splicing up, and phase 2
+ * reads its key columns again. If the row ranges alone pass it, the filter is given up too.
*
- * <p>The emit path then splices the buffered key vectors into the batch, so phase 2 never reads a
- * key column again. A row group whose buffer has passed
- * {@code spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes} gives splicing up, and
- * phase 2 reads its key columns again with the rest. If the survivors' row ranges alone pass it,
- * the filter is given up for that row group too.
- *
- * <p>A whole file is read without splicing in these cases:
+ * <p>Some files and row groups get less:
* <ul>
- * <li>every key column is missing from it under schema evolution. The predicate is then constant
- * over the file and is evaluated once, against the constants the reader materializes for
- * those columns. True keeps the file unfiltered, false skips it;</li>
- * <li>the file has none of the projected non-key columns, or none is projected. Phase 2 would
- * have nothing to read, so the reader declines the filter and reads the file plainly;</li>
- * <li>{@code parquet.filter.columnindex.enabled} is false, so its page index is not used. The
- * filter then only skips row groups it rejects whole.</li>
+ * <li>a file with none of the key columns has a constant predicate, evaluated once. False skips
+ * the file, and true reads it whole;</li>
Review Comment:
**13. "true reads it whole" vs "read plainly"**
A kept file with none of the key columns leaves `keyColumns` null, so `loadNextRowGroup` goes to `super.loadNextRowGroup()` and then `readNextFilteredRowGroup()`, which still skips the pushed data filter's pages and row groups. The next bullet calls the same path "read plainly", and `decideAllKeysMissingFile` says "reads it the way a plain scan would", for an error as well.
Maybe "true reads it plainly"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -57,56 +57,37 @@
/**
* A {@link VectorizedParquetRecordReader} that reads a row group in phases under a
* {@code ParquetStorageFilter}, so the projected columns of the rows the filter rejects are never
- * read. The filter is a deterministic conjunct of the post-scan {@code Filter}, which still holds
- * it, so any phase below may stop short and read the file the way a plain scan would.
+ * read. The post-scan {@code Filter} still holds the filter, so any phase may stop short and read
Review Comment:
**10. "are never read" is the last description that overstates the skip**
> so the projected columns of the rows the filter rejects are never read.
Phase 1 reads the key columns, which are projected, for every row the pushed filter kept. Phase 2 reads whole pages, so it also reads rejected rows that share a page with a survivor. This PR's metrics doc says so: "A row excluded within a row group may still have been read, as part of a page that held a survivor." `ParquetStorageFilter`'s class doc now points here instead of keeping its base text, "skip data-column pages that do not overlap any surviving row range".
`FileFormat`, `DataSourceScanExec` and `storageFiltersFor` now say "to skip reading the other columns of the rows they reject". Could this use the same wording?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -185,9 +184,8 @@ public VectorizedParquetRecordReader(boolean useOffHeap, int capacity) {
/**
* Implementation of RecordReader API. Super's delegates to the overload below, which runs
- * {@link #initializeInternal()}, so this does not run it a second time. Kept although it only
- * calls super, because removing it fails MiMa with a {@code DirectAbstractMethodProblem} on
- * {@code org.apache.hadoop.mapreduce.RecordReader.initialize}.
+ * {@link #initializeInternal()}. Kept although it only calls super, since removing it fails
+ * MiMa.
Review Comment:
**14. The MiMa reason lost the problem name**
`project/MimaExcludes.scala` excludes `org.apache.spark.sql.execution.*`, so "removing it fails MiMa" now looks stale to anyone who checks there. It fails only because MiMa reports a `DirectAbstractMethodProblem` on `org.apache.hadoop.mapreduce.RecordReader.initialize`, which that exclusion does not cover. The base named it. Could we keep that?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -672,42 +578,31 @@ protected boolean suppliesOwnVector(int slot) {
}
/**
- * Counts a file the filter rejects whole, which happens when every key column is missing from it
- * and the predicate is constant-false for the value the reader would have materialized. Every row
- * group the pushed data filter left rows in counts as skipped, with every projected byte of those
- * rows as avoided. Those are all non-key bytes, since the file has no key column.
+ * Counts a file whose constant predicate is false. Every row group with rows the pushed data
+ * filter kept counts as skipped, with all their projected bytes avoided, since the file has no
+ * key column.
*/
private void recordFileSkipped() {
StorageFilterMetrics m = storageFilter.metrics();
List<BlockMetaData> blocks = fileReader.getRowGroups();
for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) {
- // Measured against the rows the pushed data filter kept, which is the baseline every other
- // skip path uses. Resolving them again is a cache hit whenever the two can differ, because
- // `getFilteredRecordCount()` at initialize resolved every block's ranges then.
RowRanges blockRanges = pushedFilterRangesFor(blockIdx, blocks.get(blockIdx).getRowCount());
long survivingRows = blockRanges.rowCount();
if (survivingRows == 0) continue;
- // The key columns are missing from this file, so they contribute nothing to the walk, and the
- // whole projection is what a plain read would have transferred.
long avoidedBytes = blockAccounting.compressedBytesForRowRanges(
blockIdx, requestedPaths, blockRanges, survivingRows);
m.recordRowGroupSkipped(survivingRows, avoidedBytes);
}
}
/**
- * The rows of a block the pushed data filter allows, at column-index granularity, or every row of
- * the block when {@link #pageIndexDisabled} says the page index must not be used.
- *
- * <p>{@code getRowRanges} asks only whether a filter is pushed, not whether column-index
- * filtering is still enabled, so this is where that conf is honoured. Every phase reads within
- * these ranges, so narrowing them by an index the read may not trust would cost rows.
+ * The rows of a block the pushed data filter allows, at column-index granularity, or every row
+ * when {@link #pageIndexDisabled}. {@code getRowRanges} does not check that conf itself.
*/
private RowRanges pushedFilterRangesFor(int blockIdx, long blockRowCount) {
if (blockRowCount == 0) {
- // parquet-mr never writes an empty block, and the plain read path skips one too. For one,
- // `RowRanges.createSingle(0)` below would build Range(0, -1) and trip its `from <= to`
- // assertion.
+ // parquet-mr never writes an empty block, but `RowRanges.createSingle(0)` would fail on
Review Comment:
**15. `createSingle(0)` fails only under `-ea`**
The only check in `RowRanges.Range` is `assert from <= to`, so "would fail" holds only with assertions on, as in tests. In production it builds `Range(0, -1)` with zero rows, which both callers skip. The base said it would "trip its `from <= to` assertion". Maybe keep "assertion"?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped, see [[throwIfKilled]], since the error may be the
+ * kill's interrupt surfacing through a UDF;
+ * - a fatal error deeper in the cause chain, as deep as the executor looks, goes out as itself.
*
- * An `InternalError` among those goes out in an `UnknownError`, a fatal error of another type.
- * `FileScanRDD` would take a bare `InternalError` from a reader for a corrupt file under
- * `ignoreCorruptFiles`. A `NonFatal` wrapper does not work either. With a kill pending, the
- * executor reports one as killed and keeps running. A plain plan raises it from the post-scan
- * Filter, where neither happens, and the executor stops. It stops on the `UnknownError` too, at
- * any `spark.executor.killOnFatalError.depth` of 1 or more, since that is at the top.
+ * An `InternalError` among those goes out in an `UnknownError`, another fatal error. A bare one
+ * would be taken for a corrupt file under `ignoreCorruptFiles`, and a `NonFatal` wrapper would
+ * let the executor run on under a pending kill.
*
* An error from decoding differs in three ways:
- * - a plain read raises it from the reader, where `FileScanRDD` wraps a `NonFatal` one in
- * FAILED_READ_FILE before the executor looks. So a fatal error deeper in the chain of a
- * `NonFatal` one has to sit one level shallower to count, while any other error is searched
- * as deep as the executor does. Under `ignoreCorruptFiles` a plain read may skip the rest of
- * the file on it instead.
+ * - `FileScanRDD` wraps a `NonFatal` one before the executor looks, so a fatal cause counts one
Review Comment:
**12. "counts one level shallower" can read either way**
> so a fatal cause counts one level shallower;
The code searches one level less deep (`fatalErrorDepth - 1`), because `FileScanRDD` wraps a plain read's `NonFatal` error in FAILED_READ_FILE, which puts its cause one level deeper for the executor. "counts one level shallower" can also read as "counts as if it were one level shallower", which is the opposite. The base said the cause "has to sit one level shallower to count", and that a plain read raises it from the reader, which is what the `- 1` mirrors.
Maybe "so a fatal cause has to sit one level shallower to count;"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -523,75 +450,61 @@ private void initializeLateMaterialization() throws IOException {
}
/**
- * Loads the next row group with rows left to emit, through the three phases of the class javadoc.
- * Phase 2 reads the non-key columns over the survivors while splicing, the whole projection over
- * the survivors once splicing is given up, and the whole projection over
- * {@code pushedFilterRanges} once the filter is. Returns false when the row groups run out, which
- * ends a read the filter narrowed.
+ * Loads the next row group with rows left to emit, through the phases of the class javadoc.
+ * Returns false when the row groups run out.
*/
@Override
protected boolean loadNextRowGroup() throws IOException {
if (keyColumns == null) return super.loadNextRowGroup();
- // The previous row group is fully emitted by the time this is called, so its pages are done
- // with. Released here rather than at the next assignment, so a row group the filter empties,
- // or the end of the file, does not keep the previous one's pages alive. Its column readers go
- // too. They hold the row ranges it was read over, which the budget no longer counts, and the
- // decoders of their last page.
+ // The previous row group is fully emitted, so its pages and column readers go now rather than
+ // at the next assignment, which a row group the filter empties would delay.
Review Comment:
**9. Why the readers go before the next row group's phase 1**
The base also said: "Its column readers go too. They hold the row ranges it was read over, which the budget no longer counts". That is the reason for every row group, not only for the one the filter empties. The next assignment (`dataPages` at L524, `installRowGroup` at L559) always comes after the next row group's phase 1, whose budget starts from zero (L667-668).
Folding the release into that assignment, or dropping `releaseRowGroupReaders()` because `installRowGroup` replaces the readers anyway, would keep the previous row group's ranges and pages alive through every phase 1. That is up to about twice the cap outside the memory manager. This is about the call site, not the method javadoc from round-1 comment 8.
Could we keep "The readers hold the row ranges it was read over, which the next row group's budget does not count."?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -190,11 +151,10 @@ public class LateMaterializationParquetRecordReader extends VectorizedParquetRec
/**
* Whether {@code parquet.filter.columnindex.enabled} is false, so this file's page index must not
- * be used. That turns off phase 0 and every partial read. A wrong index would pair a row's key
- * with another row's values, or return rows the filter rejects once splicing is given up. The
- * post-scan Filter catches neither, since it takes the reader's mark on those rows. Parquet's
- * partial read does not check the conf itself. Read off {@link #readOptions}, so it agrees with
- * the row count {@link #fileReader} reported.
+ * be used. That turns off phase 0 and every partial read, which parquet does not check itself.
+ * It is the way out of a wrong page index, which would pair a row's key with another row's
+ * values, and which the post-scan Filter cannot catch on the rows this reader marks checked.
+ * Read off {@link #readOptions}, so it agrees with {@link #fileReader}.
Review Comment:
**7. `pageIndexDisabled` no longer says what it has to agree with**
The base said "so it agrees with the row count `{@link #fileReader}` reported". That is the invariant. `totalRowCount` is `fileReader.getFilteredRecordCount()`, computed with the same `useColumnIndexFilter`, and `VectorizedParquetRecordReader.nextBatch` stops once `rowsReturned >= totalRowCount`, before it asks `loadNextRowGroup`. If another switch were ORed in here, this reader would read whole row groups that parquet counted narrower, and `nextBatch` would end the split early, dropping its last rows, matching ones included, with no error. "agrees with fileReader" no longer says which part has to agree. The base class's `loadNextRowGroup` javadoc dropped the same point ("totalRowCount already says when they run out").
Also, the reason restored above in round 1 covers the mispaired values, but no longer the base's "or return rows the filter rejects once splicing is given up", which is the case of a narrowed row group whose rows are still marked checked.
Could we keep "the row count `{@link #fileReader}` reported"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetBlockAccounting.java:
##########
@@ -84,29 +77,14 @@ boolean hasOffsetIndexes(int blockIndex, Set<ColumnPath> paths) {
}
/**
- * Compressed bytes a reader transfers for the leaf columns named by {@code paths} when it reads
- * exactly {@code rowRanges} of the given block. Page headers and the dictionary page are
- * included, since both are read whenever any page of a chunk is read. {@code rowRangeCount} is
- * {@code rowRanges.rowCount()}, passed in because that walks every range and the caller has it.
- *
- * <p>Two sources, chosen so this never causes IO of its own:
- * <ul>
- * <li>{@code rowRanges} covers the whole block, and the answer is the sum of the chunks'
- * {@code getTotalSize()}, which is already in the footer. This is the case that matters.
- * Whenever nothing else has built the block's {@link ColumnIndexStore}, {@code rowRanges}
- * is necessarily the whole block, because a narrower range can only come from column-index
- * filtering, which builds the store as a side effect.
- * <li>{@code rowRanges} is a strict subset, and the walk goes over the offset index, as
- * parquet's own read path does, adding the dictionary page the way
- * {@code calculateOffsetRanges} does. The store is guaranteed to exist here, so the walk is
- * pure metadata arithmetic. For ranges a reader narrowed itself, which column-index
- * filtering had no hand in, that guarantee is an ordering one, since the reader's own read
- * of those ranges built the store first. A new caller has to keep that order.
- * </ul>
+ * Compressed bytes a reader transfers for the columns in {@code paths} when it reads exactly
+ * {@code rowRanges} of the block, page headers and the dictionary page included.
+ * {@code rowRangeCount} is {@code rowRanges.rowCount()}, which the caller has.
*
- * <p>Columns absent from this physical file (schema evolution) contribute nothing, which is
- * correct, since the reader transfers nothing for them. They fall out of the walk by themselves,
- * which goes over the block's chunks rather than over {@code paths}.
+ * <p>A whole block is the chunks' sizes from the footer. A strict subset is walked over the
+ * offset index, as parquet's read does. That reads the block's {@link ColumnIndexStore}, which
+ * costs no IO only once a read of those ranges has built it, so a new caller has to keep that
Review Comment:
**5. The no-IO note lost the cached-store hazard and the column-index case**
The no-IO precondition moved here from the class javadoc, as HyukjinKwon checked. Two parts of the deleted text did not come with it:
1. The old text said a walk before the store exists "would read the offset indexes itself and cache a store over the reader's current columns". The cached store is what breaks things. `getColumnIndexStore` builds it over the current paths and keeps it for the block, so a later `readFilteredRowGroup` over other columns gets a null offset index and throws an NPE in parquet's `filterOffsetIndex`, which the `MissingOffsetIndexException` catch does not take. With no data filter pushed, nothing builds the store before phase 2, so a walk over `survivors` placed after phase 1 would break phase 2's read. "costs no IO" makes it read as a cost only.
2. "only once a read of those ranges has built it" leaves out column-index filtering, which `recordFileSkipped` and `nonKeyBaselineBytes` rely on: `getFilteredRecordCount()` at initialize built every block's store over the whole projection.
The class javadoc's "with no IO of its own" (L41) is now unconditional too. Could this say, e.g., "That needs the block's store, which column-index filtering or a read of those ranges over the same columns builds. A walk before that builds and keeps one over the reader's current columns, so a later partial read of other columns fails."?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6090886845
I merged #59269, could you rebase this PR, @peter-toth ?
- https://github.com/apache/spark/pull/59269
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4283438528
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -264,39 +217,26 @@ object ParquetStorageFilter {
}
/**
- * Rethrows an error the reader met while applying a storage filter, if it must not fall back on
- * it by giving the filter up. `fromRead` says the error came from decoding the key pages rather
- * than from evaluating the filter. The checks run in the executor's own order:
- * - a fatal error goes out as it is, kill or no kill.
- * - a pending task kill goes out wrapped when the error is one the executor's kill case takes,
- * see [[throwIfKilled]]. The error may be the kill's interrupt surfacing through a UDF that
- * wrapped it. `PythonRunner` reads a kill off the task context in the same way.
- * - a fatal error deeper in the cause chain, found by `Executor.isFatalError` as deep as the
- * executor would look had a plain plan raised `e`, goes out as itself, so the executor finds
- * it at the top whatever wraps it on the way.
+ * Rethrows an error the reader met while applying a storage filter, unless it may give the filter
+ * up on it. `fromRead` says the error came from decoding the key pages. In the executor's order:
+ * - a fatal error goes out as it is;
+ * - a pending task kill goes out wrapped when the executor would report the error as a kill,
+ * see [[throwIfKilled]], since the error may be the kill's interrupt surfacing through a UDF;
Review Comment:
Done in [0c77f7ddd22](https://github.com/apache/spark/pull/59324/commits/0c77f7ddd22e4f70db4072d4db39d832ed53dbe7): "..., and as it is otherwise, see [[throwIfKilled]], ...".
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetBlockAccounting.java:
##########
@@ -38,14 +38,8 @@
/**
* What a partial read of a Parquet file transfers, and whether one is possible at all, answered
- * from the footer the {@link ParquetFileReader} already holds. Every method here is metadata
- * arithmetic and causes no IO of its own, which is what lets a reader report these numbers per row
- * group. For a strict subset of a block that holds only once the block's column index store is
- * built, see {@link #compressedBytesForRowRanges}. A walk before that would read the offset indexes
- * itself and cache a store over the reader's current columns.
- *
- * <p>One instance per reader, which reads one split of a file, because the warning below is
- * reported once per instance.
+ * from the metadata the {@link ParquetFileReader} already holds, with no IO of its own. One
Review Comment:
Done in [0c77f7ddd22](https://github.com/apache/spark/pull/59324/commits/0c77f7ddd22e4f70db4072d4db39d832ed53dbe7): "with no IO of its own once the block's store is built, see {@link #compressedBytesForRowRanges}".
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6121315606
@HyukjinKwon @viirya thank you for the reviews. Everything is in [0eb44e78ba1](https://github.com/apache/spark/pull/59324/commits/0eb44e78ba139cfaf181e82d29fda059f381af44).
- Fixed: the "Otherwise" polarity, the `initialize` javadoc, the interpreted-predicate case in `preparedPredicate`, the `hugeVectorReserveRatio` caveat, and the clause order in `storageFiltersFor`.
- Added to the conf doc: one sentence on files without a page index and on `parquet.filter.columnindex.enabled=false`.
- Back in: why the reader honors `columnindex.enabled` itself, that a subclass reading its own columns goes through `newColumnReader`, the "bug of ours" note in the splicing catch, why `buildReaderWithStorageFilters` returns an `Option`, and the groups of `output` behind `preparedStorageFilters`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6210189582
Thank you @dongjoon-hyun, @HyukjinKwon and @viirya for the review.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6210171533
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/27911de56caa929d0af05808c9e4501016afe3aa
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #59324:
URL: https://github.com/apache/spark/pull/59324#issuecomment-6124739108
Thanks, @peter-toth. I rechecked 0eb44e78ba1, and each item you listed is in: the "Without one" polarity fix, the `initialize` javadoc, the interpreted-predicate case in `preparedPredicate`, the `hugeVectorReserveRatio` caveat, the clause order in `storageFiltersFor`, the conf-doc sentence on files without a page index and `parquet.filter.columnindex.enabled=false`, and the restored notes on `pageIndexDisabled`, `newColumnReader`, the splicing catch, the `Option` return of `buildReaderWithStorageFilters`, and the groups of `output` behind `preparedStorageFilters`.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:6121315606","thread_id":"conversation:6121315606","verdict_sha256":"ef198e858b138b4ed0bd3a39bd53658c27b55bb8a561d49633cd71ccb234b899"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60107][SQL] Trim the storage-filter comments and conf docs [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4267724037
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -185,9 +184,8 @@ public VectorizedParquetRecordReader(boolean useOffHeap, int capacity) {
/**
* Implementation of RecordReader API. Super's delegates to the overload below, which runs
- * {@link #initializeInternal()}, so this does not run it a second time. Kept although it only
- * calls super, because removing it fails MiMa with a {@code DirectAbstractMethodProblem} on
- * {@code org.apache.hadoop.mapreduce.RecordReader.initialize}.
+ * {@link #initializeInternal()}. Kept although it only calls super, since removing it fails
+ * MiMa.
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3).
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -672,42 +578,31 @@ protected boolean suppliesOwnVector(int slot) {
}
/**
- * Counts a file the filter rejects whole, which happens when every key column is missing from it
- * and the predicate is constant-false for the value the reader would have materialized. Every row
- * group the pushed data filter left rows in counts as skipped, with every projected byte of those
- * rows as avoided. Those are all non-key bytes, since the file has no key column.
+ * Counts a file whose constant predicate is false. Every row group with rows the pushed data
+ * filter kept counts as skipped, with all their projected bytes avoided, since the file has no
+ * key column.
*/
private void recordFileSkipped() {
StorageFilterMetrics m = storageFilter.metrics();
List<BlockMetaData> blocks = fileReader.getRowGroups();
for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) {
- // Measured against the rows the pushed data filter kept, which is the baseline every other
- // skip path uses. Resolving them again is a cache hit whenever the two can differ, because
- // `getFilteredRecordCount()` at initialize resolved every block's ranges then.
RowRanges blockRanges = pushedFilterRangesFor(blockIdx, blocks.get(blockIdx).getRowCount());
long survivingRows = blockRanges.rowCount();
if (survivingRows == 0) continue;
- // The key columns are missing from this file, so they contribute nothing to the walk, and the
- // whole projection is what a plain read would have transferred.
long avoidedBytes = blockAccounting.compressedBytesForRowRanges(
blockIdx, requestedPaths, blockRanges, survivingRows);
m.recordRowGroupSkipped(survivingRows, avoidedBytes);
}
}
/**
- * The rows of a block the pushed data filter allows, at column-index granularity, or every row of
- * the block when {@link #pageIndexDisabled} says the page index must not be used.
- *
- * <p>{@code getRowRanges} asks only whether a filter is pushed, not whether column-index
- * filtering is still enabled, so this is where that conf is honoured. Every phase reads within
- * these ranges, so narrowing them by an index the read may not trust would cost rows.
+ * The rows of a block the pushed data filter allows, at column-index granularity, or every row
+ * when {@link #pageIndexDisabled}. {@code getRowRanges} does not check that conf itself.
*/
private RowRanges pushedFilterRangesFor(int blockIdx, long blockRowCount) {
if (blockRowCount == 0) {
- // parquet-mr never writes an empty block, and the plain read path skips one too. For one,
- // `RowRanges.createSingle(0)` below would build Range(0, -1) and trip its `from <= to`
- // assertion.
+ // parquet-mr never writes an empty block, but `RowRanges.createSingle(0)` would fail on
Review Comment:
Done in [1410d4684cb](https://github.com/apache/spark/pull/59324/commits/1410d4684cb7dffceb36fb7ecce3c0168e67fda3).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org