fix(arrow): evaluate filters on columns missing from a data file against their identity partition value or initial-default

[mbutrovich]

Which issue does this PR close?

• Closes Rows are dropped when a filter references a column that is missing from the data file but has an identity partition value or initial-default #3298.
• Addresses case B of Equality deletes ignore spec projection rules for dropped and newly added delete columns #3369 for the current equality-delete path. Case A is unchanged.

What changes are included in this PR?

• Add residual_for_missing_fields in arrow/reader/predicate_visitor.rs. It replaces each predicate leaf on a top-level field that is missing from the data file with AlwaysTrue or AlwaysFalse, evaluated against the value projection returns for that field: the identity partition value, otherwise initial-default. Leaves on missing fields with neither keep the existing null handling. This is the same idea as Java's ResidualEvaluator, extended to initial-default. Like Java's, it folds the constants out of AND and OR, so the row filter doesn't evaluate them per batch.
• Apply the residual in the reader pipeline right after the field ID map is built, so the row filter, row group pruning, bloom filter pruning, and page index pruning all see the same predicate. This includes the predicate built from equality deletes.
• Leave leaves on nested fields unchanged, because a nested field also reads as null in any row where an ancestor struct is null. This keeps filtering consistent with projection, which doesn't apply nested defaults yet (RecordBatchTransformer does not apply initial-default for nested fields #3261).
• Reuse ExpressionEvaluatorVisitor for leaf evaluation and constants_map for identity partition values. Both go from private to pub(crate). There are no public API changes.

Are these changes tested?

Yes, with unit tests, reader tests, and a MemoryCatalog scan test:

• Two reader tests in row_filter.rs read a file that doesn't store b. In one, b comes from initial-default. In the other, it comes from an identity partition value. Each runs filters on b and equality deletes on b (for 7, another value, and null), with row selection off and on. Both fail without the fix.
• Residual unit tests in predicate_visitor.rs follow Java's TestMetricsRowGroupFilter cases for string and double defaults. Other cases check that the partition value takes precedence over initial-default, that bucket partition values are ignored, and that a present column, a column with no value, and a nested field are left unchanged. They also check that a constant on either side of AND or OR folds away, that an AND with no constant child is kept, that NOT is pushed into its child, and that the keep predicates built from equality deletes fold to a single constant.
• test_scan_filter_on_initial_default_column_absent_from_file in scan/mod.rs follows Java's testFilterPushdownOnInitialDefaultColumnAbsentFromFile. It appends a file, adds a column with a default through update_schema, appends a second file, and scans with c = 'US' and c IS NULL. It fails without the fix.

AI Disclosure

I wrote this PR with help from Claude but understand and support the core approach, implementation, and test coverage.

[comphead]

I compared the expected results with Java's TestMetricsRowGroupFilter and TestFilterPushDown, and they match. The inline comments cover the main points. Smaller nits:

• field_ids in residual_for_missing_fields and MissingFieldResidualVisitor only holds the missing fields that resolved to a value. A name like constant_field_ids would match the constants_map wording and set it apart from predicate_field_ids.
• A one-line comment at the residual_for_missing_fields call in pipeline.rs would help. It should say that the residual runs before the row filter and every pruning layer so they all see the same predicate.
• constants_map now runs twice per file when the file has partition data and a predicate column is missing (once in with_partition, once in the residual). It is cheap, so fine to leave.
• The read closure in assert_absent_b_reads_as_7 repeats the task and reader setup in read_once, which has two callers. A partition: Option<(Arc<PartitionSpec>, Struct)> parameter on read_once would let the helper call it directly. The helper also builds actual and expected in separate passes and calls to_string() twice per case. A single loop with assert_eq!(.., "{predicate}") would do the same in one pass.

[comphead]

Question, not for this PR. The null handling that leaves on a field with no value fall back to is not uniform. In PredicateConverter, less_than and less_than_or_eq return build_always_true() for a missing column, while greater_than, greater_than_or_eq, and eq return false. So for a column added without a default, WHERE d < 5 keeps every row of an older file and WHERE d > 5 keeps none. Java's ParquetMetricsRowGroupFilter treats a missing column as all null, and TestMetricsRowGroupFilter.testColumnNotInFile expects lt and le to be unable to match. The issue calls the current null handling correct, so I wanted to check whether the < and <= branches are intended. If not, a separate issue would be enough.

[mbutrovich]

I don't think they're intended. I checked on this branch with an optional b that has no default. For a file that doesn't store b, b < 5 and b <= 5 keep all 3 rows, and b > 5, b >= 5, b = 5, and b != 5 keep none. For a file that stores b as 3 nulls, b < 5 and b <= 5 keep none. So the result depends on whether the nulls are stored or come from a missing column. The branches date back to #295, and I don't know why < and <= differ there. Java's row-level Evaluator compares with a nulls-first comparator, so lt is true for a null value there. That could be the source, but testColumnNotInFile expects the Parquet row group filter to skip the file for both. I opened #3355 for it.

[mbutrovich]

Thanks for the review, @comphead. I merged main and pushed changes for each comment.

field_ids in residual_for_missing_fields and MissingFieldResidualVisitor only holds the missing fields that resolved to a value. A name like constant_field_ids would match the constants_map wording and set it apart from predicate_field_ids.

Renamed to constant_field_ids.

A one-line comment at the residual_for_missing_fields call in pipeline.rs would help.

Added.

constants_map now runs twice per file when the file has partition data and a predicate column is missing (once in with_partition, once in the residual). It is cheap, so fine to leave.

I left it as is.

The read closure in assert_absent_b_reads_as_7 repeats the task and reader setup in read_once, which has two callers. A partition: Option<(Arc<PartitionSpec>, Struct)> parameter on read_once would let the helper call it directly. The helper also builds actual and expected in separate passes and calls to_string() twice per case. A single loop with assert_eq!(.., "{predicate}") would do the same in one pass.

read_once now takes the ArrowReader, the partition, and the delete files, and assert_absent_b_reads_as_7 calls it. Passing in the reader lets the bloom filter helper keep its reader options and lets the new helper turn row selection on and off. The helper and the residual unit tests in predicate_visitor.rs now use a single loop with a message per case.

[comphead]

The earlier review points are addressed. The residual matches Java's ParquetMetricsRowGroupFilter.predicate for initial-default (top-level only) and ResidualEvaluator for identity partition values. CI is green and the branch merges cleanly with main.

I ran seven single-line production mutations against the new tests. Details are on the test comments below.

[comphead]

Nit: this set holds at most predicate_field_ids.len() entries, so HashSet::with_capacity(predicate_field_ids.len()) bounds it.

[mbutrovich]

Nit: this set holds at most predicate_field_ids.len() entries, so HashSet::with_capacity(predicate_field_ids.len()) bounds it.

I'd like to keep HashSet::new(), because it doesn't allocate until the first insert and with_capacity allocates up front. A common case here is a column added without a default. In that case no missing field has a value, and we return early with the set still empty, so with_capacity would allocate for nothing. When a field does have a value, the first allocation from new() holds three entries, which covers most predicates.

[comphead]

Nit: the struct doc and this field name still say "partition", but the residual now passes a row of top-level field values. Reusing the visitor is the right call. Updating the doc and renaming the field to row (or values) would keep the name accurate.

[mbutrovich]

Nit: the struct doc and this field name still say "partition", but the residual now passes a row of top-level field values.

I renamed the field to row and rewrote the struct doc. The docs on ExpressionEvaluator still say partition, since that is what it evaluates.

[comphead]

Removing the residual_for_missing_fields call from pipeline.rs fails this test, but it also fails the two reader tests in row_filter.rs, and no other mutation I tried fails only this one. It repeats the reader-level initial-default matrix (c = 'US', IS NOT NULL, IS NULL, row selection off and on). What it adds is the update_schema plumbing, which this PR does not change. Consider trimming it to one or two predicates, or dropping it and keeping the Java mirror in the reader test.

[mbutrovich]

Consider trimming it to one or two predicates, or dropping it and keeping the Java mirror in the reader test.

I trimmed it to c = 'US' and c IS NULL with row selection off. I kept it because it is the only test that goes through update_schema and TableScan. Its doc now points to the reader tests for the other predicates and for row selection.

[comphead]

Thanks @mbutrovich couple of nits, but looks good overall

[mbutrovich]

Thanks for the second pass, @comphead. I pushed changes for the inline comments and replied to each one below.