peter-toth opened a new pull request, #59269:
URL: https://github.com/apache/spark/pull/59269
### 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, 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 `checked OR
conjunct`. Every other conjunct stays as it is. A `Project` above the `Filter`
drops the column, added if the plan had none.
- The reader fills the column. `FileFormat.buildReaderWithStorageFilters`
documents the three values:
- true says the reader evaluated every storage filter on the row's own
values and kept it;
- false says the reader did not check the row, so the `Filter` decides it;
- null filters exactly as false does. So a reader that does not know the
column, and reads it as missing from the file, is still right. In Parquet that
is parquet-mr, which reads the scan when the vectorized reader is turned off
after planning.
**Parquet reader**
`LateMaterializationParquetRecordReader` 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.
**Planning**
- A relation with a column of that name, compared case-insensitively, is
offered no storage filters. The name is reserved, as the row-index metadata
column's is, so a file holding a column of that name outside the relation's
schema is not supported.
- 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.
- The column is left out where it alone would push the scan past
`spark.sql.codegen.maxFields`, which would cost the scan its columnar output.
The `Filter` then evaluates the offered conjuncts on every row, as before this
PR.
- 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 scan that is offered storage filters 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's marks for a row group the filter was applied to, 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;
- 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 relation with a clashing column name, which gets no storage filters and
no column;
- the column left out where it alone would cross the field limit, and kept
below it and past it;
- 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, given up in every row group that has a survivor,
and read by parquet-mr, with the scan's metrics confirming each. It has a
partition column, and runs with and without `_metadata`;
- `ParquetStorageFilter.create` rejecting a checked column that is not a
boolean, and a filter that reads it.
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;
- dropping the column from the `_metadata` `Project`;
- placing it after the partition columns in the scan's output;
- removing the name guard, the rewrite, or either `create` check;
- matching the offered conjuncts by plain equality;
- adding the column when nothing is offered;
- never leaving the column out at the field limit, or leaving it out past
the limit too;
- not closing the vector, or not putting it in the slot.
`ParquetStorageFilterSuite` passes, 130 tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
