This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new f56e2870a6fa [SPARK-60034][SQL] Skip a storage filter on the rows the
Parquet reader already checked
f56e2870a6fa is described below
commit f56e2870a6fa9344ec1c7039971947c10952a3cd
Author: Peter Toth <[email protected]>
AuthorDate: Sat Oct 10 00:13:47 2026 +0100
[SPARK-60034][SQL] Skip a storage filter on the rows the Parquet reader
already checked
### What changes were proposed in this pull request?
This is a follow-up to #58895 (SPARK-59620), the last item of its follow-up
list.
A storage filter stays in the post-scan `Filter`, so the reader is free to
give it up anywhere. Today the `Filter` then evaluates it a second time on
every row the scan emits, including the rows the reader already checked. This
PR lets the `Filter` skip those rows.
**How it works**
- When `FileSourceStrategy` offers storage filters to a scan that returns
columnar batches, it adds an internal nullable boolean column,
`_tmp_storage_filter_checked`, to the scan's required schema and output.
- In the post-scan `Filter`, each offered conjunct becomes `IF(checked,
true, conjunct)`. Every other conjunct stays as it is. A `Project` above the
`Filter` drops the column, added if the plan had none.
- An `IF` rather than an `OR`, so that a null mark leaves the conjunct, and
an `AND` around it, as they were. It also keeps subexpression elimination from
computing a hash that two blooms share ahead of the skip.
- The reader fills the column. `FileFormat.buildReaderWithStorageFilters`
documents the values:
- true says the reader evaluated every storage filter on the row and kept
it;
- false or null says the reader did not check the row, so the `Filter`
decides it as if the column were not there. So a reader that does not know the
column, and reads it as missing from the file, is still right.
**Parquet reader**
`LateMaterializationParquetRecordReader` finds the column by name in the
requested schema. It puts a vector of its own in the column's slot and fills it
per batch. A batch never spans two row groups, so one value covers each batch.
- True for a row group the filter was applied to. Its rows are then exactly
its survivors, whether splicing was kept or given up.
- False for a row group the filter was given up on. That is an error
evaluating the filter or decoding its key pages, survivor ranges past the
budget, a row group that cannot be narrowed, or an offset index parquet cannot
read. Also false for a file the reader declines.
- For a file with none of the key columns, true when the constant predicate
kept the file, and false when it raised an error.
The vector is on or off heap like the rest of the batch. The reader never
decodes a primitive file column of that name, which a user-given schema can
pair with the checked column. Nor does such a file column count as a non-key
column when the reader decides whether to decline a file.
**Planning**
- The column is added only where the scan still returns columnar batches
with it, as `FileSourceScanExec.supportsColumnar` decides. So it counts against
the field limit, and it never changes how the scan runs. Above a row-based
scan, the `Project` that drops it would copy every surviving row. The `Filter`
then evaluates the offered conjuncts on every row, as before this PR.
- Above a columnar scan, whole-stage codegen fuses that `Project` with the
`Filter`. Where something keeps the `Filter` out of codegen, e.g. a
`CodegenFallback` conjunct, the `Project` copies the surviving rows once.
Measured locally, that cost about 10% with 90 long columns and half the rows
surviving, and nothing measurable with 8 columns or 5% surviving.
- A relation with a column of that name, compared case-insensitively, is
offered no storage filters. A file holding one outside the relation's schema is
not supported, as for the row-index metadata column. The conf doc says both.
- `ParquetStorageFilter.create` rejects a key column of that name, as it
does one named like the row-index column.
- The offered conjuncts are found in the `Filter` through an
`ExpressionSet`. The scan's `storageFilters` carry the relation's column names,
while the `Filter` keeps the query's spelling.
- The `Project` that builds `_metadata` sits below the `Filter`, so it
keeps the column. It is now built from the scan's output without the flattened
metadata columns.
- A format that opts in to storage filters gets the column in every reader
it builds for the scan, including the ordinary one the scan falls back to when
the format declines. `FileFormat` says so. Only `ParquetFileFormat` itself opts
in today, and its subclasses do not.
A wrong page index can now do one more thing. Where splicing is given up,
phase 2 reads the key columns again over the survivor ranges. A wrong index
there returns rows the filter rejects, and the `Filter` no longer catches them,
since the rows are marked true. A wrong index could already pair a row's key
with another row's values, and `parquet.filter.columnindex.enabled=false` is
the escape hatch for both. The conf doc says so.
### Why are the changes needed?
- It removes the second evaluation of the filter on the rows the reader
already checked. For a runtime bloom that is a hash and a few bit probes per
surviving row.
- The planned DSv2 support builds on it. A connector cannot promise at
planning time that it filters every row, so the plan keeps the `Filter`. The
connector then says per row what it checked, through the same column.
### Does this PR introduce _any_ user-facing change?
No. The feature is unreleased and behind
`spark.sql.parquet.storageFilterPushdown.enabled`, which is off by default.
With it on, 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. A columnar scan that is offered them shows the column in its
output and its `ReadSchema`, and the `Filter` shows the rewritten conjunct.
### How was this patch tested?
New tests in `ParquetStorageFilterSuite`. The reader tests build the reader
through `ParquetFileFormat`, as a scan does.
- the reader's marks for a row group the filter was applied to, over
several batches, and for one that gave only splicing up, which the bytes read
confirm;
- the marks for these give-ups: an evaluation error, survivor ranges past
the budget, a file with no offset index, and an offset index parquet cannot
parse;
- the marks for a file the reader declines, and for a file with no key
columns, kept or raising an error;
- a file column named like the checked column: a long one, which the reader
must not decode, and a boolean one, which must not stop the reader from
declining the file;
- off heap, the column's vector is freed with the reader;
- the planner's rewrite, with the query spelling the key differently from
the relation, a conjunct it does not offer left alone, and the column not
leaving the `Filter`;
- a null mark in front of a bloom that rejects the row and a conjunct that
throws, generated and interpreted. The `IF` rejects the row, and an `OR` would
throw;
- a relation with a clashing column name, which gets no storage filters and
no column;
- the column added only to a scan that stays columnar with it, with a
partition column: below the field limit, where the column alone would cross it,
past it, and with whole-stage codegen off. The scan's columnar output is
checked at each;
- an end-to-end query filtered on a bloom alone, against a plain read. A
runtime bloom is redundant with its join, so a join would hide a wrong mark. It
runs with the filter applied and with the filter given up in every row group
that has a survivor, with the scan's metrics confirming each. Where the filter
is applied, the marks are also read off the scan itself. It has a partition
column, and runs with and without `_metadata`;
- a runtime bloom from a join, whose conjunct holds a subquery, rewritten
in the `Filter` and marked by the reader;
- `ParquetStorageFilter.checkedColumnIndex` finding the column, and
rejecting one that is not a boolean, and `create` rejecting a key column of
that name.
Each was checked by mutation. These all make at least one of them fail:
- forcing the mark to true, or to false;
- marking true only while splicing;
- never marking a file with no key columns true;
- rewriting to `OR` instead of `IF`;
- dropping the column from the `_metadata` `Project`;
- placing it after the partition columns in the scan's output;
- removing the name guard or the rewrite;
- matching the offered conjuncts by plain equality;
- adding the column when nothing is offered;
- adding it with whole-stage codegen off, or past the field limit, or
counting only the data columns against the limit, or not counting the column
itself;
- reading the file's column into the slot, counting it as a non-key column,
or comparing its name case-sensitively there;
- dropping the type check on the column, or `create`'s check on a key
column of that name;
- not closing the vector, or not putting it in the slot.
`ParquetStorageFilterSuite` passes, 133 tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
Closes #59269 from peter-toth/SPARK-60034-storage-filter-checked-column.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 12 +-
.../LateMaterializationParquetRecordReader.java | 104 +++--
.../spark/sql/execution/DataSourceScanExec.scala | 6 +-
.../sql/execution/datasources/FileFormat.scala | 22 +
.../execution/datasources/FileSourceStrategy.scala | 86 ++--
.../datasources/parquet/ParquetStorageFilter.scala | 21 +-
.../parquet/ParquetStorageFilterSuite.scala | 453 ++++++++++++++++++++-
7 files changed, 630 insertions(+), 74 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 994d1ec68808..3a72fd622a2a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1982,16 +1982,22 @@ object SQLConf {
"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. A row group where
it stops reads " +
+ "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. Setting " +
+ "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. A file without a page index
falls back the same " +
+ "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 " +
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java
index 0ae8200e9110..b3ac40ee5eff 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java
@@ -42,10 +42,12 @@ import org.apache.spark.internal.SparkLoggerFactory;
import org.apache.spark.memory.MemoryMode;
import org.apache.spark.sql.catalyst.expressions.BasePredicate;
import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.execution.datasources.FileFormat;
import org.apache.spark.sql.execution.vectorized.OffHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.DecimalType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
@@ -93,6 +95,19 @@ import org.apache.spark.sql.vectorized.ColumnarBatch;
* pay a second value copy per surviving row, a byte copy for a
variable-length key, where the
* swap is one array write per key column per batch.
*
+ * <p>When the scan has the column {@code
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME}, this
+ * reader fills it, so the post-scan Filter skips the storage filter on the
rows it checked. A batch
+ * never spans two row groups, so one value covers each batch:
+ * <ul>
+ * <li>true for a row group the filter was applied to, whose rows are then
exactly its survivors,
+ * spliced or not;</li>
+ * <li>false for a row group or a file the filter was given up on or
declined for;</li>
+ * <li>for a file with none of the key columns, true when the constant
predicate kept it.</li>
+ * </ul>
+ * The slot holds a vector of this reader's own. A file holding a column of
that name outside the
+ * relation's schema is not supported, see {@code
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME}. A
+ * primitive one is still never decoded, see {@link #suppliesOwnVector}.
+ *
* <p>What {@code FileSourceStrategy.storageFiltersFor} and {@code
ParquetStorageFilter.create}
* guarantee is asserted here, since a violation is a planner bug.
*/
@@ -176,9 +191,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, which the post-scan Filter cannot catch.
Parquet's partial read does
- * not check the conf itself. Read off {@link #readOptions}, so it agrees
with the row count
- * {@link #fileReader} reported.
+ * 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.
*/
private boolean pageIndexDisabled;
@@ -199,6 +215,16 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
private WritableColumnVector[] keyScratchVectors;
private ColumnarBatch keyScratchBatch;
+ /**
+ * The batch slot of the column this reader marks the rows it checked in, or
-1 when the scan has
+ * none. Looked up by name, as the base looks up the row-index column.
+ */
+ private int checkedSlot = -1;
+ /** What {@link #checkedVector} says for every row of the next batch, see
the class javadoc. */
+ private boolean rowsChecked;
+ /** The vector in the checked column's slot, or null when the scan has no
such column. */
+ private WritableColumnVector checkedVector;
+
/**
* Splicing state. Phase 1 keeps the surviving key values it decoded, and
the emit path splices
* them into the output batch. That saves phase 2 a second read of the key
columns, which parquet
@@ -255,14 +281,22 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
try {
if (persistentBatchColumns != null) {
// Through the persistent vectors rather than the emitted batch,
whose key slots may hold
- // survivor vectors the splicing state closes instead. Super then
has no batch to close.
+ // survivor vectors the splicing state closes instead, and whose
checked slot holds this
+ // reader's own vector. Super then has no batch to close.
columnarBatch = null;
closeAll(persistentBatchColumns);
persistentBatchColumns = null;
spliceBatchColumns = null;
}
} finally {
- closeSplicingState();
+ try {
+ closeSplicingState();
+ } finally {
+ if (checkedVector != null) {
+ checkedVector.close();
+ checkedVector = null;
+ }
+ }
}
} finally {
try {
@@ -299,23 +333,30 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
}
/**
- * Takes the vectors of the batch super built. For a file the filter is
applied to row by row, it
- * then hands out a batch of its own over a copy of that array, whose key
slots the emit path
- * rewrites in place, and allocates phase 1's key vectors. {@link
ColumnarBatch} holds the array
- * by reference, its staging row included, so rewriting a slot is what
publishes it.
+ * Takes the vectors of the batch super built. For a file the filter is
applied to row by row, or
+ * a scan with the checked column, it then hands out a batch of its own over
a copy of that array.
+ * The emit path rewrites that batch's key slots in place, and its checked
slot holds
+ * {@link #checkedVector}. Phase 1's key vectors are allocated only for a
file the filter is
+ * applied to row by row. {@link ColumnarBatch} holds the array by
reference, its staging row
+ * included, so rewriting a slot is what publishes it.
*
* <p>Every slot got a vector, key columns included. A spliced row group
leaves a key slot's
- * vector unused, and a row group that gave splicing up reads into it.
+ * vector unused, and a row group that gave splicing up reads into it. The
checked slot's vector
+ * from super stays unused.
*/
private void takeOverBatch() {
persistentBatchColumns = new ColumnVector[columnarBatch.numCols()];
for (int i = 0; i < persistentBatchColumns.length; i++) {
persistentBatchColumns[i] = columnarBatch.column(i);
}
- if (keyColumns == null) return;
+ if (keyColumns == null && checkedSlot < 0) return;
spliceBatchColumns = persistentBatchColumns.clone();
+ if (checkedSlot >= 0) {
+ checkedVector = newVector(DataTypes.BooleanType);
+ spliceBatchColumns[checkedSlot] = checkedVector;
+ }
columnarBatch = new ColumnarBatch(spliceBatchColumns);
- allocateKeyScratch();
+ if (keyColumns != null) allocateKeyScratch();
}
/**
@@ -324,7 +365,7 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
*
* <p>False skips the file, which then yields no rows. True, or an error,
reads it the way a plain
* scan would, under the same fail-open rule as a row's value, since the
constant can be one the
- * predicate throws on.
+ * predicate throws on. An error leaves the rows unchecked.
*/
private void decideAllKeysMissingFile() {
// Not closed, and must not be, since the vectors in it belong to the
batch this reader emits.
@@ -334,6 +375,7 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
boolean keep;
try {
keep = predicate.eval(keyRow.getRow(0));
+ rowsChecked = keep;
} catch (Throwable t) {
giveUpOnError(t, FILE_GIVEN_UP_ON_EVALUATION, false);
keep = true;
@@ -349,7 +391,8 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
* plain reader does, and then points the emitted batch's key slots at this
batch's vectors. A
* spliced row group takes them from the head of the survivor queue, and any
other row group from
* the persistent vectors, which is what a row group following a spliced one
needs. Non-key slots
- * keep {@link #initBatch}'s vectors for the reader's life.
+ * keep {@link #initBatch}'s vectors for the reader's life. Last, it marks
the batch's rows in the
+ * checked column.
*
* <p>The row-index column the base fills is never a key column, since
* {@code ParquetStorageFilter.create} rejects its name. So it always lands
in a persistent
@@ -379,12 +422,16 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
publishedKeyVectors != null ? publishedKeyVectors[k] :
persistentBatchColumns[slot];
}
}
+ if (checkedVector != null) {
+ checkedVector.putBooleans(0, columnarBatch.numRows(), rowsChecked);
+ }
return true;
}
@Override
protected void initializeInternal() throws IOException,
UnsupportedOperationException {
super.initializeInternal();
+ checkedSlot =
ParquetStorageFilter.checkedColumnIndex(sparkRequestedSchema);
initializeLateMaterialization();
}
@@ -436,9 +483,12 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
// 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.
+ // 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.
+ String checkedName = checkedSlot >= 0 ?
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME() : null;
List<ColumnDescriptor> nonKey = requestedColumns.stream()
.filter(column -> !keyTopLevelNames.contains(column.getPath()[0]))
+ .filter(column -> !column.getPath()[0].equalsIgnoreCase(checkedName))
.toList();
if (nonKey.stream().noneMatch(column ->
fileSchema.containsPath(column.getPath()))) return;
nonKeyColumns = nonKey;
@@ -595,6 +645,8 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
m.rowsExcludedWithinRowGroup().add(baselineRows - finalRowCount);
}
+ // The rows phase 2 read are exactly the survivors unless the filter was
given up.
+ rowsChecked = !filterGivenUp;
// Phase 2 read exactly these rows, so its readers are handed the ranges
rather than left to
// rebuild them from the store's one row index per row.
installRowGroup(dataPages, readerRangesFor(finalRanges, finalRowCount,
blockRowCount));
@@ -604,12 +656,14 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
}
/**
- * A spliced row group's key slots take their values from the survivor queue
at emit, so phase 2
- * reads no pages for them. Key columns are primitive, which is what the
base class asks of a slot
- * answered true here.
+ * The checked slot always holds this reader's own vector, so a primitive
file column of that
+ * name, which a case-insensitive or user-given schema can match, is never
decoded. A spliced row
+ * group's key slots take their values from the survivor queue at emit, so
phase 2 reads no pages
+ * for them. Both are primitive, which is what the base class asks of a slot
answered true here.
*/
@Override
protected boolean suppliesOwnVector(int slot) {
+ if (slot == checkedSlot) return true;
if (!spliceCurrentRowGroup) return false;
for (KeyColumn key : keyColumns) {
if (key.batchSlot() == slot) return true;
@@ -694,7 +748,10 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
/**
* Evaluates the storage filter over every row of a key-only {@link
PageReadStore}, in
* capacity-sized chunks, and returns the surviving rows as {@link
RowRanges} in block-row
- * coordinates. The result is a subset of {@code pushedFilterRanges}.
+ * coordinates. The result is exactly the rows the filter kept, within
+ * {@code pushedFilterRanges}. The checked column marks every row read over
it true (see
+ * {@link #loadNextRowGroup}), so ranges that also covered rejected rows
would need a mark per
+ * row.
*
* <p>While splicing, each survivor's key values are appended to the
accumulators for the emit
* path to splice, until the buffer passes its cap.
@@ -833,7 +890,7 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
// the vectors allocated so far reachable for `close()`, which walks this
array element-wise.
keyScratchVectors = new WritableColumnVector[keyColumns.length];
for (int i = 0; i < keyColumns.length; i++) {
- keyScratchVectors[i] = newKeyVector(i);
+ keyScratchVectors[i] = newVector(keyColumns[i].type());
}
// Missing key columns keep their batch vectors (see
`keyVectorsFromBatch`), which the emitted
// batch owns.
@@ -859,9 +916,8 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
return vectors;
}
- /** A capacity-sized vector for the i-th key column this file has, in the
reader's memory mode. */
- private WritableColumnVector newKeyVector(int keyIdx) {
- DataType dt = keyColumns[keyIdx].type();
+ /** A capacity-sized vector in the reader's memory mode, like the rest of
the batch. */
+ private WritableColumnVector newVector(DataType dt) {
return MEMORY_MODE == MemoryMode.OFF_HEAP
? new OffHeapColumnVector(capacity, dt)
: new OnHeapColumnVector(capacity, dt);
@@ -880,7 +936,7 @@ public class LateMaterializationParquetRecordReader extends
VectorizedParquetRec
// allocated so far reachable for `abandonSplicing`.
currentKeyAccumulators = new WritableColumnVector[keyColumns.length];
for (int i = 0; i < keyColumns.length; i++) {
- currentKeyAccumulators[i] = newKeyVector(i);
+ currentKeyAccumulators[i] = newVector(keyColumns[i].type());
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
index 6355393b75da..076d00042743 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
@@ -821,9 +821,9 @@ case class FileSourceScanExec(
// 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.
//
- // `output` is `readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++
- // constantMetadataColumns` and `requiredSchema` is the StructType of the
first two groups, so
- // the first `requiredSchema.length` attributes line up with its fields.
+ // `output` is `readDataColumns ++ generatedMetadataColumns ++
storageFilterCheckedColumn ++
+ // partitionColumns ++ constantMetadataColumns` and `requiredSchema` is the
StructType of the
+ // first three groups, so the first `requiredSchema.length` attributes line
up with its fields.
@transient
protected lazy val preparedStorageFilters: Seq[Expression] = {
val requestedDataAttrs = output.take(requiredSchema.length)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
index 8d1bf829c5a0..bd5a31ddf5b0 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
@@ -177,6 +177,20 @@ trait FileFormat {
* leaves every one of them in the post-scan `Filter` as well, so ignoring
one is a missed
* optimization rather than a wrong answer.
*
+ * The `Filter` can skip them on the rows the reader says it has checked.
The planner asks for
+ * that only where the scan returns columnar batches. `requiredSchema` then
ends with a nullable
+ * boolean column named [[FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME]],
which is not a column
+ * of the relation, and the `Filter` evaluates each storage filter as
`IF(checked, true, filter)`.
+ * - True says every storage filter was evaluated on the row and kept it.
+ * - False or null says the row was not checked, and the `Filter` decides
it as if the column
+ * were not there. So a reader that does not know the column, and reads
it as missing from
+ * the file, is still right.
+ * A reader that marks rows writes the column itself rather than decoding it
from the file. A file
+ * holding a column of that name is not supported, see
+ * [[FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME]]. The column is in the
`requiredSchema` of the
+ * ordinary reader too, the one the caller builds when this returns `None`,
so that reader has to
+ * read it as missing or mark it as well.
+ *
* Being optional is also an obligation. A storage filter is evaluated out
of the plan's order,
* without the conjuncts that precede it, so it can raise an error on a row
those conjuncts would
* have rejected, which is an error a plain scan never raises. A reader must
not fail the query
@@ -344,6 +358,14 @@ object FileFormat {
*/
val OPTION_RETURNING_BATCH = "returning_batch"
+ /**
+ * The column a reader that applies storage filters marks the rows it has
checked in. See
+ * [[FileFormat.buildReaderWithStorageFilters]]. The name is reserved, as
the row-index metadata
+ * column's is. A relation with a column of that name is offered no storage
filters, and a file
+ * holding one outside the relation's schema is not supported.
+ */
+ val STORAGE_FILTER_CHECKED_COLUMN_NAME = "_tmp_storage_filter_checked"
+
/**
* Schema of metadata struct that can be produced by every file format,
* metadata fields for every file format must be *not* nullable.
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
index 7b782a05488a..5cbc33a86111 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
@@ -32,9 +32,9 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.trees.TreePattern.{PLAN_EXPRESSION,
SCALAR_SUBQUERY}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.classic.Strategy
-import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan,
WholeStageCodegenExec}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DoubleType, FloatType, StructType}
+import org.apache.spark.sql.types.{BooleanType, DoubleType, FloatType,
StructType}
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.collection.BitSet
@@ -158,24 +158,27 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
* filters do, rather than the spelling the query used.
*
* They stay in the post-scan `Filter` as well, the way a pushed data filter
does. The reader is
- * offered them, not obliged to honor them, so the plan keeps the exact
check. What it costs is
- * evaluating the conjunct a second time for the rows the scan emits. What
it buys is a reader
- * free to give up on a file it cannot read in part, or on a stretch of rows
whose survivors cost
- * too much to keep track of, without the answer depending on it.
+ * offered them, not obliged to honor them, so the plan keeps the check.
What it buys is a
+ * reader free to give up on a file it cannot read in part, or on a stretch
of rows whose
+ * survivors cost too much to keep track of, without the answer depending on
it. Where the scan
+ * also gets a column for the reader to mark the rows it checked in, the
`Filter` skips them on
+ * those rows, see [[FileFormat.buildReaderWithStorageFilters]].
*
- * Two of the conditions are per scan, and failing either offers nothing:
+ * Three of the conditions are per scan, and failing any offers nothing:
* - [[FileFormat.supportsStorageFilterPushdown]] holds. That is where a
format reads the conf
* that enables this, so a format's own conf never decides for another
format, and asking it
* first keeps everything below off the path of a scan that will not use
it.
- * - [[FileFormat.supportBatch]] holds for the schema the reader will see,
- * `partitionSchema ++ outputDataSchema`, which is the schema a format's
reader builder derives
- * its own vectorized-read decision from. Late materialization needs a
batch read, so this asks
- * about batch support rather than naming a format.
+ * - No column of the relation is named
[[FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME]],
+ * compared case-insensitively. That name is reserved for the column
above.
+ * - [[FileFormat.supportBatch]] holds for `partitionSchema ++
readerDataSchema`, which is the
+ * schema a format's reader builder derives its own vectorized-read
decision from. Late
+ * materialization needs a batch read, so this asks about batch support
rather than naming a
+ * format. The checked column is left out, since every built-in batch
read takes a boolean.
*
* The rest are per conjunct:
* - It is deterministic, which every one of `normalizedFilters` is. The
reader drops the rows
- * the conjunct rejects and the post-scan `Filter` evaluates it again on
the rows it keeps, so
- * the two evaluations have to agree.
+ * the conjunct rejects and the post-scan `Filter` takes its word on the
rows it checked, so
+ * the reader's evaluation has to be the one the `Filter` would make.
* - It references at least one column, and every column it references is a
projected data
* column. A reference to something the scan does not read cannot be
evaluated by the reader.
* - [[FileFormat.supportsStorageFilter]] accepts it. That is where the
expression shapes and
@@ -191,10 +194,14 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
normalizedFilters: Seq[Expression],
fsRelation: HadoopFsRelation,
readDataColumns: Seq[Attribute],
- outputDataSchema: StructType): Seq[Expression] = {
+ readerDataSchema: StructType): Seq[Expression] = {
val sparkSession = fsRelation.sparkSession
if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return Nil
- val resultSchema = StructType(fsRelation.partitionSchema.fields ++
outputDataSchema.fields)
+ if (fsRelation.schema.fieldNames.exists(
+ _.equalsIgnoreCase(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME))) {
+ return Nil
+ }
+ val resultSchema = StructType(fsRelation.partitionSchema.fields ++
readerDataSchema.fields)
if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema))
return Nil
val dataAttrs = AttributeSet(readDataColumns)
@@ -350,23 +357,35 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
- val outputDataSchema = (readDataColumns ++
generatedMetadataColumns).toStructType
-
// Offered conjuncts become `storageFilters` on the scan and stay in the
post-scan Filter too.
- // This runs here because eligibility depends on `outputDataSchema`.
- val storageFilters =
- storageFiltersFor(normalizedFilters, fsRelation, readDataColumns,
outputDataSchema)
+ // This runs here because eligibility depends on the data columns the
reader reads.
+ val readerDataColumns = readDataColumns ++ generatedMetadataColumns
+ val storageFilters = storageFiltersFor(
+ normalizedFilters, fsRelation, readDataColumns,
readerDataColumns.toStructType)
+ // The column the reader marks the rows it checked in, see
+ // `FileFormat.buildReaderWithStorageFilters`. It is added only where
the scan still returns
+ // columnar batches with it, as `FileSourceScanExec.supportsColumnar`
decides, so the column
+ // never changes how the scan runs. Above a row-based scan, the Project
that drops the
+ // column would copy every surviving row.
+ val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
+ }.filter { checked =>
+ val conf = fsRelation.sparkSession.sessionState.conf
+ val output = readerDataColumns ++ partitionColumns ++
constantMetadataColumns :+ checked
+ conf.wholeStageEnabled && !WholeStageCodegenExec.isTooManyFields(conf,
output.toStructType)
+ }
// The output rows will be produced during file scan operation in three
steps:
- // (1) File format reader populates a `Row` with `readDataColumns` and
- // `fileFormatReaderGeneratedMetadataColumns`
+ // (1) File format reader populates a `Row` with `readDataColumns`,
+ // `fileFormatReaderGeneratedMetadataColumns` and
`storageFilterCheckedColumn`
// (2) Then, a row containing `partitionColumns` is joined at the end.
// (3) Finally, a row containing `fileConstantMetadataColumns` is also
joined at the end.
// By placing `fileFormatReaderGeneratedMetadataColumns` before
`partitionColumns` and
// `fileConstantMetadataColumns` in the `outputAttributes` we make these
row operations
// simpler and more efficient.
- val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
- partitionColumns ++ constantMetadataColumns
+ val outputDataColumns = readerDataColumns ++ storageFilterCheckedColumn
+ val outputDataSchema = outputDataColumns.toStructType
+ val outputAttributes = outputDataColumns ++ partitionColumns ++
constantMetadataColumns
// Rebind metadata attribute references in filters after the metadata
attribute struct has
// been flattened. Only data filters can contain metadata references.
After the rebinding
@@ -417,12 +436,27 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
val metadataAlias =
Alias(KnownNotNull(CreateStruct(structColumns.toImmutableArraySeq)),
FileFormat.METADATA_NAME)(exprId = metadataStruct.exprId)
+ // Every column of the scan but the flattened metadata ones, which the
struct replaces.
+ val flattenedMetadata = AttributeSet(generatedMetadataColumns ++
constantMetadataColumns)
execution.ProjectExec(
- readDataColumns ++ partitionColumns :+ metadataAlias, scan)
+ outputAttributes.filterNot(flattenedMetadata.contains) :+
metadataAlias, scan)
}.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.
+ val postScanFilters = storageFilterCheckedColumn match {
+ case Some(checked) =>
+ val offered = ExpressionSet(storageFilters)
+ afterScanFilters.toSeq.map { f =>
+ if (offered.contains(f)) If(checked, Literal.TrueLiteral, f) else f
+ }
+ case None => afterScanFilters.toSeq
+ }
// bottom-most filters are put in the left of the list.
- val finalFilters =
afterScanFilters.toSeq.reduceOption(expressions.And).toSeq ++ stayUpFilters
+ val finalFilters = postScanFilters.reduceOption(expressions.And).toSeq
++ stayUpFilters
val withFilter = finalFilters.foldLeft(withMetadataProjections)((plan,
cond) => {
execution.FilterExec(cond, plan)
})
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala
index d312986e2bfc..4381bf03e92d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala
@@ -23,8 +23,9 @@ import org.apache.spark.{SparkContext, SparkEnv,
SparkException, TaskContext, Ta
import org.apache.spark.executor.Executor
import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
import org.apache.spark.sql.catalyst.expressions.{And, BasePredicate,
BloomFilterMightContain, BoundReference, Expression, Predicate}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
-import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.types.{BooleanType, DataType, StructType}
import org.apache.spark.util.Utils
/**
@@ -202,6 +203,10 @@ object ParquetStorageFilter {
"a storage filter must not have a key column named " +
s"${ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME}, because the
reader writes row " +
"indexes over that column, so what it reads back depends on how its
row group was read")
+ require(!keyFields.exists(f => f.name ==
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME),
+ "a storage filter must not have a key column named " +
+ s"${FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME}, because the reader
writes the rows " +
+ "it checked over that column")
val indexMap = originalOrdinals.zipWithIndex.toMap
val remapped = expr.transform {
@@ -211,6 +216,20 @@ object ParquetStorageFilter {
new ParquetStorageFilter(originalOrdinals.toArray, remapped, metrics,
maxSplicedRowGroupBytes)
}
+ /**
+ * The index in `schema` of the column the reader marks the rows it checked
in, or -1 when the
+ * scan has none, see `FileFormat.buildReaderWithStorageFilters`. The reader
looks it up by name,
+ * as it does the row-index column, and writes booleans into it, so another
type is a planner bug.
+ */
+ def checkedColumnIndex(schema: StructType): Int = {
+ val index =
schema.fieldNames.indexOf(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME)
+ if (index >= 0 && schema.fields(index).dataType != BooleanType) {
+ internalError(s"${FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME} must be
a boolean " +
+ s"column, but it is ${schema.fields(index).dataType.catalogString}")
+ }
+ index
+ }
+
/**
* Whether `dt` is usable as a storage-filter key column type, which is
whether the reader's
* per-type value copier handles it. [[isSupportedStorageFilter]] asks this
for the planner and
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala
index 542cd95e59a4..2179ac12be11 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala
@@ -50,13 +50,14 @@ import org.mockito.invocation.InvocationOnMock
import org.apache.spark.{SparkEnv, SparkException, TaskContext,
TaskKilledException}
import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
import org.apache.spark.paths.SparkPath
-import org.apache.spark.sql.{sources, DataFrame, QueryTest, Row, SparkSession}
+import org.apache.spark.sql.{classic, sources, DataFrame, QueryTest, Row,
SparkSession}
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
AttributeReference, BloomFilterMightContain, BoundReference, Cast, Coalesce,
EqualTo, Expression, GreaterThanOrEqual, In, IsNull, LessThan, LessThanOrEqual,
Literal, Or, Predicate, Rand, Remainder, SecondsToTimestamp, Substring,
UnaryExpression, XxHash64}
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
AttributeReference, BloomFilterMightContain, BoundReference, Cast, Coalesce,
CodegenObjectFactoryMode, EqualTo, Expression, GreaterThanOrEqual, If, In,
IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Or, Predicate, Rand,
Remainder, SecondsToTimestamp, Substring, UnaryExpression, XxHash64}
import org.apache.spark.sql.catalyst.expressions.aggregate.BloomFilterAggregate
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.catalyst.plans.logical.{Filter => LogicalFilter}
-import org.apache.spark.sql.execution.{CollapseCodegenStages,
ColumnarToRowExec, FileSourceScanExec, FilterExec, LocalLimitExec, SparkPlan,
WholeStageCodegenExec}
+import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
+import org.apache.spark.sql.catalyst.plans.logical.{Filter => LogicalFilter,
LogicalPlan}
+import org.apache.spark.sql.execution.{CollapseCodegenStages,
ColumnarToRowExec, ExecSubqueryExpression, FileSourceScanExec, FilterExec,
LocalLimitExec, SparkPlan, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
FileFormat, FileSourceStrategy, OutputWriterFactory, PartitionedFile}
import org.apache.spark.sql.execution.metric.SQLMetric
@@ -65,7 +66,7 @@ import org.apache.spark.sql.functions.col
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
-import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.spark.util.Utils
import org.apache.spark.util.sketch.BloomFilter
@@ -536,6 +537,23 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
bloomLiteralOf(42L),
new XxHash64(Seq(rowIndexAttr)))),
"a bloom over a column named like the row-index column must not be
offered")
+
+ // The column the reader marks the rows it checked in, which the planner
adds as a boolean. The
+ // reader looks it up by name and writes booleans into it, so another type
fails.
+ val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+ assert(ParquetStorageFilter.checkedColumnIndex(requested.add(checkedName,
BooleanType)) == 2)
+ assert(ParquetStorageFilter.checkedColumnIndex(requested) == -1)
+ val notBoolean = intercept[SparkException] {
+ ParquetStorageFilter.checkedColumnIndex(requested.add(checkedName,
LongType))
+ }
+ assert(notBoolean.getMessage.contains("must be a boolean column"),
notBoolean.getMessage)
+ // Nor can a key column have its name, since the reader writes its marks
into that slot.
+ val checkedKey = intercept[IllegalArgumentException] {
+ createFilter(Seq(keyAtLeast(0L)), StructType(Seq(
+ StructField(checkedName, LongType, nullable = false),
+ StructField("v", StringType, nullable = false))))
+ }
+ assert(checkedKey.getMessage.contains(checkedName), checkedKey.getMessage)
}
// A runtime bloom's `Literal` holding one long key, built the way the
planner builds one. It is
@@ -945,13 +963,15 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}.sum
}
- // The rows the scans with storage filters excluded, by both row metrics.
Through `collect(plan)`,
+ // The rows the scans with storage filters excluded, by both row metrics
unless `metrics` names
+ // others. Through `collect(plan)`,
// which finds those scans in a plan AQE wrapped too.
- private def rowsExcludedByStorageFilters(plan: SparkPlan): Long =
+ private def rowsExcludedByStorageFilters(
+ plan: SparkPlan,
+ metrics: Seq[String] =
Seq(StorageFilterMetrics.ROWS_EXCLUDED_BY_ROW_GROUP,
+ StorageFilterMetrics.ROWS_EXCLUDED_WITHIN_ROW_GROUP)): Long =
collect(plan) {
- case s: FileSourceScanExec if s.storageFilters.nonEmpty =>
- s.metrics(StorageFilterMetrics.ROWS_EXCLUDED_BY_ROW_GROUP).value +
- s.metrics(StorageFilterMetrics.ROWS_EXCLUDED_WITHIN_ROW_GROUP).value
+ case s: FileSourceScanExec if s.storageFilters.nonEmpty =>
metrics.map(s.metrics(_).value).sum
}.sum
// Counts BloomFilterMightContain expressions inside
FileSourceScanExec.storageFilters.
@@ -1924,9 +1944,184 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ // The `(k, v)` schema with the column the planner adds for the reader to
mark the rows it checked
+ // in. A reader built over it fills that column.
+ private def checkedSchema(valueType: DataType = StringType): StructType =
+ kvSchema(valueType =
valueType).add(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME, BooleanType)
+
+ // Reads `path` through the format, the way a scan with `storageFilters`
over `schema` does, under
+ // the SQL confs in `conf`. Returns each row's key, as `keyOf` reads it,
with what the checked
+ // column says for the row.
+ private def readChecked(
+ path: String,
+ schema: StructType,
+ storageFilters: Seq[Expression],
+ conf: Map[String, String] = Map.empty,
+ keyOf: (ColumnarBatch, Int) => Long = (b, i) => b.column(0).getLong(i))
+ : Seq[(Long, Boolean)] = withSQLConf(conf.toSeq: _*) {
+ val checked =
schema.fieldIndex(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME)
+ val file = PartitionedFile(
+ InternalRow.empty, SparkPath.fromPathString(path), 0, new
File(path).length())
+ drain(formatReader(schema, Nil, storageFilters)(file)) { it =>
+ it.flatMap { case batch: ColumnarBatch =>
+ val marks = batch.column(checked)
+ (0 until batch.numRows()).map { i =>
+ assert(!marks.isNullAt(i), "the reader must mark every row")
+ (keyOf(batch, i), marks.getBoolean(i))
+ }
+ }.toSeq
+ }
+ }
+
+ // The keys in `keys`, each with the mark `checked`.
+ private def marked(keys: Seq[Long], checked: Boolean): Seq[(Long, Boolean)] =
+ keys.map(_ -> checked)
+
+ test("the checked column is true exactly for the rows of a row group the
filter was applied to") {
+ // The post-scan Filter skips the storage filter on a row marked true, so
true must mean the
+ // filter kept that row. A row group the filter was given up on comes back
whole and must be
+ // marked false. The two row groups of the fixture show that the mark is
per row group, so a
+ // give-up in the first leaves the second its own. Batches of 4 rows split
each row group, and
+ // every batch of it must be marked alike.
+ withTempDir { dir =>
+ val rows = (0L until 40L).map(i => (i, i.toString))
+ val path = writeSingleParquetFile(dir, rows.toDF("k", "v"),
+ rowGroupSize = ParquetWriter.DEFAULT_BLOCK_SIZE,
+ options = Map(ParquetOutputFormat.BLOCK_ROW_COUNT_LIMIT -> "20"))
+ assert(footerOf(path).getBlocks.asScala.map(_.getRowCount) == Seq(20L,
20L),
+ "the fixture must be two row groups of 20 rows")
+ val schema = checkedSchema()
+ val k = BoundReference(0, LongType, nullable = true)
+ def filtersOn(key: Expression): Seq[Expression] =
Seq(GreaterThanOrEqual(key, Literal(10L)))
+ def read(key: Expression = k, cap: String = "64MB"): Seq[(Long,
Boolean)] =
+ readChecked(path, schema, filtersOn(key), conf = Map(
+
SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key -> cap,
+ SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> "4"))
+
+ // Applied to both row groups, whose survivors are spliced.
+ assert(read() == marked(10L until 40L, checked = true),
+ "the survivors of a filtered row group must be marked true")
+ // A cap of 100 bytes gives splicing up in both, since 9 bytes per
buffered key and the 40 of
+ // the one range pass it. It keeps the filter, since the range alone
fits.
+ assert(read(cap = "100b") == marked(10L until 40L, true),
+ "a row group that gave only splicing up still returns exactly its
survivors")
+ // The bytes show it is that case, since giving splicing up reads the
key column again.
+ def bytesUnder(cap: String): Long = readUnderCap(path, schema,
filtersOn(k), cap, 4)._2
+ assert(bytesUnder("100b") > bytesUnder("64MB"),
+ "a cap of 100 bytes must give splicing up, or the case above proves
nothing")
+ // A cap of 39 bytes does not fit the range either, which gives the
filter up in both.
+ assert(read(cap = "39b") == marked(0L until 40L, false),
+ "a row group whose ranges pass the cap must be marked false")
+ // An evaluation error on key 5 gives up the first row group alone.
+ assert(read(key = OnKey(k, at = 5L, error = new SparkException(_))) ==
+ marked(0L until 20L, checked = false) ++ marked(20L until 40L, true),
+ "a row group given up on an evaluation error must be marked false, and
only that one")
+ }
+ }
+
+ test("the checked column is false for a row group read without its offset
index") {
+ // A row group the footer declines, with no offset index, and one whose
offset index parquet
+ // then fails to parse both come back whole, the latter through phase 2's
retry.
+ withTempDir { dir =>
+ val blocks = (0 until 3).map(b => ((b * 100 + 1L) to (b * 100 +
100L)).map(i => (i, i)))
+ val schema = checkedSchema(valueType = LongType)
+ val evenKeys = Seq(EqualTo(
+ Remainder(BoundReference(0, LongType, nullable = true), Literal(2L)),
Literal(0L)))
+ def keys(block: Int): Seq[Long] = blocks(block).map(_._1)
+
+ val noIndex = writeParquetFileByHand(dir, blocks, hasOffsetIndex = (_,
_) => false)
+ assert(readChecked(noIndex, schema, evenKeys) == marked((0 until
3).flatMap(keys), false),
+ "every row group of a file with no offset index must come back whole
and marked false")
+
+ // The middle row group's value index is damaged. Phase 2 needs it to
read that column over
+ // the survivors, and the retry then reads the row group whole.
+ val damaged = writeParquetFileByHand(dir, blocks, (_, _) => true)
+ breakOffsetIndex(damaged, blockIdx = 1, columnIdx = 1)
+ val (result, warnings) = withLogged(partialReadWarning) {
+ readChecked(damaged, schema, evenKeys)
+ }
+ assert(warnings == 1, s"the read must have gone through the retry; got
$warnings")
+ val even = (block: Int) => keys(block).filter(_ % 2 == 0)
+ assert(result ==
+ marked(even(0), checked = true) ++ marked(keys(1), false) ++
marked(even(2), true),
+ "the damaged row group must come back whole and marked false, and the
others filtered")
+ }
+ }
+
+ test("the checked column for a file the filter is not applied to row by
row") {
+ withTempDir { dir =>
+ val schema = checkedSchema()
+ val k = BoundReference(0, LongType, nullable = true)
+ val atLeastTen = Seq(GreaterThanOrEqual(k, Literal(10L)))
+ // A file with none of the projected non-key columns, which the reader
declines.
+ val keysOnly = writeSingleParquetFile(dir, spark.range(0, 20).toDF("k"),
+ rowGroupSize = 64 * 1024L)
+ assert(readChecked(keysOnly, schema, atLeastTen) == marked(0L until 20L,
checked = false),
+ "a file the reader declines must come back whole and marked false")
+
+ // A file with none of the key columns, whose constant predicate is
decided once. Its keys
+ // are null, so it is read through the value column.
+ val valuesOnly = writeSingleParquetFile(dir,
+ spark.range(0, 20).selectExpr("CAST(id AS STRING) AS v"), rowGroupSize
= 64 * 1024L)
+ val valueOf: (ColumnarBatch, Int) => Long =
+ (b, i) => b.column(1).getUTF8String(i).toString.toLong
+ assert(readChecked(valuesOnly, schema, Seq(IsNull(k)), keyOf = valueOf)
==
+ marked(0L until 20L, true),
+ "a file whose constant the predicate keeps must be marked true")
+ val throwsOnNull = Seq(GreaterThanOrEqual(AssertNotNull(k), Literal(0L)))
+ assert(readChecked(valuesOnly, schema, throwsOnNull, keyOf = valueOf) ==
+ marked(0L until 20L, false),
+ "a file whose constant the predicate throws on must be marked false")
+ }
+ }
+
+ test("a primitive file column named like the checked column is never
decoded") {
+ // A user-given schema, or a case-insensitive match, can pair the checked
column with a column
+ // a file holds. The reader marks the rows itself and never decodes that
column, here a long
+ // one that a boolean vector could not take. Nor does the column count as
a non-key column, so
+ // a file with no other one is still declined.
+ withTempDir { dir =>
+ val schema = checkedSchema()
+ val atLeastTen = Seq(GreaterThanOrEqual(BoundReference(0, LongType,
nullable = true),
+ Literal(10L)))
+ val name =
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME.toUpperCase(Locale.ROOT)
+ val withColumn = writeSingleParquetFile(dir,
+ spark.range(0, 20).selectExpr("id AS k", "CAST(id AS STRING) AS v",
s"id AS $name"),
+ rowGroupSize = 64 * 1024L)
+ assert(readChecked(withColumn, schema, atLeastTen) == marked(10L until
20L, checked = true),
+ "the file's column must not be read, and the reader must mark the
survivors itself")
+ val keysAndColumn = writeSingleParquetFile(dir,
+ spark.range(0, 20).selectExpr("id AS k", s"true AS $name"),
rowGroupSize = 64 * 1024L)
+ assert(readChecked(keysAndColumn, schema, atLeastTen) == marked(0L until
20L, false),
+ "a file whose only other column is named like the checked one must
still be declined")
+ }
+ }
+
+ test("off heap: the checked column's vector is off heap and is freed with
the reader") {
+ // The reader puts a vector of its own in the slot, in its memory mode,
like the rest of the
+ // batch. The batch is emitted over that vector, so the reader frees it.
That the slot holds
+ // the reader's vector rather than the one built for the missing column,
which reads null, is
+ // what `readChecked`'s null check shows.
+ withTempDir { dir =>
+ val path = writeKeyParquetFile(dir, 1L to 20L)
+ var marks: ColumnVector = null
+ // The conf is set outside the task, since a task sees a read-only one.
+ val rows = withSQLConf(SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key ->
"true") {
+ asTask(TaskContext.empty()) {
+ readChecked(path, checkedSchema(), Seq(keyAtLeast(10L)),
+ keyOf = (b, i) => { marks = b.column(2); b.column(0).getLong(i) })
+ }
+ }
+ assert(rows == marked(10L to 20L, checked = true), s"got $rows")
+ assert(marks.isInstanceOf[OffHeapColumnVector],
+ s"the checked column must be off heap; got ${marks.getClass}")
+ assert(address(marks) == 0, "the checked column's vector must be freed
by close")
+ }
+ }
+
test("FileSourceStrategy leaves a non-deterministic bloom in the post-scan
Filter") {
- // The reader drops the rows a storage filter rejects and the post-scan
Filter evaluates it
- // again on the rows the reader keeps, so the two evaluations have to
agree, and a
+ // The reader drops the rows a storage filter rejects and the post-scan
Filter takes its word on
+ // the rows it checked, so the reader's evaluation has to be the one the
Filter would make. A
// non-deterministic conjunct has to stay behind. No producer builds one
today, hence the
// hand-built plan. `InjectRuntimeFilter`'s blooms hash join keys, which
are deterministic.
withTempDir { dir =>
@@ -2010,6 +2205,228 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ // An attribute of `plan`'s output, by name.
+ private def columnOf(plan: LogicalPlan, name: String): Attribute =
+ plan.output.find(_.name == name).getOrElse(fail(s"no $name in the output
of $plan"))
+
+ // A runtime bloom holding the key 42, over `key`, as the planner builds one.
+ private def bloomOn(key: Expression): Expression =
+ BloomFilterMightContain(bloomLiteralOf(42L), new XxHash64(Seq(key)))
+
+ // What `FileSourceStrategy` plans for `logical`.
+ private def planned(logical: LogicalPlan): SparkPlan =
+ FileSourceStrategy(logical).headOption
+ .getOrElse(fail(s"FileSourceStrategy did not plan $logical"))
+
+ test("FileSourceStrategy skips an offered conjunct on the rows the reader
marks as checked") {
+ // The scan gets a column for the reader to mark the rows it checked in,
and the post-scan
+ // Filter evaluates the offered conjunct as `IF(checked, true, conjunct)`.
A conjunct that is
+ // not offered stays as it is, and the column goes no further than the
Filter. The bloom names
+ // `K` where the relation says `k`, as the query can spell it, so the
Filter's conjunct and the
+ // one offered to the scan differ in name.
+ withTempDir { dir =>
+ val path = writeKeyParquetFile(dir, 0L until 50L, rowGroupSize = 64 *
1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val bloom = bloomOn(columnOf(relation, "k").withName("K"))
+ val notOffered = EqualTo(columnOf(relation, "v"), Literal("42"))
+ val physical = planned(LogicalFilter(And(bloom, notOffered), relation))
+ val scan = scanOf(physical)
+ val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+ val checked = scan.output.find(_.name == checkedName)
+ .getOrElse(fail(s"the scan must output the checked column; got
${scan.output}"))
+ assert(scan.requiredSchema.fieldNames.toSeq == Seq("k", "v",
checkedName),
+ s"the reader must be asked for the column too; got
${scan.requiredSchema}")
+ val conjuncts =
+ physical.collect { case f: FilterExec =>
splitConjunctivePredicates(f.condition) }.flatten
+ assert(conjuncts.exists(_.semanticEquals(If(checked,
Literal.TrueLiteral, bloom))),
+ s"the offered conjunct must be skipped on checked rows; got
$conjuncts")
+ assert(conjuncts.exists(_.semanticEquals(notOffered)),
+ s"a conjunct not offered must stay as it is; got $conjuncts")
+ assert(!physical.output.exists(_.name == checkedName),
+ s"the column must not leave the scan's Filter; got
${physical.output}")
+ }
+ }
+ }
+
+ test("a null mark leaves the offered conjunct as it was, an And around it
included") {
+ // A reader that does not know the checked column reads it as null. The
conjunct after the
+ // bloom must then run only where it ran before, so not on a row the bloom
rejects. An `Or`
+ // would not do, since `NULL OR false` is null, and an `And` goes on past
a null.
+ val checked =
AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME, BooleanType)()
+ val k = AttributeReference("k", LongType)()
+ val input = Seq(checked, k)
+ val row = InternalRow(null, 7L)
+ val bloom = bloomOn(k)
+ assert(!Predicate.create(bloom, input).eval(row),
+ "the bloom must reject 7, or this proves nothing")
+ val after = IsNotNull(OnKey(k, at = 7L, error = new SparkException(_)))
+ val modes = Seq(CodegenObjectFactoryMode.CODEGEN_ONLY,
CodegenObjectFactoryMode.NO_CODEGEN)
+ modes.foreach { mode =>
+ withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> mode.toString) {
+ val skipped = And(If(checked, Literal.TrueLiteral, bloom), after)
+ assert(!Predicate.create(skipped, input).eval(row),
+ s"a null mark must reject the row the bloom rejects, without the
conjunct after ($mode)")
+ intercept[SparkException] {
+ Predicate.create(And(Or(checked, bloom), after), input).eval(row)
+ }
+ }
+ }
+ }
+
+ test("FileSourceStrategy offers nothing to a relation with a column named
like the checked one") {
+ // The scan adds a column of that name for the reader to fill, so a column
of the relation's own
+ // must not be taken for it, in any case. With nothing offered, the scan
gets no such column.
+ withTempDir { dir =>
+ val name =
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME.toUpperCase(Locale.ROOT)
+ val path = writeSingleParquetFile(dir,
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v",
s"true AS $name"),
+ rowGroupSize = 64 * 1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val scan = scanOf(planned(LogicalFilter(bloomOn(columnOf(relation,
"k")), relation)))
+ assert(scan.storageFilters.isEmpty, s"nothing must be offered; got
${scan.storageFilters}")
+ assert(!scan.requiredSchema.fieldNames.contains(
+ FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME),
+ s"nor a checked column added; got ${scan.requiredSchema}")
+ }
+ }
+ }
+
+ test("FileSourceStrategy adds the checked column only to a scan that stays
columnar with it") {
+ // The Filter and the Project that drops the column are fused into
whole-stage codegen only
+ // above a columnar scan. Above any other, that Project would copy every
surviving row, so the
+ // scan gets the filter but no column, and the Filter evaluates the bloom
on every row. The
+ // field limit counts the whole output, so a partition column counts too.
+ withTempDir { dir =>
+ val path = new File(dir, "parted").getAbsolutePath
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v", "id
% 3 AS p")
+ .repartition(1).write.partitionBy("p").parquet(path)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val logical = LogicalFilter(bloomOn(columnOf(relation, "k")), relation)
+ // `k`, `v` and the partition column `p` are 3 fields, and 4 with the
checked column.
+ Seq(
+ ("4", "true", true, true),
+ ("3", "true", false, true),
+ ("2", "true", false, false),
+ ("100", "false", false, false)
+ ).foreach { case (maxFields, wholeStage, withColumn, columnar) =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_MAX_NUM_FIELDS.key -> maxFields,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage) {
+ val scan = scanOf(planned(logical))
+ val setting = s"a field limit of $maxFields and whole-stage
codegen $wholeStage"
+ assert(scan.storageFilters.size == 1,
+ s"$setting must still offer the bloom; got
${scan.storageFilters}")
+ assert(scan.requiredSchema.fieldNames.contains(
+ FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME) == withColumn,
+ s"$setting must ${if (withColumn) "add" else "leave out"} the
checked column; got " +
+ scan.requiredSchema)
+ assert(scan.supportsColumnar == columnar,
+ s"the checked column must not change whether $setting reads
columnar batches")
+ }
+ }
+ }
+ }
+ }
+
+ Seq(false, true).foreach { withMetadata =>
+ test("the post-scan Filter still filters the rows the reader gave up on " +
+ s"(_metadata selected = $withMetadata)") {
+ // A runtime bloom is redundant with its join, which would drop such
rows anyway, so this
+ // filters on a bloom alone, against a plain read. The scan's metrics
show whether the filter
+ // was applied. A cap of 39 bytes gives the filter up in every row group
with a survivor, and
+ // the post-scan Filter must then filter those rows itself. A partition
column puts the
+ // checked column between columns of the scan's output, and with
`_metadata` selected, the
+ // Project that builds it sits below the Filter and must keep the
checked column.
+ withTempDir { dir =>
+ val path = new File(dir, "parted").getAbsolutePath
+ rows400.toDF("k", "v").withColumn("p", col("k") % 3)
+ .repartition(1).write.partitionBy("p")
+ .option(ParquetOutputFormat.BLOCK_SIZE, 256L).parquet(path)
+ val columns = Seq(col("k"), col("v"), col("p")) ++ (if (withMetadata) {
+ Seq(col("_metadata.row_index").as("ri"),
col("_metadata.file_name").as("file"))
+ } else {
+ Nil
+ })
+ val relation = spark.read.parquet(path).select(columns:
_*).queryExecution.analyzed
+ val filtered = LogicalFilter(bloomOn(columnOf(relation, "k")),
relation)
+ // Plans and reads under `confs`, and returns the plan with its rows.
+ def run(confs: (String, String)*): (SparkPlan, Seq[Row]) =
withSQLConf(confs: _*) {
+ val df = classic.Dataset.ofRows(spark, filtered)
+ (df.queryExecution.executedPlan,
df.collect().toSeq.sortBy(_.getLong(0)))
+ }
+ val pushdown = SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key
+ def excludedWithinRowGroups(plan: SparkPlan): Long =
+ rowsExcludedByStorageFilters(
+ plan, Seq(StorageFilterMetrics.ROWS_EXCLUDED_WITHIN_ROW_GROUP))
+
+ val (_, plain) = run(pushdown -> "false")
+ assert(plain.exists(_.getLong(0) == 42L) && plain.size < rows400.size,
+ s"the bloom must keep 42 and drop most rows, or this proves nothing;
got $plain")
+
+ val (appliedPlan, applied) = run(pushdown -> "true")
+ assert(countBloomFiltersInStorageFilters(appliedPlan) == 1,
+ s"the bloom must be offered to the scan.\nPlan:\n$appliedPlan")
+ assert(applied == plain, "the filter applied must return what a plain
read returns")
+ assert(excludedWithinRowGroups(appliedPlan) > 0,
+ "the filter must have dropped rows inside a row group, or this
proves nothing")
+ // The marks the read path wrote, read off the scan below the Filter.
A reader that was not
+ // handed the column reads it as null, and every check above would
still pass.
+ assert(checkedMarks(scanOf(appliedPlan)).forall(_.contains(true)),
+ "the rows of a row group the filter was applied to must be marked
true")
+
+ val (givenUpPlan, givenUp) = run(pushdown -> "true",
+
SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key ->
"39b")
+ assert(givenUp == plain, "the filter given up must return what a plain
read returns")
+ assert(excludedWithinRowGroups(givenUpPlan) == 0,
+ "a cap of 39 bytes must give the filter up in every row group with a
survivor")
+ }
+ }
+ }
+
+ // The marks a scan's read path writes in the checked column, read off the
scan itself, one per
+ // row, None for a null. Fails when the scan wrote none.
+ private def checkedMarks(scan: FileSourceScanExec): Seq[Option[Boolean]] = {
+ val ordinal = scan.output.indexWhere(_.name ==
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME)
+ assert(ordinal >= 0, s"the scan must have the checked column; got
${scan.output}")
+ val marks = scan.executeColumnar().flatMap { batch =>
+ val column = batch.column(ordinal)
+ (0 until batch.numRows()).map(i =>
Option.when(!column.isNullAt(i))(column.getBoolean(i)))
+ }.collect().toSeq
+ assert(marks.nonEmpty, "the scan must have returned rows, or this proves
nothing")
+ marks
+ }
+
+ test("a runtime bloom from a join is skipped on the rows the reader marks as
checked") {
+ // `InjectRuntimeFilter`'s bloom holds a scalar subquery rather than a
literal. The post-scan
+ // Filter's conjunct has to match the one offered to the scan for the skip
to happen at all.
+ withBloomFilterTables {
+ withSQLConf(
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val (plan, rows) = runBloomFilterJoin()
+ assert(rows.map(r => (r.getLong(0), r.getLong(1))).toSeq == Seq((5L,
5L)),
+ s"the query must return the joined row; got ${rows.mkString(", ")}")
+ val scan = collect(plan) { case s: FileSourceScanExec if
s.storageFilters.nonEmpty => s }
+ .headOption.getOrElse(fail(s"no scan was offered the
bloom.\nPlan:\n$plan"))
+ val checked = scan.output.find(_.name ==
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME)
+ .getOrElse(fail(s"the scan must output the checked column; got
${scan.output}"))
+ val conjuncts =
+ collect(plan) { case f: FilterExec =>
splitConjunctivePredicates(f.condition) }.flatten
+ val skipped = conjuncts.collect {
+ case If(mark, Literal.TrueLiteral, bloom: BloomFilterMightContain)
+ if mark.semanticEquals(checked) => bloom
+ }
+ assert(skipped.size == 1 &&
skipped.head.exists(_.isInstanceOf[ExecSubqueryExpression]),
+ s"the runtime bloom must be skipped on checked rows.\nPlan:\n$plan")
+ assert(checkedMarks(scan).forall(_.contains(true)),
+ "the rows of a row group the filter was applied to must be marked
true")
+ }
+ }
+ }
+
// ----- Generic reader plumbing for the coverage tests below -----
// Writes a single-column (`k`) parquet file from a SQL expression over
`id`, avoiding the need
@@ -2051,8 +2468,9 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
.flatMap(_.getEncodings.asScala).toSet
// Reads every batch, projecting each row through `extract`. `storageFilter`
may be null, which
- // builds the plain vectorized reader rather than the late-materialization
one. Every read helper
- // in this suite goes through here.
+ // builds the plain vectorized reader rather than the late-materialization
one. Every helper that
+ // builds a reader by hand goes through here, except `readChecked`. That one
needs the Hadoop
+ // initialize, to project a column the file does not have.
//
// `tryInitializeResource` closes the reader if anything inside throws and
leaves it open
// otherwise, which is the contract these helpers need. The caller closes it
once its assertions
@@ -3311,9 +3729,10 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
test("column-index filtering off: the filter keeps only the row groups it
empties") {
// `parquet.filter.columnindex.enabled=false` is the escape hatch for a
file whose page index is
// wrong, and it has to cover phase 2 as well as phase 0. Phase 2 reads
part of a row group
- // through the offset index, which parquet consults whatever that conf
says, so a wrong index
- // there would pair a row's key with another row's values. The post-scan
Filter cannot catch
- // that, since the key it sees is the right one.
+ // through the offset index, which parquet consults whatever that conf
says. So a wrong index
+ // there 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.
//
// What the filter keeps in that case is the row groups it empties, which
needs no index at all.
// So the two arms differ in granularity, not in correctness. With the
conf on the scan returns
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
