feat(core): serve Hudi metadata tables through reader v2, sharded and memory-bounded - #706
Merged
yihua merged 34 commits intoSep 3, 2026
Merged
Conversation
linliu-code
force-pushed
the
feat/bounded-concurrent-read-bench
branch
2 times, most recently
from
August 31, 2026 00:25
37b6879 to
f803ae0
Compare
linliu-code
changed the title
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #706 +/- ##
==========================================
+ Coverage 79.25% 79.28% +0.03%
==========================================
Files 122 126 +4
Lines 12069 12311 +242
==========================================
+ Hits 9565 9761 +196
- Misses 2504 2550 +46 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
yihua
force-pushed
the
feat/bounded-concurrent-read-bench
branch
from
September 2, 2026 16:15
f8fd7cb to
1b0541b
Compare
This was referenced Sep 2, 2026
yihua
reviewed
Sep 2, 2026
yihua
left a comment
yihua
left a comment
Contributor
There was a problem hiding this comment.
Thanks for working on this — and for restructuring the branch onto current main; the +5.5k diff now reads as one coherent change: the metadata table served through reader v2 with shard routing, the five-source valid-instant set, rollback/restore as first-class timeline actions, and a scan-wide memory budget shared by the core and DataFusion paths.
The core work here is very solid — the Java hashCode port with its UTF-16 and i32::MIN edge cases, the mutation-hardened per-source tests on the valid-instant set, and the measured admission coefficients are all careful in ways that are easy to get wrong. The inline comments cluster around three themes: two places where the valid-instant set diverges from Java's semantics (the pending-set scope for source 2, and error handling on unreadable rollback metadata); the benchmark harness, which ships with a broken allocation cap, a compare script that crashes on the harness's own output, and a spill detector that always fires on the default directory; and internal ENG-/machine references that need to come out before this can land in an ASF repo. The PR description also predates the rebase (it still describes 28 commits / +17.4k and the old #702-based composition) and is worth refreshing.
This is one review pass, not an approval — I'd like another look after the valid-instant-set questions are settled.
A metadata partition was read from exactly one file slice and anything else was refused. That held for `files`, which is one file group in practice. It does not hold for the partitions that shard -- record index above all, and secondary index -- whose design is many file groups with keys hashed across them. The fixture's own record_index has ten. The reader never cared how many slices there were; only this check did. The fan-out is not new code. `Table::read_file_slices_bounded` already read slices with a ceiling, in order, chunk by chunk, and its reasoning is worth keeping: order is preserved because a caller concatenating batches should not see rows move with scheduling, and it chunks rather than sliding because the futures borrow their slice and the higher-ranked lifetimes then defeat `Send` inference for any caller that spawns the read. That logic is now `util::concurrency::bounded_in_order`, and both the table scan and the metadata read call it -- one implementation rather than two that drift. Tying the closure's input lifetime to the slice is load-bearing: a plain `Fn(&T) -> Fut` cannot name the borrow its future holds, and the compiler rejects it with an unsatisfiable `'1 must outlive '2`. The same `hoodie.read.file.slice.read.concurrency` bounds both, so a sharded metadata partition cannot open more readers at once than a table scan already would. Concatenation of a single slice returns its batch untouched -- no copy added to the path that existed before sharding was supported. Tested on a `files` partition made genuinely multi-slice, by copying its base file under a second file id; slice discovery is a storage listing, so this needs no timeline edit, which on table version 8 is Avro-encoded. The assertion is a doubled row count, which separates three outcomes: the old refusal errors, a one-slice read returns N, a correct read returns 2N. The test asserts its own premise first, and caught a wrong one -- the partition holds two base files in a single file group, not two groups. Mutation-checked with a positive control: reading only the first slice fails with 4 against 8, restoring the refusal fails with its error, and the unmutated tree passes. What this does not do is serve partitions other than `files`: the pruner is still pinned to it, which is ENG-47771's scope. This is the execution model, so that routing has somewhere to route to.
Java's String.hashCode over UTF-16 code units with wrapping i32 arithmetic, then abs(abs(h) % n), matching mapRecordKeyToFileGroupIndex. Vectors come from Java's published hashCode values rather than from this code, including polygenelubricants, whose hash is exactly i32::MIN and is the only input the doubled abs exists for. Parked: the metadata read layer was narrowed to parity with the existing reader, which serves one file group, so nothing calls this yet.
With a metadata partition now readable across several file slices, a key lookup still opened every one of them. A partition that shards -- record index has ten file groups in the fixture -- hashes each key to exactly one, so nine of those ten reads return nothing. The routing is cherry-picked from `wip/mdt-file-group-routing`, not rewritten, as its own ticket asks: Java's `String.hashCode` over UTF-16 code units with wrapping `i32` arithmetic, then `abs(abs(h) % n)`, with vectors from Java's published values including `polygenelubricants`, whose hash is exactly `i32::MIN` and is the only input the doubled `abs` exists for. Two traps it avoids are silent rather than loud: iterating Rust `char`s hashes a non-BMP character as its scalar and routes it elsewhere, and computing in `i64` diverges once the hash exceeds `i32`. `slices_for_keys` turns that into a selection. Slices are sorted by file id first, because the function returns a position among the shards while the listing order is the storage's -- Hudi embeds the shard number in the file id (`record-index-0003-0`), so a lexicographic sort recovers it. Indexing an unsorted list sends a key to the wrong shard, which returns no rows rather than an error, and that is the failure this sort prevents. An empty key set still opens every slice. That is not an omission: a prefix lookup cannot know which shard holds a match, because sharding is by the full key, and Java says the same at `getRecordsByKeyPrefixes:239`. Fan-out width is asserted as slices selected rather than inferred from rows returned, which is what the ticket asks for: a wrong shard and an empty shard both return no rows, so counting rows cannot tell them apart. One test also checks the selected shard is the one the hash names, not merely some shard, and another checks selection does not change when the listing order is reversed. Mutation-checked with a positive control: dropping the sort, dropping the shard deduplication, and always scanning every slice each fail, and each is caught by a different test.
The metadata table filters its log blocks by set membership, not by a window. That set has holes -- a pending data instant is excluded while instants on either side are included -- and members from outside the data timeline entirely, such as metadata-only indexing delta commits. `InstantRange` could express only bounds, and no bound admitting both neighbours can exclude what sits between them. Why this matters beyond tidiness: for the `files` partition a window is safe, because every listed file is gated downstream on `CompletionTimeView::is_committed`, so a file named by an uncommitted block is dropped anyway. The other five partitions have no such gate. A `column_stats` record drives pruning directly, so a record from a block written for a still-pending instant would be used as-is, and wrong statistics prune away files that hold matching rows -- a silently wrong query result rather than a failure. `InstantRange::exact_match` mirrors Java's `RangeType::EXACT_MATCH`: when the set is present, bounds are ignored entirely and membership is the whole test. Both predicates consult it, since the log-block gate falls back to the lexicographic one for instants that will not parse, and the two disagreeing would make the filter depend on whether a timestamp happened to be parseable. The test that matters is the one a window cannot pass: a set admitting two instants and excluding the one between them, paired with a control showing the equivalent window really does admit that middle instant. Without the control the assertion would be true for reasons unrelated to the change. Building that control found its own bug: the window's start was written `20250100000000000`, a day zero, which failed to parse. The fixture was wrong, not the assertion, and it failed loudly rather than quietly passing. Mutation-checked with a positive control: ignoring the set in either predicate, and letting an empty set admit everything, each fail -- the last mattering because a filter handed no instants that admitted everything would be a no-op wearing the shape of a filter. Nothing builds the valid-instant set yet; that is the metadata layer's next piece, and it needs this to have somewhere to put it.
…les one Slice discovery was pinned to the `files` partition by a filter written into the reader. Nothing about listing slices depends on which partition is being listed; the partitions that shard differ only in how many come back and how their records decode. `partition_reader` takes the partition name, and `files_partition_reader` becomes a thin wrapper over it. This deliberately does not widen what is publicly served. `files` remains the only partition with a decoded record type and the only one with a caller. What the parameter buys now is a test that could not be written before: shard routing checked against the fixture's real record index -- ten file groups, written by Hudi, with Hudi's own file-id naming. That matters because the existing routing tests use synthetic `FileSlice`s. They prove the selection logic, but not that the sort recovering shard order survives real ids like `record-index-0003-0`. It does, and the mutation shows the difference is not theoretical: dropping the sort selects `record-index-0004-0` where the hash names `record-index-0002-0` -- a shard that exists, holds records, and simply does not hold this key. A lookup there returns nothing, which is indistinguishable from a key that was never written. The test states what it does not do: `record_index` has no decoded record type yet, so which keys each shard holds is unknown here. It pins discovery and selection, not decoding. Mutation-checked with a positive control: dropping the file-id sort fails on the selected shard, and restoring the partition pin fails on the premise -- one slice where the fixture has ten.
`Action` had three variants, and the parser rejected everything else. A `.rollback` or `.restore` file therefore failed to parse -- and the loader discards a parse failure at debug level rather than propagating it, so such a table opens fine and those instants are simply invisible. That invisibility is why this is needed. Hudi's metadata table counts as valid the commits a rollback rolled back: their log blocks were written, rolled back and re-applied, and excluding them drops metadata records that are genuinely there. Under a partition like column_stats, dropped records mean pruning away files that hold matching rows -- a silently wrong query result rather than a failure. Deliberately not added to `DEFAULT_LOADING_ACTIONS`. A rollback is not a commit, and that constant fills `completed_commits`, which drives file-slice discovery and the commit-visibility gate. Adding them there would change every existing read; they are loaded only by a caller that names them, which is how the single-action selectors already work. Verified against Hudi's own source rather than inferred: `getValidInstantTimestamps` (HoodieTableMetadataUtil:2081) names rollback and restore instants as two of the five sources, and `getRollbackedCommits:2158` reads each instant's metadata for the commits it rolled back. Tested both directions. The round-trip pins every action against the on-disk suffix, because a mismatch is discarded rather than reported -- exactly how these two went unseen. The second test pins that neither is in the default set, and asserts the set is non-empty first so it cannot pass vacuously. Mutation-checked with a positive control: a wrong suffix fails the round-trip, and adding Rollback to the default set fails with the set printed. This is the representation only. Reading a rollback instant's metadata to learn which commits it rolled back is the next piece, and that metadata is Avro (HoodieRollbackMetadata.avsc, `commitsRollback: array<string>`).
The metadata table counts as valid the commits a rollback rolled back: their log blocks were written, rolled back and re-applied, so excluding them drops records that are genuinely there. Java reads them from the rollback instant's own metadata (`getRollbackedCommits`, HoodieTableMetadataUtil:2158); this is the same read. A subset of the schema on purpose. Hudi's record also carries timings, per-partition detail and a version, none of which the valid-instant set consults, and Avro ignores fields the target struct does not name -- so a schema that grows does not break this. The file is an Avro object container because Hudi writes it with `DataFileWriter`, which is the shape the commit-metadata reader already handles, so this follows that rather than inventing a second convention. The fixture is built from Hudi's own schema files, copied verbatim into the test data, rather than a schema typed into the test. That is what made it a real check: three separate mistakes were caught by the schema rather than by me. `version` is a union of [Int, Null], so Int is branch 0 and the 1 I first wrote selected Null. The reference to `HoodieInstantInfo` is unqualified in the file, not fully qualified as I assumed. And the two schemas live in separate files, so neither the writer nor the reader can resolve the reference alone. That last one is worth recording, because it is a claim about the format rather than the test: Hudi writes the container header with `SpecificDatumWriter`, whose schema has the named type defined at first use. A header carrying an unresolved cross-file reference is not something a real `.rollback` file contains, so the fixture inlines the dependency to match what Hudi actually writes. Mutation-checked with a positive control: returning an empty result instead of parsing, and swallowing a decode failure as an empty result, each fail. The second matters most -- an empty set there silently drops every commit the rollback covered, which is the failure this whole filter exists to prevent.
A restore is made up of several rollbacks, so the commits it rolled back are the union of its rollbacks' own. Java walks `getHoodieRestoreMetadata().values()` and flattens each entry's `commitsRollback` (HoodieTableMetadataUtil:2178-2183); this does the same, reusing `RollbackMetadata` rather than redeclaring those fields, because the schema genuinely nests that record type. Flattened rather than kept per-instant: the valid-instant set is a set, and which rollback covered which commit does not affect membership. The fixture nests two rollbacks under two different keys, each naming a different commit, so the test distinguishes "flattened everything" from "took the first map entry" -- a single-entry fixture would pass either way. Mutation-checked in both of those directions, and a positive control on the restored tree. The schema inliner is now general. `HoodieRestoreMetadata` references `HoodieRollbackMetadata`, which in turn references `HoodieInstantInfo`, so the splice runs in dependency order and asserts each reference is actually present -- a splice that silently matched nothing would leave the reference unresolved and fail later with a message about the format rather than about the fixture. Also removes two helpers left from the first approach to building these fixtures, which clippy caught only once the disk had room to run it.
…e is empty Java prefers the completed rollback instant's metadata and falls back to its `.requested` plan when that file is empty, taking `getInstantToRollback().getCommitTime()` (HoodieTableMetadataUtil:2165-2172). Without the fallback that recoverable case becomes a lost set of commits, and a lost commit here means its log blocks are excluded from the metadata read. A plan names one instant where completed metadata can name several, which is why it is the fallback rather than the primary source, and why both readers exist. `instantToRollback` is nullable in Hudi's schema, so a plan naming no instant yields nothing rather than failing -- a reader that errored there would reject a timeline Hudi considers valid. Both cases are tested, and the empty one is not merely the absence of the first: it pins that the reader distinguishes "no instant named" from "could not read". Mutation-checked with a positive control: making the plan always report no instant fails the test that expects one.
Five sources, ported from `HoodieTableMetadataUtil.getValidInstantTimestamps` (:2081): completed data instants; completed metadata delta commits whose data instant is not pending; commits rolled back by the data table's rollbacks and restores, bounded below by the earliest valid instant as Java bounds it; the metadata table's own rollback and restore instants; and metadata delta commits carrying the `00000000000000` sentinel prefix. Each source is its own function, and that shape is not tidiness -- it is what makes them testable. The first attempt tested the union, and it could not fail: three mutations, one per source, all passed. The reason is worth recording, because it will recur. Sources 1, 2 and 5 overlap almost entirely on a real table -- a data commit and the metadata delta commit that records it share a timestamp -- so every member of the union survives dropping any single source, and a membership assertion over the union proves nothing about which source supplied it. Tested per source, the same three mutations all fail. Source 2's exclusion needed an input the fixture does not contain. The fixture has zero pending data instants, so the branch that excludes a metadata commit whose data instant is still pending was never reached by real data -- which is exactly why deleting that exclusion passed every fixture-only test. The pending set is injected in one test so the branch has an input that reaches it, and the test asserts the instant is included beforehand, so the exclusion is a change rather than a coincidence. `Timeline::load_instant_bytes` is added because the rollback, restore and plan records are not commit metadata, so `load_instant_metadata` cannot decode them. Not yet wired into a read, and marked `#[allow(dead_code)]` rather than given a synthetic caller. The set is assembled by the data table while `read_files_partition` runs on the metadata table, so attaching it means threading it down that chain -- a change to the read path, and the next step. `MetadataTableV2Reader::with_valid_instants` is the receiving end and is already in place.
…a window The set built in the previous commit now reaches the reader. `InstantRange::exact_match` gets its first caller, and the metadata read stops using a bounded window it cannot express the right thing with. Threading it took a parameter through four functions because the set needs both timelines -- the data table's completed and pending instants, and the metadata table's own -- while the read runs on the metadata table alone. It is built at the two data-table entry points and in `Table::get_file_slices`, each of which holds both, and passed down from there. Clippy found a real defect in the process: `read_files_partition_batch` took the set and never used it, so the Arrow path would have kept the window while the decoded path used the set. Two read paths silently disagreeing about which log blocks are readable is worse than either choice alone. Verified by mutation rather than by the suite passing: making the set come back empty fails six tests. That is the check that matters here, because a set that is merely *wrong* rather than empty would still let most reads succeed -- and a suite that passes either way would prove only that the filter is inert. `get_file_slices` now takes eight parameters, one over clippy's limit, allowed with the reason: the eighth cannot be derived inside a view that holds only the metadata table's timeline, and bundling the other seven is a wider refactor than this change.
…rces Sources 3 and 4 ran on every metadata read while being exercised by nothing: the fixture carries zero rollback and zero restore instants, so the code that walks them had no input that reached it. A rollback instant is now written into a copy of the table -- real Avro `HoodieRollbackMetadata`, built against Hudi's own schema by the same fixture code the reader tests use. The rolled-back commit is a timestamp that appears nowhere else on either timeline, and the test asserts that before asserting anything else. That is what makes the result attributable: no other source can supply it, so its presence in the set can only have come from source 3. Without that, the assertion would be the union test's mistake again -- overlapping sources make membership prove nothing about origin. Mutation-checked, each caught by the assertion naming its own source: never reading rolled-back commits fails source 3, dropping the metadata table's own rollback instants fails source 4, and making the earliest-instant bound exclude everything fails source 3 as well -- that last one being the bound Java applies, which nothing else in the suite touches. All five sources now have a test that fails when the source is removed.
Key pushdown selects which data blocks to fetch, but every Avro-carrying HFile committed here has exactly one data block. With one block a seek and a full scan fetch the same bytes, so a reader that ignored the predicate outright passed the whole corpus. Rows cannot close that gap. Block selection over-includes and `decode_window` filters afterwards, so a predicate read returns the same rows whether it fetched one block or nine — a row assertion reproduces the hole it is meant to catch. Bytes are the only observable difference. Observing them needs the reader to outlive the read, so `read_stream` splits into the existing `open` plus `stream_from`, and `reads_handle` hands out a fetcher clone that shares the reader's counts. Both are on the production path: the test opens and reads exactly as `read_stream` does. The fixture is one 9-block `record_index` HFile, committed as the file rather than the 40 MB table around it, with the SQL that generated it. Measured: 37,593 bytes over 1 block against 240,563 over 9, same row. Mutation-checked twice, both failing the test: ignoring the predicate when selecting blocks makes the two arms fetch 240,563 bytes each; dropping it before `decode_window` returns 60,000 rows where 1 is required. Against a single-block fixture the test refuses to run rather than passing vacuously.
The corpus reaches three metadata partition types: files, partition_stats and secondary_index. `bloom_filters` is the fourth and nothing covered it — and it is the one whose payload is a raw byte buffer rather than a struct of scalars, so a decoder that works for the others can still fail on it. The fixture is one file slice, base HFile plus the log written after it, from a 4,000-row table. Scale matters here: the same script at 600,000 rows writes a 4.55 MB bloom-filter base against 158 KB, since a filter is sized by the entries it holds. The row count is the assertion that gives the log side weight — 2 records from the base and 2 from the log, under keys the base does not have. Mutation-checked by dropping the log from the split, which reads 2 and fails. The test deliberately does not claim to pin partition routing: rebuilding the record context for `files` instead changes nothing observable, because `type` is decoded from the record rather than chosen by the context. That is recorded in the doc comment as measured rather than asserted.
…t needs Adds the file-group benchmark to the repository and extends it into a gate for one requirement: given a fixed amount of memory, a read may be slower but must not fail. `fg-bench` existed as untracked local work and, in an older form, on an internal branch. The two had drifted, and the local copy is the one imported because it carries a macOS `ru_maxrss` fix the other lacks -- the BSDs report bytes where Linux reports kilobytes, so reading it as kilobytes overstates peak RSS by 1024x and turns a 21 MB process into a 21 GB one. In a harness that exists to measure memory that is not cosmetic. It is ported to the public `FileGroupReader` surface. The untracked copy reached into `file_group::reader_v2`, which is `pub(crate)`, so it compiled against neither repository -- likely why it was never committed. The knobs survive the move because they are config keys rather than constructor arguments. What does not survive is `HoodieReadStats`: stage timings and the merge map's accounted peak are not public, and widening an internal module to suit a benchmark is the wrong trade. Spill is instead observed from outside, by watching the spill directory, so a passing run can still distinguish "stayed under budget because it spilled" from "stayed under because the data never got large". Three additions make it a gate rather than a report: - `--slice-concurrency` reads slices through `buffer_unordered`, mirroring the DataFusion fan-out, so the harness can reproduce the real shape. Default 1 is sequential with no coordination cost, the baseline a bounded-memory claim must not regress. - `--max-rss-bytes` fails the run when peak RSS exceeds a declared budget, with a non-zero exit. - `fg-gen` generates a table of a target size, because a gigabyte cannot be committed -- the checked-in fixtures are tens of kilobytes. It emits table version 6, whose commit metadata is JSON; version 8 encodes it as Avro, which a generator would have to reimplement for no gain to the read path under test. With `--log-files` it writes Hudi log files carrying Avro data blocks, so the table is merge-on-read and the read builds a real merge map. What this measures, and what it does not: `--max-rss-bytes` is an assertion made after the allocation, not an enforcement. It answers "how much did this read want", not "does this read survive on a small machine". The allocator cap that answers the second question is a separate change.
… see spill `--max-rss-bytes` asserts after the fact: it reads peak RSS once the allocation has already succeeded, so on a machine with memory to spare it answers "how much did this read want", never "does this read survive on a small machine". Nothing ever told the process no. `FG_BENCH_ALLOC_CAP_BYTES` does. A global allocator refuses past the ceiling and the runtime aborts, which is deliberately not graceful -- an abort proves the ceiling is real, and a read that wants to degrade instead has to stay under it. Off unless the variable is set, and measured at no throughput cost (579ms against 621ms on the same read, which is noise). The spill detector was wrong twice, and both ways reported "never spilled" for a read that spilled a gigabyte. It scanned one level deep while RocksDB writes into a subdirectory; and once that was fixed, it still saw nothing, because RocksDB removes its directory when the reader closes -- before the read call returns -- so a before/after sample finds an empty directory at both ends. It now samples on a thread while the read runs and keeps the high-water mark. Verified in both directions rather than only the one that confirms it: a merge-on-read slice with a 1 MiB budget reports spilled=true with a 1032 MB peak, and a copy-on-write read with no merge reports spilled=false. A detector that only ever says no is indistinguishable from a broken one. What this buys is a measurement that changes the diagnosis: with the budget honoured at 1 MiB, the merge map spills 1032 MB to disk and the resident set is still 1467 MB. Spilling moves the accounted bytes out and RSS does not follow, so the memory is not in the structure the merge budget accounts for.
Clippy on this crate never actually ran in the previous commit: the disk filled during the rocksdb build, and I read a grep count rather than the output, so build failures were mistaken for a clean gate. Run properly, it found six dead items left by the port. `configs_and_storage` built a `HudiConfigs` and a `Storage` for the hand-assembled reader that no longer exists -- the public `FileGroupReader` builds its own from the table path. The two accounting-drift constants outlived the detector they served. `ReadConfig::data_schema` was passed to the old constructor and is now derived by the reader. And `ReadOptions` was built by mutating a `Default`, which clippy rejects in favour of struct-update syntax. Verified after the change rather than assumed: the harness still reads a generated merge-on-read table (300k rows, spilled=true), the RSS gate still exits 1 on a budget it cannot meet, and the hard allocation ceiling still aborts with 134. A cleanup that quietly broke either gate would be worse than the dead code.
The memory harness on this branch answers how much a read allocates. It says
nothing about how long a thread is held, and that is the other half of the
question: the merge runs wherever its consumer runs, so a tokio worker blocked
inside it stalls every other task on that worker.
Release-only and `#[ignore]`d, because a debug build's number says nothing about
production CPU and the measurement is too slow for the normal suite:
cargo test -p hudi-core --release --lib merge_cpu_bench -- --ignored --nocapture
It builds a key-based record buffer directly rather than reading a table, so the
number is merge cost per chunk with no I/O in it — the quantity a scheduling
decision actually needs.
`--spill-dir` was watched but never passed to the reader. The reader's own default is `/tmp` (`spillable_map.rs:209`), so the two halves of the measurement looked at different places, and neither reading meant anything: - left at its default, the watcher summed `/tmp` recursively. On macOS that is `/private/tmp`, so any fixture staged there was counted as spill — this reported 1473 MiB of spill on runs that spilled nothing. - pointed at a clean directory, it reported zero while the reader went on spilling to `/tmp` unwatched. The watcher exists to separate "stayed under budget because it spilled correctly" from "stayed under because the data never got large", which is exactly the distinction the memory work turns on, and it could make neither. Passing the flag through as `hoodie.memory.spillable.map.path` makes the watched directory the one the reader writes to. Verified by positive control: an 8 MiB merge budget now reports spilled=true with 633.7 MiB, where the same run at 64 MiB and above reports zero — so the zeros are absences rather than a dead probe.
…nothing
A merge-on-read slice held 728 MiB reading a 1 GiB table, against 55 MiB for the
same data as copy-on-write. The streaming path was being asked for and not
delivered: `force_eager = !streaming || instant_range.is_some()`, and
`resolver.rs` sets that range unconditionally, so every merge-on-read read drained
the whole base file into a `Vec<RecordBatch>` regardless.
The range it materialized for admits everything. `adapter.rs` calls
`with_unbounded_end_timestamp` before building the context, so the default range
is (None, MAX] — it excludes no commit, and filtering against it cannot remove a
row.
`InstantRange::admits_all` names that case, and `force_eager` consults it. A range
that bounds something still materializes and still filters; only the case where
filtering is provably a no-op skips it.
Measured on a generated 1 GiB / 10-file merge-on-read table, 8,000,000 rows,
identical row counts before and after:
MOR 1 GiB, streaming 728 MiB -> 184 MiB
MOR 512 MiB base 308 MiB -> 62 MiB
MOR 256 MiB base 183 MiB -> 63 MiB
COW 1 GiB (control) 55 MiB -> 53 MiB
concurrency 4 1656 MiB -> 534 MiB
Peak RSS also stops tracking base-file size, which is the property that made the
old behaviour unbounded: 256 MiB and 512 MiB of base now measure the same 62 MiB.
Why skipping is safe rests entirely on the predicate, so the predicate is what the
test pins. It matters most on a table with `populate.meta.fields = false`, where
`create_commit_time_filter_mask` returns `None` and this gate is the only thing
enforcing the instant window — a predicate that called a bounding range unbounded
would silently drop that. Mutation-checked three ways, each failing the test:
returning `true` outright, ignoring the start bound, and treating any end bound as
unbounded.
Memory is bounded per merge and never per scan, so the real peak is the product of the per-merge budget, the slices open inside one engine partition, and the partitions running at once — a number no configuration expresses. On a 16-core box at defaults that is up to 64 merge maps live at once. Shrinking the per-merge budget does not fix it, and measurement says it makes it worse: tightening `hoodie.memory.merge.max.size` from 128 MiB to 32 MiB on a 1 GiB merge-on-read table raised peak RSS from 182 MiB to 229 MiB and cost six times the wall clock, because the disk tier it pushes work into consumes more memory than it frees. Slices in flight is the lever that responds, and it is linear over the same table: 181, 299, 408, 532, 781 and 1052 MiB at 1, 2, 3, 4, 6 and 8 slices. So a byte budget divides into an admission count. Cost does not track slice size. Once the base file streams rather than materializes, a 108 MiB slice costs 33 MiB while a 133 MiB slice costs 117 MiB; what separates them is log bytes, because the merge map is built from log records. Fitting the measured shapes gives a 33 MiB streaming working set plus roughly 3.2x log bytes, rounded to 4 here — over-estimating admits fewer slices, which costs throughput, and the requirement this serves is that a scan may get slower but may not fail. This is the derivation only; no call site consumes it yet. Seven tests, each mutation-checked: granting the budget per partition rather than sharing it, averaging slice cost instead of taking the worst, dropping log bytes from the estimate, letting a tiny budget admit zero, ignoring the budget, and treating an unrecorded log size as zero — which would admit the most slices exactly when the least is known.
Adds `hoodie.read.scan.max.memory.size` and consumes it, so the peak a scan can
reach is a number someone can set rather than the emergent product of three that
nobody multiplies.
The DataFusion plan derives its slice concurrency from the budget, shared across
input partitions rather than granted to each, and clamped by the existing
`hoodie.read.file.slice.read.concurrency` — this only ever lowers the fan-out.
Unset, every existing scan behaves exactly as before.
`FileSlice::log_size_bytes` reports `None` when a log file has no recorded size,
where `total_size_bytes` counts it as zero. That difference is the whole point:
counting an unlisted slice as free would admit the most slices exactly when the
least is known about them, so an unestimatable scan admits one.
The benchmark derives concurrency through the same library function rather than
its own flag, so the gate exercises the shipped decision instead of the harness.
Gate, on 1 GiB across 10 merge-on-read files, 8,000,000 rows:
256 MiB budget -> concurrency 1 190 MiB peak 4452 ms PASS
2 GiB budget -> concurrency 8 1049 MiB peak 2519 ms
no budget -> concurrency 8 1028 MiB peak FAILS a 256 MiB ceiling
The middle row is the control the requirement asks for: raising the bound must
show materially higher peak RSS, or the harness is not exercising it — 190 MiB
against 1049 MiB. The last row keeps the gate honest by still failing. The cost
of the bound is 1.77x wall time, which is the trade the requirement names: a
scan may get slower, it may not fail.
The plan-time budget cannot see what else is running. `HudiScanExec` ignored its `TaskContext` entirely, so DataFusion's accounting — which exists to make an operator wait or spill rather than allocate — could not see this scan at all. It now registers a `MemoryConsumer` and reserves for the slices it is about to open, using the same per-slice estimate the plan-time derivation uses. When the pool cannot grant the planned fan-out it grants fewer, down to one; the reservation is held for the life of the stream, so the bytes come back when the scan finishes or is dropped. One slice is admitted even when the pool grants nothing. The reservation is an estimate rather than a measurement, and declining to start on it would fail scans that would have fit — where the requirement is that a scan under pressure gets slower, not that it stops. Four tests against a `GreedyMemoryPool`, each mutation-checked: taking the plan regardless of the pool, returning zero instead of degrading to one, and skipping the reservation entirely all fail the tests, which pass unmutated.
`eb00389` claimed the DataFusion plan derived its slice concurrency from `hoodie.read.scan.max.memory.size`. It did not: the change was never in that commit. Only the benchmark consumed the budget, so every planned scan kept using the raw `hoodie.read.file.slice.read.concurrency` and the bound applied to nothing a query would run. Two things made that easy to miss, and both are addressed here. The wiring is one call whose absence compiles perfectly, and `split_into_chunks(flat_slices, input_partitions)` appears in both `scan_parquet` and `scan_hudi` — only the second builds a `HudiScanExec`, so a change aimed by string match can land in the path that does not matter. The test asserts the wiring rather than the arithmetic, which `slices_in_flight` already covers: with no budget the plan shows the default ceiling of 4, and with a 1 MiB budget it shows 1. It reads `HudiScanExec`'s own verbose display, so it checks the plan the planner built. Mutation-checked by restoring exactly what `eb00389` shipped — passing the unmodified field through — which fails it.
`read_file_slices_bounded` read `hoodie.read.file.slice.read.concurrency` raw, so `hoodie.read.scan.max.memory.size` bounded the DataFusion path and left the `Table` read path unbounded — half of the ticket's requirement that both respect it. It now derives its concurrency exactly as the plan does, so the two cannot bound a scan differently. The budget is not divided again here: one `Table` read is one partition's worth of work, and dividing twice would bound this path below what the caller asked for. The raw ceiling lookup moved inside that derivation rather than staying a method of its own, which makes the bypass a compile error instead of a silent regression. That is deliberate: the equivalent wiring on the DataFusion side was claimed, compiled, and shipped absent, and the first test written for this path missed the same mistake because it called the derivation directly instead of the read. A test can miss a bypass; `no method named file_slice_read_concurrency` cannot.
The gate deciding whether a base file falls inside an instant range could be disabled outright — `if true || base_file_in_instant_range(...)` — with the whole suite still green. So could the predicate added in 5a5c650 that skips it when the range bounds nothing. Neither had end-to-end coverage. The reason is a property of every merge-on-read fixture here: their log records update base keys. Excluding the base leaves those updates with nothing to update, so the read returns nothing whether the gate fired or not, and the two outcomes are indistinguishable. `mor_log_inserts` has log records that **insert** instead — 2,000 base rows at one commit, 40 log records at a later one under keys the base does not hold — so excluding the base leaves 40 rows rather than none. The test also turns `hoodie.populate.meta.fields` off. With it on, `create_commit_time_filter_mask` removes the same rows at row level after the merge, so both mechanisms agree and neither can be isolated. Off, that mask returns `None` — which is also the real configuration in which this gate is the only thing enforcing an instant window. Fixture is 292 KB, generated by `fg-gen` with a new `--log-key-offset` that shifts log records past the base file's key range; without it they are updates, which is the whole difficulty this closes. Mutation-checked both ways, each failing the test: the gate never excluding a base file, and `admits_all` returning true so the gate is skipped. The second is the mutation that previously passed all 1300 tests.
Rebasing onto the sharded-metadata work renamed `FileGroupMergeIterator` to `FileGroupMergeStream` and reshaped its constructor: the base source is handed to the stream rather than set on the buffer, and the chunk size is no longer a constructor argument. The timing loop now blocks on each chunk. The measurement is a comparison across cases and the parking cost is identical in each, so it shifts every number by the same small constant rather than changing which case is dearer — and this fixture has no I/O, so the future is always immediately ready and never actually parks.
`eb00389` in this branch adds `hoodie.read.scan.max.memory.size`, which bounds a scan by deciding how many file slices may be open at once. `FileGroupReader` is handed one slice per call and has no fan-out for it to divide, so setting the key there does nothing — and the C++ bridge forwards arbitrary `key=value` pairs into that constructor (`cpp/src/lib.rs:77-83`), so it is reachable from outside Rust. Accepted and silently inert is worse than absent: the caller who set it is asking to be bounded, sees success, and believes it applied. It is now refused with an error saying why and where the key does belong. The refusal sits on `new_with_options`, the caller-facing constructor, and not on `new_with_overrides`, which is how `Table` builds the readers underneath its own fan-out — there the budget is honoured one level up. That distinction is the whole content of the change, so the test asserts both halves and both are mutation-checked: dropping the rejection, and rejecting in both constructors, each fail it. The second mutation initially passed, because the table half only constructed a `Table` and `Table::new_with_options` builds no reader — it now performs a read and asserts rows come back.
`Table::read` fans out through `read_file_slices_bounded`, bounded by `hoodie.read.file.slice.read.concurrency`. `read_stream` does not: `.then` awaits each slice's future before starting the next, so exactly one slice is in flight however many the table has, and that config has no effect on this path. No comment said so. The comments immediately around it justify the filter handling and the batch size, which made the silence about concurrency read as an oversight rather than a decision. It is a decision: a streaming read exists so the whole result is never resident, and fanning out N ways would hold N slices' batches at once — spending the memory the caller chose this API to avoid. A caller who wants the slices read concurrently wants `read`. The consequence worth stating alongside it: peak memory here does not grow with slice count, so a scan memory budget has nothing to bound on this path. The test pins the consequence a caller can check — that the row total matches the eager read and does not move with the ceiling — and says in its own doc comment that it does not prove sequentiality, because nothing observable from outside distinguishes one slice in flight from four. It runs on a partitioned fixture for a reason: the first version used a single-slice table, where `.take(1)` on the slice iterator passes the test. On `V6ComplexkeygenHivestyle` that mutation fails.
It was added so the streaming path could skip materializing the base file for a range that bounds nothing, but the whole-file instant decision moved ahead of the open and the predicate ended with no production caller. Its doc comment still described the deleted path, which is the kind of stale rationale worth removing before it misleads.
Five defects in the file-group benchmark harness, each of which made a number it printed untrue rather than merely imprecise. The allocation ceiling could not be used at all. `alloc` counted only once the cap was installed, but `dealloc` subtracted from that same moment, so every allocation the tokio runtime and the environment made before `main` reached the cap was freed without ever having been added. The counter underflowed to near `u64::MAX`, every later `live > cap` test failed, and a capped run aborted with "memory allocation of 64 bytes failed" whatever its real usage. Counting unconditionally, from the process's first allocation, keeps the counter balanced. The spill detector reported the spill directory's absolute size, and defaulted to `/tmp` — so an ordinary host reported gigabytes of spill before the read began, and re-walked all of `/tmp` every 100 ms to do it. It now spills into a fresh per-run subdirectory and reports growth over what the directory held when sampling started, which is the number `spilled` claims to be. `fg-gen` wrote its log blocks at an instant it never committed. The table was readable through a standalone `FileGroupReader`, which has no timeline and admits every block, while `Table::read` silently dropped every log record — 80,000 rows against 90,000 on the same generated table. Writing the delta commit makes the two paths agree. `compare.py` indexed a `read_stats` object no report has carried since the harness moved to the public reader surface, so the documented A/B workflow raised `KeyError` on its first input. It now compares the fields the binary emits. `--async-stream` and `--streaming` documented two mechanisms and called one, so a sync-vs-async comparison measured the same path twice. Folded into one flag, since the public reader offers one streaming entry point. Also drops SKILL.md and the ENG-/milestone references, which pointed at machines, branches and trackers that do not exist for this repo. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Both make a metadata read serve listings that are missing log blocks, which is a wrong answer returned as if it were right. The pending-instant exclusion saw only three actions. Source 2 admits a completed metadata delta commit unless its data instant is still pending, and it tested `Timeline::pending_instants` — built from DEFAULT_LOADING_ACTIONS, so commit, deltacommit and replacecommit alone. Java takes `datasetPendingInstants` from `filterInflightsAndRequested()` over the whole active timeline. A metadata delta commit written for a data compaction or clean that then crashed leaves a `.compaction.inflight` on the data timeline, which the narrow view cannot represent at all and therefore admits. `Timeline::all_pending_instant_times` closes that by listing the active timeline and parsing instant times structurally, since the `Action` enum has no variant for those actions and a selector-based listing drops them before they can be counted. It is deliberately separate from `pending_instants` rather than a widening of it: that field also feeds `completion_gate_inputs`, where a wider inflight set would start rejecting log blocks the gate admits today. An unreadable rollback silently shrank the set. `commits_rolled_back_by` logged at debug and returned an empty list when the instant would not load or parse, so the commits a rollback rolled back and re-applied were dropped from the valid set and their log blocks went missing from the listing. Java raises HoodieMetadataException. It now does the same, and keeps the one fallback Java keeps: an unreadable completed rollback file falls back to its requested plan, now at warn rather than in silence.
…ocs true Follow-ups from review, none of which change what a correct read returns. `FileSystemView::get_file_slices` took a metadata table and its valid-instant set as two optional parameters, so a metadata table with no set was expressible — and fell back to an empty one, which admits no metadata log blocks at all. The strongest possible filtering, reached through an argument that reads as "no filtering". They are now one `MetadataListing`, and the test that was exercising the degraded path without being able to notice now passes a real set and asserts it is non-empty. That also brings the parameter count back under clippy's limit, so the `too_many_arguments` allow is gone. The DataFusion plan's reservation read an unrecorded log size as zero bytes, reserving only the per-slice floor for exactly the slices least is known about, while `slices_in_flight` admits one slice in the same situation. `planned_slices` now applies that one rule, with a test pinning the two together. The rollback scan awaited one GET per rollback instant in sequence, and the bound above it is the only thing limiting how many there are. The reads are now concurrent. The bound itself is unchanged and matches apache/hudi's `getValidInstantTimestamps`, which filters on the earliest valid instant alone. `bounded_read_concurrency_for` cloned every routed slice only to hand `bounded_read_concurrency` an owned slice; both now derive the log-size vector from their own borrows. Docs: `files_partition_reader` carried its superseded single-slice block above the current one, a NOTE named a `partition_reader_with_valid_instants` that never existed, `read_files_partition`'s doc block appeared twice, `bounded_read_concurrency_for`'s had a fragment spliced into the middle of another method's, and `read_file_slices_bounded` linked a `file_slice_read_concurrency` that is gone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
linliu-code
force-pushed
the
feat/bounded-concurrent-read-bench
branch
from
September 3, 2026 01:55
a50c6e6 to
72eb5e9
Compare
yihua
force-pushed
the
feat/bounded-concurrent-read-bench
branch
2 times, most recently
from
September 3, 2026 02:05
c889470 to
72eb5e9
Compare
Ticket ids and references to milestone docs that do not exist for readers of this repository, in a reader test comment, two fixture docs, and the bench harness.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Serves the Hudi metadata table through the v2 file-group reader, sharded and memory-bounded:
HoodieFileGroupReader, with multi-slice discovery for any partition and record keys routed to their shard by a faithful port of Java'sString.hashCode(UTF-16 code units, wrappingi32,abs(abs(h) % n)); thefilespartition is served end to end, and only the shards a key lookup routes to are opened.getValidInstantTimestamps(completed deltacommits, the data timeline's pending set, and instants rolled back by rollback/restore), with rollback and restore now first-class timeline actions whose Avro metadata is readable, including the requested-plan fallback when a completed rollback file is empty.hoodie.read.scan.max.memory.sizederives slices-in-flight from a per-slice cost model, applied by both the coreTable::readpath and the DataFusion plan; the plan also registers its reservation with DataFusion's memory pool.benchmark/filegroup/(bounded-memory gate, spill detection, and the MOR table generator it needs).The diff exceeds the usual size guideline because it consolidates the metadata-table-on-reader-v2 stack into one reviewable unit: its bases (#692, #700) have merged into main, and the remainder previously split across #702, #705, and #707 is carried here (those PRs can be closed).
How are the changes test-covered
Per-source tests on the valid-instant set (rollback, restore, pending-instant exclusion), a value-for-value parity test of the v2-backed metadata reader against the existing one, shard-routing vectors from Java's own
hashCodeoutput (including thei32::MINedge case), DataFusion plan tests for budget-derived fan-out, a timeline test on the generated bench tables, and new metadata/MOR fixtures with documented generation scripts.