linliu-code opened a new issue, #684:
URL: https://github.com/apache/hudi-rs/issues/684
## Problem
A single file-slice read crosses the sync/async boundary three times, and
the log read is what forces all three.
| Layer | Model |
|---|---|
| Public API (`Table`, `FileGroupReader`) | async |
| Base file read | async
(`ParquetRecordBatchStreamBuilder<ParquetObjectReader>`) |
| **Log file read** | **sync** (`Read + Seek` over `StorageReader`) |
| Merge | sync (takes `Box<dyn arrow_array::RecordBatchReader + Send>`) |
The crossings are: the async parquet stream wrapped in a sync reader that
iterates via `block_on`; each sync log read spawning onto
`OBJECT_STORE_RUNTIME` and waiting on a `sync_channel(1)`; the sync merge
re-exposed as a `Stream` through `spawn_blocking` behind a depth-1 channel.
The cost is not the plumbing, it is the **non-local contracts each crossing
creates**, none of which the compiler can check:
- `get_range_blocking` needs three separate verdicts about where it may be
called from: a blocking-pool or plain thread is fine; an async task on another
runtime's worker works but starves that worker and "remains the caller's
obligation"; an `OBJECT_STORE_RUNTIME` worker is a genuine deadlock, refused by
a thread-name check. `in_object_store_runtime()`, the named worker threads, and
that whole body of reasoning exist only because sync code is reachable from
async code.
- `make_base_file_source(streaming: bool)` returns `Box<dyn
RecordBatchReader + Send>` in both branches, but one is a lazy reader that
`block_on`s during iteration ("caller MUST be in sync context") and the other
is fully drained ("safe to consume from async callers"). Same type, different
safety contract, selected by a bool.
## Why this is now tractable
#682 already converted the log path from a cursor to
**locate-then-ranged-fetch**, which is the shape parquet uses:
`LogBlockContentLocation { content_position, content_length }` plus
`LogBlockFetcher::read_content(offset, length)`. Only two sync I/O sites
remain: the header walk, and the per-block content fetch.
`StorageReader` also has **exactly one consumer**
(`file_group/log_file/reader.rs`). Nothing else in the crate uses the sync
cursor, so this work deletes the bridge outright rather than removing one of
several callers.
## Scope
The log read only. Making the merge a `Stream` is the other half and is
tracked separately, because the two are independently shippable and have very
different risk profiles.
## Proposed slices, each shippable alone
**1. Batch the content fetch.** After the gates decide, every admitted
block's `(offset, length)` is known, so the fetch can be one `get_ranges(...)`
instead of one serial round trip per block. Nothing in the crate currently
calls `get_ranges` or `get_byte_ranges`, so every admitted block is its own
round trip today, issued serially from Pass 3. On object storage that is the
dominant cost.
**Caveat that must shape the design:** this pulls directly against #682,
whose justification was peak memory (+200 MB to +40 MB, measured by
`memory_bench`, by holding one block's content at a time). Fetching every
admitted block at once gives that back. The resolution is bounded batching,
coalescing within a window of K blocks and dropping each group after decode,
which keeps most of the round-trip saving with a bounded peak.
`hoodie.memory.dfs.buffer.max.size` already exists and means something
adjacent, so it is the natural knob.
**2. Async the content fetch.** `read_content` becomes `get_range().await`.
Contained, and covers the bulk of the bytes.
**3. Async the header walk.** Replace the `Read + Seek` cursor with a
buffered async cursor: fetch a window, parse what it holds, advance. With this
done, `get_range_blocking`, `OBJECT_STORE_RUNTIME`,
`OBJECT_STORE_RUNTIME_WORKERS`, `OBJECT_STORE_RUNTIME_THREAD_NAME`,
`in_object_store_runtime()` and the three call-site rules can all be deleted.
## Known risk
The walk is not a pure forward parse. `is_block_corrupted` recovers from a
corrupt block by scanning forward for the next `MAGIC` over a 1 MiB window.
That is the one place the parse backtracks, and it is the most likely thing to
make slice 3 uglier than it looks. Worth prototyping that path first rather
than last.
## Validation
The gold parity sweep is the oracle: output must stay byte-identical across
all fixtures and both reader versions. `assert_lazy_matches_eager` already pins
"reading lazily equals reading eagerly", which is exactly the invariant these
slices must preserve. `memory_bench` measures slice 1's memory trade directly,
since it already compares eager against lazy. Mutation-check each load-bearing
change rather than trusting a green run.
## Non-goals
The bindings still need a runtime of their own: Python's process-wide one
and the cxx per-reader `current_thread` one do not go away because the core
became async. And `spawn_blocking` may still be wanted for decoding a large
block; this removes it for **I/O**, which is the part that is currently
misplaced.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
