[core] Support file index and predicate push down in DataEvolutionSplitRead

[wombatu-kun]

Purpose

Resolves the TODO in DataEvolutionSplitRead.withFilter, which was a no-op: Support File index push down (all conditions) and Predicate push down (only if no column merge).

As a result, data evolution tables never used their file indexes on read: AppendOnlyFileStoreScan evaluates the embedded index in filterByStats(ManifestEntry), but DataEvolutionFileStoreScan overrides that method and only checks row id ranges. Bloom, bitmap and bsi indexes were written for data evolution files and never read back. Format level push down did not reach the reader either, since the FormatReaderMapping.Builder was built with filters = null.

Push down is now applied wherever it can not interfere with the column merging:

• No column merge: the file index can skip the file, a BitmapIndexResult narrows the read to the matching positions, and the filters reach the format reader, mirroring RawFileSplitRead.
• Merged group: whole group skipping only. DataEvolutionFileReader zips its inner readers positionally, so dropping rows in one of them would break the column alignment. Skipping the group is sound because a logical row takes each column from exactly one file.

The existing enabledFilterPushDown flag of FormatReaderMapping.Builder already draws that line (the merge path calls build(..., false)), so passing filters into the builder needs no extra branching.

Three safety points:

• Only plain data files may skip a merged group. mergeRangesAndSort asserts areAllRangesSame(dataFiles), while a blob or vector-store file covers only a sub range and proves nothing about the other rows.
• Filters on _ROW_ID and _SEQUENCE_NUMBER are never pushed down: they are assigned from the manifest entry, and data evolution may reassign row ids, so a physical copy in the file can be stale.
• A filter is pushed to a single file only for the columns that file actually wrote. A column missing from a data evolution file is not null, its values live in another file of the same row id range, and formats such as Parquet read a column absent from the file schema as all null and would drop every row. The missing set is computed from the file writeCols, and the single file reader mappings use a dedicated cache keyed by writeCols so they can not collide with the merge path cache.

Semantics are unchanged: withFilter is a hint, and this only drops rows proven not to match.

Benchmark (throwaway, not in the diff): 200k rows in 40 row id groups, ids interleaved so min/max stats prune nothing, bloom filter, point filter. 39 of 40 groups skipped.

Scenario	Rows read	best	Speedup
Groups of one file	200,000 to 5,000	74-79 ms to 24-26 ms	~3.1x
Merged groups	200,000 to 5,000	83-90 ms to 24 ms	~3.6x

Left for a follow up: row level bitmap selection inside a merged group, which needs the bitmap translated into global row id ranges and fed through the existing rowRanges mechanism.

Tests

New DataEvolutionFileIndexTest (11 tests). Readers are created directly from the splits without executeFilter(), so an empty result proves the reader itself pruned the rows; filter values sit inside the column min/max range so the scan does not drop the split first.

Covered: a single file group skipped by a standalone .index file, a merged group skipped by an embedded index, column alignment preserved in a merged group, exact bitmap index selection, a merged group skipped after RENAME COLUMN (the filter has to be devolved by field id), a filter on a column absent from the read file returns all rows across parquet, orc and avro, a bitmap selection composed with a deletion vector, file-index.read.enabled = false, and a filter mixing _ROW_ID with a data column.

[JingsongLi]

• skipByFileIndex iterates through all files in a merge group; if any old file’s index is determined to be a mismatch, the entire group is skipped.
• However, the same field may have been overwritten by an updated file, rendering the old file effectively obsolete.
• New local regression: Old f1=a*, new f1=c*; query f1=‘c050’; expected 100 rows, actual 0 rows; consistently reproducible.

I suggest you use ChatGPT and set it to the highest review level; as far as I recall, your PR has already contained several serious errors. This is dangerous for the community.

[wombatu-kun]

Fixed in 40f9115: skipByFileIndex now evaluates each file's index only over the columns it is the newest writer of (matching the winner selection in evolutionStats and createUnionReader). Added regression test testMergedGroupKeptWhenFilterColumnOverwritten reproduces the report (old f1=a*, new f1=c*, query f1='c050' -> 100 rows).

[JingsongLi]

It is better to introduce Table tests to validate reading with filter.

[wombatu-kun]

Done d3c0e6f. Each scenario now also reads the whole plan through TableRead#executeFilter() and asserts the rows a query returns, compared against a full scan filtered in memory.

[JingsongLi]

Blocking: filters on non-projected columns can incorrectly drop matching rows

DataEvolutionSplitRead now passes fileFilters(file) to the format reader, even when a predicate field is not included in readRowType. This can produce false negatives.

I reproduced it with a single file containing f1 = a000 ... a099:

RowType readType = RowType.of(SpecialFields.ROW_ID);
List<InternalRow> rows =
        readWithFilter(table, equalF1(f1(50)), readType);

The expected result is one row with _ROW_ID = 50, but the reader returns an empty result. The same issue occurs when projecting only f0 and filtering on f1. It is also reproducible without configuring a file index, so this is caused by format predicate pushdown rather than bitmap selection or Row ID synthesis.

A pushed-down filter may leave false positives for the caller to recheck, but it must never remove matching rows. A query such as:

SELECT _ROW_ID FROM t WHERE f1 = 'a050'

can therefore return an incorrect result.

The existing testRowTrackingFilterIsNotPushedDown does not cover this case: it reads the default table row type and never projects or asserts the returned _ROW_ID.

Please separate the predicates used for file-index evaluation from those pushed into the format reader. Format predicates should only be pushed when all referenced fields are available to the physical reader, or the internal read type should include the predicate fields and project them out afterward.

Please also add regressions for:

1. Reading only _ROW_ID while filtering on a regular column.
2. Reading a projection that excludes the filter column.
3. Asserting the exact returned Row ID after bitmap selection.

The manual assignFirstRowId construction itself is reasonable and matches the production partial-write path, although the helper should ideally assert that only one normal file was produced or assign consecutive Row ID ranges when multiple files are present.

[wombatu-kun]

Done b6ec861.

Leaving the file index predicates unrestricted, which is what separating the two sets amounts to, is not safe either. DataEvolutionFileStoreScan#pruneByReadType drops the files of a row id group that write no column of the read type, so for a column outside it the file that wins the column merge can be absent from the split, and the older file left behind holds a dead copy whose index vetoes rows that do match. A predicate is therefore pushed to both the index and the format only when every field it reads belongs to the read type, and only to the files that wrote those fields. DataEvolutionTableRead#createBlobViewReader reaches that shape outside tests, it reads with a read type projected to the blob view fields alone while passing the full user predicate.

Your repro now returns the row with _ROW_ID = 50, but not only it: with readType = RowType.of(SpecialFields.ROW_ID) nothing can be pushed down at all, and executeFilter has no f1 to narrow the result with. An exact single row in that shape is only reachable through a bitmap index, which is what testBitmapSelectionReturnsExactRowId asserts, with f1 in the read type so the index applies.