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