feat(core): port the log record readers and the key-based buffer - #652
Open
linliu-code wants to merge 17 commits into
Open
feat(core): port the log record readers and the key-based buffer#652linliu-code wants to merge 17 commits into
linliu-code wants to merge 17 commits into
Conversation
Adds the context the merge-on-read file group reader needs and the resolver that derives it from a table's configs. Nothing consumes it yet: the reader itself lands in later changes, and the existing read path is untouched. Resolving the context first gives the reader a defined target before any of it is ported, and puts the read-semantics decisions in one reviewable place rather than spread across the port. Two decisions worth review: - `append_only` is rejected rather than mapped. The merge-on-read reader always merges by record key, so no merge mode reproduces "keep every version". The table config derives `append_only` whenever meta fields are disabled or no ordering field is set, which makes it reachable for ordinary tables, so guessing a mode here would silently change which rows a query returns. - `should_merge_use_record_position` stays off. Log blocks already carry record positions but nothing reads them, so merging by key is what a read does today. Deferred until something reads them: the schema handler, the record context, and the bootstrap flags. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`resolve_instant_range` reproduces the bounds `FileGroupReader::create_instant_range_for_log_file_scan` computes, so that a read through either path admits the same log blocks. Nothing enforced that: the two live in different modules and neither calls the other, so editing one would have drifted silently. Compares the `Debug` rendering rather than field by field, since `InstantRange`'s fields are private. That also means a field added to the struct is covered here for free instead of being silently skipped. Verified the assertion can fail, by mutating the resolver twice and confirming each was caught: flipping `end_inclusive`, and dropping the start timestamp. Widens the legacy method to `pub(crate)`. No behavior change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds the entry point callers will use once the merge engine exists. `read` returns `Unsupported` for now. The entry point lands ahead of the engine so the shape callers depend on is settled before anything is built against it, and so the gap is something a test can point at rather than an absence. It shares a name with the file group reader that serves reads today, which it replaces once the engine behind it is written — so it carries the name it will keep, and the eventual swap is a module deletion rather than a rename touching every call site. Until then the two are told apart by module path, and nothing outside the module uses this one. `read` fails rather than returning an empty batch: an empty batch is indistinguishable from a file slice that genuinely has no rows, which would let an unfinished reader look like it worked. Construction stays separate from reading, so callers wiring up a reader do not have to handle a failure until they ask for rows. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Three fixes from self-review: Configs that steer this crate's own behavior (`hoodie.internal.*`, `hoodie.plan.*`) were being swept into the table properties, which are meant to be what the table declares about itself. A reader has no way to tell a swept-in crate config from a real table property, so they now reach it through neither map. The module carried a blanket `allow(dead_code)` because nothing consumes it yet. That also silenced items dead by mistake — it was already hiding that `MergeMode::CommitTimeOrdering` cannot be constructed, since a table with no ordering field derives `append_only`, which the resolver rejects. Each item now carries its own allow, so a newly dead one still warns, and the unreachable variant says why it is unreachable and what makes it reachable. The reader construction test asserted a field the resolver tests already cover and named a guarantee the type system already gives, since `new` cannot fail. It now checks that the reader hands back the parameters it was built with, using non-default values so substituted defaults would fail it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Four modules with no dependencies inside the reader subsystem: read statistics, the profiling macro, the iterator mode, and the input split that names what a read covers. Ported as-is from the merge-on-read reader. Nothing consumes them yet, so each file carries a file-scope allow(dead_code); every item in them is live upstream, so a per-item allow would just be noise. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The delete context carries what a read needs to recognize a delete: the marker field and value a payload uses to mark one, resolved from table config. The output converter projects a merged batch down to the columns a caller asked for. Both are leaves — no dependencies inside the reader subsystem. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two schema helpers the merge-on-read reader needs. `avro_schema_utils` decides whether two Avro schemas are projection-equivalent, which is how the reader tells a genuine schema change from a reordering or a narrowing that needs no work. `parquet_list_norm` normalizes the two Parquet list encodings — the legacy two-level repeated group and the modern three-level one — so batches read from files written by different engines line up. Brings the two fixtures its tests read; they are force-added past the `**/data` ignore, the same way the existing fixtures under crates/test/data are tracked. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two modules that let a read span a schema change. `extended_promotion` decides which type changes a read can absorb — the widenings Hudi permits, such as int to long or a decimal gaining scale — and rejects the rest rather than producing a silently wrong value. `batch_evolution` applies a target schema to a batch read under an older one: added columns are back-filled as null, dropped ones are dropped, and promoted ones are cast through the rules above. Together they are what makes a merge-on-read slice readable when its base file and its log blocks were written under different schemas. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`row_serde` is the Arrow IPC pair the merge buffer uses to move a record to disk and back when it spills, plus the body-only variant that drops the per-record schema framing. `record_positions` decodes the positions a log block carries for its records. Hudi stores them as a base64-encoded roaring bitmap, so this adds `base64` and `roaring` — both are the storage format's own encoding rather than a choice, and neither pulls a transitive tree of its own. Nothing reads positions yet; the merge that uses them lands later. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The unit the merge works over: one record's key, its ordering value, and either its payload or the fact that it is a delete. Carries the ordering comparison the merge uses to pick a winner between two versions of a key, including the decimal rescaling that comparison needs. Ported against the shape from onehouseinc/hudi-rs-internal#112, which moved the spill serialization out of the record context so the two stop importing each other. That is why this depends on row_serde rather than record_context. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…erter `record_merger` picks a winner between two versions of a record key. Two modes: commit-time ordering, where the later commit wins, and event-time ordering, where the higher ordering value does. Deletes resolve through the same comparison, so a delete only wins if it is genuinely newer. `update_processor` counts what a merge did — inserts, updates, deletes — which is what the read statistics report. `buffered_record_converter` is the seam for turning an engine record into a buffered one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`record_context` reads the fields a merge needs out of a batch: the record key, the ordering value for a given row, and whether a row is a delete. It resolves those field names from table config once, then works per row. `row_extraction` pulls a single row out of a batch as a standalone batch, which is how a merged record leaves the buffer. Two adaptations from upstream, both because this repo is ahead: - Ordering fields resolve from `hoodie.table.ordering.fields` first, with the deprecated `hoodie.table.precombine.field` still honored. Upstream checks the deprecated key first, predating the rename. - A test that built an Arrow schema from Avro JSON through arrow-avro now goes through this crate's own Avro-to-Arrow conversion, which produces the same shape without the forked dependency. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Works out the three schemas a read needs: what the caller asked for, what must be read from storage to satisfy it, and what the merged output looks like. The required schema is wider than the requested one whenever the merge needs a column the caller did not ask for, such as an ordering field. Adds `schema::resolver::avro_json_to_arrow_schema`, which the handler needs to turn a log block's Avro schema into an Arrow one. It goes through this crate's own Avro-to-Arrow conversion rather than the forked arrow-avro the source uses. That conversion has a known gap, recorded on the new function: it marks a list's element field non-nullable even when the Avro items schema is a ["null", T] union, so a slice whose arrays hold NULL elements resolves to a schema that disagrees with its data. Fixing it changes a conversion the existing read path also uses, so it belongs in its own change with a regression test rather than riding along here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The merge map the reader holds records in while it merges a file slice. It tracks its own size and, once past the configured budget, spills entries to RocksDB rather than growing without bound. A separate peak ceiling fails the read loudly instead of letting it consume the host. Adds rocksdb as a required dependency, plus foldhash for the map's hasher, uuid for collision-free spill directory names, and tempfile for their lifecycle. rocksdb is a real cost: it bundles RocksDB 8.10 and needs libclang to build. The alternative considered was an in-memory-only tier behind a feature flag, which would have meant writing a spill backend that does not exist upstream, and a merge map that silently cannot spill is worse than one that needs a build dependency. Licenses are compatible — rocksdb is Apache-2.0, librocksdb-sys is MIT/Apache-2.0/BSD-3-Clause. Also adds CoreError::MemoryLimitExceeded, which the peak ceiling reports. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`buffer` defines what a record buffer is: it takes log blocks, merges each record against what it already holds, and hands back the result. `record_buffer` is the shared machinery every implementation builds on — the merge map, the merger, the delete context, and the update counting, wired together. `merge_iterator` drives that buffer, emitting the merged result one batch at a time rather than materializing the whole slice. Makes `log_file::log_block` public. The buffer reads block metadata directly, and the module was already public in all but declaration — its types are reachable through the log file reader. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The context added earlier was written against this repo before the reader it serves had been ported. Now that the reader is arriving, the ported context is the real target, so it replaces the placeholder and the resolver is rewired to build it. The resolver keeps its typed MergeMode and converts to the string form the context carries, so the mapping stays checked at the point where it is decided rather than becoming a bare string. Fields with no caller in this repo resolve to inert values, each with a comment saying why: no bootstrap support, no predicate pushdown into the merge path, and a completion gate that needs a timeline the caller has not loaded. Adds storage::RowFilterBuilder, the type the context's pushdown field is typed against. Nothing installs one yet. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`log_record_reader` walks a file slice's log files and decides which blocks a read admits — rollbacks remove their target, compaction blocks subsume what they compacted, and the completion gate excludes instants that never finished. `merged_log_record_reader` drives that scan into a buffer. `key_based` is the buffer that merges by record key. Three adaptations, all because upstream reads log blocks lazily and this crate reads them whole: - The Pass-1 header sweep uses the existing whole-file log reader rather than a streaming one. Same block set, more memory, and no more than the existing read path already uses. Adding the streaming reader would mean rewriting a storage type the existing reader shares. - The scan asks for an unbounded instant range, leaving admission to the gates rather than filtering twice under different rules. - The two `inflate` calls are dropped: blocks arrive with their content, so there is nothing to fetch. Makes `log_file::log_format` public, alongside `log_block` from the previous change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This was referenced Aug 2, 2026
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
1 participant
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.
Stacked on #639–#651 — their commits appear here until they merge. Review only the last commit.
log_record_reader— walks a slice's log files and decides which blocks a read admits: rollbacks remove their target, compaction blocks subsume what they compacted, the completion gate excludes instants that never finished.merged_log_record_reader— drives that scan into a buffer.buffer::key_based— the buffer that merges by record key.Three adaptations, one cause
Upstream reads log blocks lazily (windowed streaming + per-block inflate); this crate reads them whole. So:
StorageReader, a type the existing reader shares, which is exactly what this migration is trying not to do.inflatecalls are dropped — blocks arrive with content, so there is nothing to fetch.Worth noting: upstream's
new_streamingtakes anArc<ReaderContext>, which would have made the sharedlog_filelayer depend onreader_v2. Avoiding that inversion is why this takes the eager path rather than porting the streaming one.Also makes
log_file::log_formatpublic, alongsidelog_blockfrom #650.Build warning-free; 1076 lib tests green (up from 949).
🤖 Generated with Claude Code