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: [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]
