feat(parquet): route reader through the object-store seam (rfc0013 green) - #233
Conversation
…een) Wire the sync `Reader` read path onto the RFC 0013 `Store` seam: `open_file` / `open_partition` now read bytes via `Store::local(parent).get_blocking(name)` → `open_bytes`, so the local read path goes through the same abstraction the S3 backend will. Behaviour-preserving — every existing RFC 0005 / RFC 0009 suite passes unchanged through the seam (RFC0013.2, read half). `Store` gains the sync→async bridge the sync storage layer needs (`object_store` is async; compaction must reach S3 per RFC0013.3, so a local-only `std::fs` shortcut won't do): `get_blocking` / `put_blocking` drive the async ops on a shared multi-thread(1) runtime, running `block_on` on a fresh OS thread via `std::thread::scope`. A plain thread never inherits the caller's tokio context, so the bridge is safe even when called from inside a runtime (a `#[tokio::test]` that reads back via `Reader` does exactly that) — `block_on` on the caller's thread would panic "runtime within a runtime". Empirically validated both the spawn_blocking and inside-#[tokio::test] call sites. Invariants: no on-disk format change (§3.5 untouched — same Parquet bytes). The querier hot path is unaffected (it reads via DataFusion's own object_store integration, not `ourios_parquet::Reader`); the only sync consumers are compaction and tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 42 minutes and 16 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughAdds a Tokio runtime bridge to ChangesSync→async bridge and Reader integration
Sequence Diagram(s)sequenceDiagram
participant Reader as Reader::open_file
participant read_object as read_object
participant Store as Store::local
participant get_blocking as Store::get_blocking
participant block_on as block_on_off_runtime
participant from_bytes as Reader::from_bytes
Reader->>read_object: read_object(path)
read_object->>Store: Store::local(parent_dir)
Store-->>read_object: Store
read_object->>get_blocking: get_blocking(file_name_key)
get_blocking->>block_on: block_on_off_runtime(future)
block_on-->>get_blocking: Vec<u8>
get_blocking-->>read_object: Vec<u8>
read_object-->>Reader: Vec<u8>
Reader->>from_bytes: from_bytes(bytes, path)
from_bytes-->>Reader: Reader
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
Wires ourios_parquet::Reader’s synchronous file-open path through the RFC 0013 Store abstraction by reading file bytes via object_store (using a sync→async bridge), so the local filesystem and future S3 backends share the same read seam.
Changes:
- Added a process-wide Tokio runtime bridge plus
Store::{get_blocking, put_blocking}to support synchronous storage call sites. - Refactored
Readerto open files by fetching bytes throughStore::local(...).get_blocking(...), and introduced a sharedfrom_byteshelper used by bothopen_fileandopen_bytes. - Promoted
tokioto a normal dependency ofourios-parquetto support the blocking bridge.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| crates/ourios-parquet/src/store.rs | Adds sync blocking bridge over async object_store operations and tests for runtime-safety. |
| crates/ourios-parquet/src/reader.rs | Routes open_file through the Store seam by reading bytes via the blocking bridge; refactors reader construction. |
| crates/ourios-parquet/Cargo.toml | Adds Tokio as a direct dependency for the sync→async bridge runtime. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…go doc -D warnings)
Address review on the object-store read seam: - Fix a flaky "cannot drop a runtime in async context" panic: the shared OnceLock<Runtime> bridge dropped the loser's runtime during an init race on the caller's thread, which can be inside a #[tokio::test]. Replace it with a per-call current_thread runtime built, driven, and dropped entirely on the scoped OS thread — no shared state, no init race, no illegal drop. Also drops the unnecessary enable_all() (local I/O uses spawn_blocking only; the S3 backend adds it plus the net/io/time features in its own slice). - Flatten block_on_off_runtime to return Result<T, StoreError> directly instead of a nested Result unwrapped with a trailing `?` (clearer at the call sites). - Move the §3.9 baseline-column rustdoc back onto require_baseline_columns; it had been stranded above read_object after the refactor. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/ourios-parquet/src/store.rs`:
- Around line 49-53: The s.spawn() call in the scoped thread block will panic on
OS thread creation failure instead of returning an error. Replace the current
s.spawn() pattern with std::thread::Builder::spawn_scoped() which returns a
Result<JoinHandle> that can be mapped to handle io::Error failures. Map any
io::Error from spawn_scoped to StoreError::Runtime just like the runtime build
error on the next line. Additionally, update the documentation for the
StoreError::Runtime variant to indicate it covers both thread creation failures
and runtime build failures, since the public API methods (get_blocking,
put_blocking) document this variant as the failure surface for the sync-to-async
bridge.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 574b1e88-25c3-4006-aa23-1260211183bf
📒 Files selected for processing (3)
crates/ourios-parquet/Cargo.tomlcrates/ourios-parquet/src/reader.rscrates/ourios-parquet/src/store.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- crates/ourios-parquet/src/reader.rs
- Use std::thread::Builder::spawn_scoped (not Scope::spawn, which .expect()s) in the sync→async bridge so OS thread-creation failure surfaces as StoreError::Runtime instead of panicking; name the thread for observability and widen the variant doc to cover thread + runtime build failures. - store_io_err wraps StoreError as the io::Error source via io::Error::other instead of stringifying it, keeping the backend cause inspectable through the error chain (and dropping an allocation). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ead diagnostics (copilot)
RFC 0013 green slice 3 — wires the sync
Readerread path onto theStoreseam.What
Reader::open_file/open_partitionread bytes throughStore::local(parent).get_blocking(name)→open_bytes, routing the local read path through the RFC 0013 abstraction (the same one the S3 backend will use). Extracted a sharedfrom_byteshelper;open_bytesreuses it.Storegainsget_blocking/put_blocking— the sync→async bridge the sync storage layer needs (object_storeis async; compaction must reach S3 per RFC0013.3, so a local-onlystd::fsshortcut won't do).tokiopromoted to a normal dep (rt,rt-multi-thread) — already in the tree via DataFusion / object_store, noCargo.lockchurn.Bridge design (the non-obvious bit)
The sync storage code (
Writer/Reader/compaction) must drive asyncobject_storeops.block_onpanics "runtime within a runtime" if called on a thread that's inside a tokio runtime — and one consumer (ourios-querier'srfc0001_time_preservedtest) opens the reader from a#[tokio::test]. So the bridge runsblock_onon a fresh OS thread viastd::thread::scope: a plain thread never inherits the caller's tokio context, making it safe from any call site (sync test,spawn_blockingcompaction sweep, or inside a runtime).scopelets the future borrow the caller'sself/key. Both the spawn_blocking and inside-#[tokio::test]paths are covered by new tests.RFC0013.2 (local regresses nothing — read half)
Every existing RFC 0005 / RFC 0009 suite passes unchanged now that reads flow through the seam (full workspace
cargo testgreen). No on-disk format change (§3.5 untouched — identical Parquet bytes).Scope notes
ListingTable/object_store integration, notourios_parquet::Reader. The only syncReaderconsumers are compaction and tests.Store.put) and the manifest conditional-PUT (RFC0013.3/.4) land in following slices; the §5 stubs stay#[ignore]d until then.🤖 Generated with Claude Code
Summary by CodeRabbit