You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/09 23:14:00 UTC
(spark) branch master updated: [SPARK-60034][SQL] Skip a storage filter on the rows the Parquet reader already checked
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 <pe...@gmail.com>
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 <pe...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org