fix(core): legacy 2-level lists, log-block predicates, and schema promotion

[linliu-code]

Stacked on #639–#668 — review only the last five commits.

Three changes to the parquet paths, each wiring up code that was already in the tree with tests but no caller.

1–2. Legacy 2-level lists (fix)

A parquet file with an array<map> column written by parquet-avro fails to read:

Map cannot be repeated

Both readers, both file kinds — base files and parquet log blocks. The file is simply unreadable.

That writer defaults to write-old-list-structure=true, encoding the column as a 2-level legacy list whose element is a REPEATED map group:

required group obj_ids (LIST) {
  repeated group array (MAP) {              // element IS the map, and REPEATED
    repeated group key_value (MAP_KEY_VALUE) { required key; required value; }
  }
}

parquet-rs takes the repeated child for the list wrapper and the map for the element, then dispatches it to visit_map, which rejects a REPEATED map unconditionally. arrow-rs and arrow-cpp both reject this schema, so the encoding genuinely is legacy — but Hudi's own reader accepts it, and tables already written this way must stay readable.

The reject can't be avoided at the Arrow level, so the parquet schema is rewritten before the Arrow build: the legacy repeated map becomes a synthetic repeated list wrapper around a required map. The footer parse never trips the reject; only the Arrow build does, so normalizing in between fixes every reader built from that metadata.

Nothing about the data changes. Every leaf keeps its definition and repetition levels and its DFS position, so the same bytes decode to the same values and the original row groups are reused as they are.

Commit 1 is shared code — the legacy reader, the v2 engine, file listing, schema resolution and statistics all read base files through ParquetBaseFileReader, so it fixes the reader in use today. Commit 2 is the same fix for parquet log blocks.

3. Predicate pushdown into parquet log blocks (feat)

The base file read pushes a row filter down (#662); a parquet log block read did not, so every row of every block was decoded and handed to the merge even when the predicate had already ruled it out.

The safety decision stays with the merge-on-read reader, the only place that knows. A log record can update a row, so a predicate may be evaluated before the merge only when the merge cannot change its answer. Log blocks exist only on merge-on-read, so the copy-on-write arm of can_push_row_filter() is unreachable here and the gate reduces entirely to a predicate over primary keys — immutable across upserts. Anything else is left for the post-merge filter.

Carried as a builder rather than a filter, because the predicate has to be matched against each block's own schema, which doesn't exist until its footer is read. A builder that declines reads every row.

Both the log file reader and the block decoder take it through a setter defaulting to unset, so the log scanner and every other caller are unaffected.

4. A partial-update block stays narrow (test)

Such a block carries only the columns that were written, and the merge distinguishes "not in this update" from "set to null". The block's own schema is the only place that signal lives.

Decoding was already against that schema, so this adds no behavior — it pins it. Upstream needs an explicit check because it decodes against a required schema; this crate did not, which made the behavior correct by construction and therefore easy to lose. Commit 5 is exactly the change that would have lost it.

5. Resolve a log block up to the promoted schema (fix)

A table whose column was promoted int → long could not be read:

column 'num' type Int32 incompatible with target Int64

Both the base file and the older log blocks predate the promotion, and neither was brought up to the current type. Two halves:

• Log blocks — Avro defines int→long as a promotion, so the required schema is handed to the decoder and the block resolves as it is read. Blocks carrying IsPartial are excluded, which is what commit 4 guards.
• Base file — parquet, so Avro resolution can't reach it. It goes through batch evolution, which knew how to widen a float but not an int. Widening an integer is exact, so it casts directly rather than through a string the way float→double must for parity with Java.

The fixture reads a base parquet, a log block written before the promotion, and one written after, and compares the merged result against the Spark snapshot — including a value beyond what an int can hold.

Tests

Two legacy-list reads against the real captured fixture — a genuine Hudi-written block in the rejected encoding — one per path.

One end-to-end pushdown case: the fixture holds keys k1/k2/k3, and a _hoodie_record_key = k2 predicate must return exactly that row. It exercises the whole chain — reader decides it's safe, hands it to the log file reader, which hands it to the block decoder — so it fails if any link drops it. That matters here: this is exactly how base-file pushdown was silently lost during the port.

Three unit tests on the Avro resolution: int→long, the same through a ["null", T] union (the shape every real Hudi column has), and no reader schema leaving the writer's type alone.

Both new behaviors verified non-vacuous by mutation:

normalization disabled  →  ParquetError(ArrowError("Map cannot be repeated"))   FAILED
pushdown wiring cut     →  row count mismatch: actual=2 expected=1              FAILED

Full workspace green: 1194 lib + 79 table-read + 39 datafusion + 21 + 12. No new clippy findings in the changed lines (the two in log_record_reader.rs are at 218–258, pre-existing).

🤖 Generated with Claude Code