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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
