From 6d40e061d311ccdf820ec2bdebbf1e3a762a2390 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Tue, 16 Jun 2026 02:00:09 +0200 Subject: [PATCH 1/2] feat(parquet): route the manifest through the object-store seam (rfc0013 green) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the RFC 0009 per-partition manifest read/write onto the RFC 0013 `Store` seam (local backend), behaviour-preserving: - `Manifest::read` reads via `Store::local(partition_dir).get_blocking_opt`, guarded by a `try_exists` check so an absent partition directory still maps to `Ok(None)` (the prior `std::fs` NotFound→None contract; `Store::local` canonicalises its root and would otherwise error on a missing dir). - `Manifest::write_atomic` writes via `Store::put_blocking`. object_store's local put stages to a private temp object and renames it into place, so the atomic-commit-point and no-fsync semantics are unchanged from the prior tmp+rename — last-writer-wins `Overwrite`. New `Store` primitives: - `get_blocking_opt` (NotFound→None) + `StoreError::is_not_found`. - `put_if_absent` / `put_if_absent_blocking` (create-if-absent, If-None-Match) — the local-testable half of RFC 0013 conditional PUT. The generation compare-and-swap (RFC0013.3/.4) is deferred to the S3 backend: `LocalFileSystem` rejects `PutMode::Update`, so CAS can't be implemented or tested on the local backend. Every existing RFC 0009 manifest + compaction suite passes unchanged through the seam. Co-Authored-By: Claude Opus 4.8 --- crates/ourios-parquet/src/manifest.rs | 72 ++++++++++++-------- crates/ourios-parquet/src/store.rs | 98 ++++++++++++++++++++++++++- 2 files changed, 141 insertions(+), 29 deletions(-) diff --git a/crates/ourios-parquet/src/manifest.rs b/crates/ourios-parquet/src/manifest.rs index ea1deb4d9..26c6797c3 100644 --- a/crates/ourios-parquet/src/manifest.rs +++ b/crates/ourios-parquet/src/manifest.rs @@ -17,6 +17,14 @@ use std::path::Path; use serde::{Deserialize, Serialize}; +use crate::store::{Store, StoreError}; + +/// Map a [`StoreError`] from the manifest seam onto [`ManifestError::Io`], +/// keeping the backend cause in the error chain. +fn store_io(err: StoreError) -> ManifestError { + ManifestError::Io(std::io::Error::other(err)) +} + /// Canonical manifest filename inside a partition directory. pub const MANIFEST_FILENAME: &str = "manifest.json"; @@ -103,7 +111,8 @@ fn is_partition_local_parquet(name: &str) -> bool { } impl Manifest { - /// Read `/manifest.json`. + /// Read `/manifest.json` through the object-storage + /// [`Store`] seam (RFC 0013). /// /// `Ok(None)` when the manifest is absent — the pre-compaction /// (and current) case, where the reader falls back to globbing @@ -114,15 +123,24 @@ impl Manifest { /// [`ManifestError`] if the file exists but can't be read, or its /// bytes aren't valid manifest JSON. pub fn read(partition_dir: &Path) -> Result, ManifestError> { - match std::fs::read(partition_dir.join(MANIFEST_FILENAME)) { - Ok(bytes) => { + // A missing partition directory means no manifest (matches the prior + // `std::fs` NotFound→None). Checked up front because `Store::local` + // canonicalises its root and would error on an absent directory. + if !partition_dir.try_exists().map_err(ManifestError::Io)? { + return Ok(None); + } + let store = Store::local(partition_dir).map_err(store_io)?; + match store + .get_blocking_opt(MANIFEST_FILENAME) + .map_err(store_io)? + { + Some(bytes) => { let manifest: Self = serde_json::from_slice(&bytes).map_err(ManifestError::Parse)?; manifest.validate()?; Ok(Some(manifest)) } - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), - Err(e) => Err(ManifestError::Io(e)), + None => Ok(None), } } @@ -157,35 +175,35 @@ impl Manifest { serde_json::to_vec(self) } - /// Atomically (re)write the partition's manifest: serialize to a - /// sibling `manifest.json.tmp`, then `rename` it over - /// `manifest.json`. The rename is the **commit point** (RFC 0009 - /// §3.4) — a *concurrent reader* observes either the old manifest - /// or the new one, never a partial write. The manifest is - /// validated before any bytes hit disk, so an invalid set is never - /// published. + /// Atomically (re)write the partition's manifest through the + /// object-storage [`Store`] seam (RFC 0013). The `Store` put is the + /// **commit point** (RFC 0009 §3.4): the local backend stages to a + /// private temp object and renames it into place, so a *concurrent + /// reader* observes either the old manifest or the new one, never a + /// partial write. The manifest is validated before any bytes are + /// written, so an invalid set is never published. /// - /// This is atomic, **not** crash-durable: without an `fsync` of the - /// file and its directory, a host crash mid-write can still leave a - /// truncated or missing `manifest.json` (the same caveat as the - /// `Writer`). That is safe by construction — a reader with no - /// manifest falls back to the `*.parquet` glob, and a compaction - /// that crashed before this commit left its inputs intact — so the - /// worst case is reverting to the prior generation, never data - /// loss. Durable fsync is a later refinement. + /// On the local backend this is last-writer-wins `Overwrite`, atomic but + /// **not** crash-durable (no `fsync`) — the same contract as before the + /// seam, and the same caveat as the `Writer`. That is safe by + /// construction: a reader with no manifest falls back to the `*.parquet` + /// glob, and a compaction that crashed before this commit left its inputs + /// intact, so the worst case is reverting to the prior generation, never + /// data loss. Compare-and-swap (generation CAS, `If-Match`) lands with the + /// S3 backend — `LocalFileSystem` does not support conditional update. /// /// # Errors /// /// [`ManifestError::InvalidFilename`] if any entry isn't a - /// partition-local `*.parquet` name; [`ManifestError::Io`] on a - /// write or rename failure. + /// partition-local `*.parquet` name; [`ManifestError::Io`] on a write + /// failure. pub fn write_atomic(&self, partition_dir: &Path) -> Result<(), ManifestError> { self.validate()?; - let bytes = serde_json::to_vec(self).map_err(ManifestError::Parse)?; - let tmp = partition_dir.join(format!("{MANIFEST_FILENAME}.tmp")); - let final_path = partition_dir.join(MANIFEST_FILENAME); - std::fs::write(&tmp, &bytes).map_err(ManifestError::Io)?; - std::fs::rename(&tmp, &final_path).map_err(ManifestError::Io)?; + let bytes = self.to_json().map_err(ManifestError::Parse)?; + let store = Store::local(partition_dir).map_err(store_io)?; + store + .put_blocking(MANIFEST_FILENAME, bytes) + .map_err(store_io)?; Ok(()) } } diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index 66c5860cb..e0aa127f8 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -20,7 +20,7 @@ use std::sync::Arc; use object_store::local::LocalFileSystem; use object_store::path::Path as ObjectPath; -use object_store::{ObjectStore, ObjectStoreExt, PutPayload}; +use object_store::{ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload}; /// Drive `fut` to completion synchronously — the bridge from the **sync** /// storage API (`Writer`, `Reader`, `compaction`, the manifest) to async @@ -137,6 +137,15 @@ impl std::error::Error for StoreError { } } +impl StoreError { + /// True if this is a "no such object" backend error — the caller may + /// treat the object as absent (see [`Store::get_blocking_opt`]). + #[must_use] + pub fn is_not_found(&self) -> bool { + matches!(self, Self::Backend(object_store::Error::NotFound { .. })) + } +} + impl Store { /// Local-filesystem backend rooted at `root` (dev / test / CI). Preserves /// today's on-disk layout — the RFC 0005 Hive keys become paths under @@ -258,11 +267,54 @@ impl Store { pub fn put_blocking(&self, key: &str, bytes: Vec) -> Result<(), StoreError> { block_on_off_runtime(self.put(key, bytes)) } + + /// Write `bytes` to `key` only if no object exists there + /// (create-if-absent — `If-None-Match: *`). The local-testable half of + /// RFC 0013 conditional PUT; the compare-and-swap half (`If-Match`) needs + /// an S3 backend, since `LocalFileSystem` rejects `PutMode::Update`. + /// + /// # Errors + /// [`StoreError::Backend`] if an object already exists at `key`, or the + /// put otherwise fails. + pub async fn put_if_absent(&self, key: &str, bytes: Vec) -> Result<(), StoreError> { + self.inner + .put_opts( + &self.resolve(key), + PutPayload::from(bytes), + PutOptions::from(PutMode::Create), + ) + .await + .map_err(StoreError::Backend)?; + Ok(()) + } + + /// Read the object at `key`, mapping a missing object to `None` rather + /// than an error — for sync call sites where absence is expected (e.g. a + /// partition with no manifest yet). + /// + /// # Errors + /// As [`Self::get_blocking`], except a not-found object yields `Ok(None)`. + pub fn get_blocking_opt(&self, key: &str) -> Result>, StoreError> { + match self.get_blocking(key) { + Ok(bytes) => Ok(Some(bytes)), + Err(e) if e.is_not_found() => Ok(None), + Err(e) => Err(e), + } + } + + /// Blocking [`Self::put_if_absent`] for the sync storage call sites. + /// + /// # Errors + /// As [`Self::put_if_absent`], plus [`StoreError::Runtime`] if the bridge + /// runtime can't be built. + pub fn put_if_absent_blocking(&self, key: &str, bytes: Vec) -> Result<(), StoreError> { + block_on_off_runtime(self.put_if_absent(key, bytes)) + } } #[cfg(test)] mod tests { - use super::Store; + use super::{Store, StoreError}; /// A byte object round-trips through the local backend, and a delete /// removes it. (Foundation for the RFC0013 consumer migration; the §5 @@ -312,4 +364,46 @@ mod tests { b"inside-runtime" ); } + + /// `get_blocking_opt` maps a missing object to `None` (the manifest's + /// "no manifest yet" case) and yields the bytes when present. + #[test] + fn get_blocking_opt_maps_missing_to_none() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + assert_eq!( + store.get_blocking_opt("manifest.json").expect("get_opt"), + None, + "absent object is None, not an error" + ); + store + .put_blocking("manifest.json", b"{}".to_vec()) + .expect("put"); + assert_eq!( + store.get_blocking_opt("manifest.json").expect("get_opt"), + Some(b"{}".to_vec()), + ); + } + + /// `put_if_absent` (create-if-absent) writes when the key is free and + /// refuses to clobber an existing object — the local-testable half of + /// RFC 0013 conditional PUT. + #[test] + fn put_if_absent_refuses_to_clobber() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + let key = "manifest.json"; + store + .put_if_absent_blocking(key, b"first".to_vec()) + .expect("first create"); + let err = store + .put_if_absent_blocking(key, b"second".to_vec()) + .expect_err("create over an existing object must fail"); + assert!(matches!(err, StoreError::Backend(_)), "got {err:?}"); + assert_eq!( + store.get_blocking(key).expect("get"), + b"first", + "the original object is untouched" + ); + } } From 1d0392be1ce3ed5875d9d83f9f1eb81b7e881b19 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Tue, 16 Jun 2026 02:17:14 +0200 Subject: [PATCH 2/2] perf(parquet): share one bridge runtime instead of building per call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sync→async bridge built a fresh tokio runtime on every *_blocking call. Manifest::read goes through it once per partition, and the querier's resolve_live_files (on its async task) walks every partition per query — so a multi-partition query paid a runtime build per partition on the hot path. Build the runtime once via OnceLock::get_or_init (multi-thread, 1 worker) and reuse it. get_or_init can't drop a surplus runtime on a caller thread the way the earlier manual get/set could (that flaked with "drop a runtime in async context" when the loser was inside a #[tokio::test]); the runtime lives for the process and is never dropped. The build Result is cached so failure still surfaces as StoreError::Runtime without an expect. The per-call scoped thread stays — it's required so block_on runs off the caller's tokio context — but it no longer also builds a runtime. rt-multi-thread is back (a shared runtime is driven by concurrent block_on from many bridge threads; a current-thread runtime can't). Co-Authored-By: Claude Opus 4.8 --- crates/ourios-parquet/Cargo.toml | 14 ++++--- crates/ourios-parquet/src/store.rs | 63 +++++++++++++++++++++--------- 2 files changed, 52 insertions(+), 25 deletions(-) diff --git a/crates/ourios-parquet/Cargo.toml b/crates/ourios-parquet/Cargo.toml index a5b614d97..9c5b569de 100644 --- a/crates/ourios-parquet/Cargo.toml +++ b/crates/ourios-parquet/Cargo.toml @@ -32,12 +32,14 @@ object_store = { version = "0.13" } # arrow/parquet tree; declared directly. bytes = { version = "1", default-features = false } # Drives the async `object_store` calls from the sync storage API -# (`Store::*_blocking` bridge — `Writer`/`Reader`/compaction are sync, -# `object_store` is async). Only `rt`: the bridge builds a `current_thread` -# runtime per call on its own thread. The S3 backend adds `enable_all()` and the -# `net`/`io`/`time` features in its slice. Already in the workspace tree via -# DataFusion / object_store. -tokio = { version = "1", default-features = false, features = ["rt"] } +# (`Store::*_blocking` bridge — `Writer`/`Reader`/compaction/manifest are sync, +# `object_store` is async). `rt-multi-thread` because the bridge shares one +# process-wide runtime and drives it with `block_on` from many threads +# concurrently (parallel queries resolving manifests); a multi-thread runtime +# supports that, a current-thread one does not. The S3 backend adds +# `enable_all()` and the `net`/`io`/`time` features in its slice. Already in the +# workspace tree via DataFusion / object_store. +tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread"] } # UUIDv7 file naming per RFC 0005 §3.4 — timestamp-in-high-bits # means lexicographic sort = creation-order. uuid = { version = "1", default-features = false, features = ["v7", "std"] } diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index e0aa127f8..0f0307a68 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -16,49 +16,74 @@ //! depend on this crate, so the type is visible to every storage consumer. use std::future::Future; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use object_store::local::LocalFileSystem; use object_store::path::Path as ObjectPath; use object_store::{ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload}; +use tokio::runtime::Runtime; + +/// The process-wide runtime that drives the async `object_store` calls behind +/// the sync storage API. Built once, lazily, via `get_or_init` so there is no +/// init-race that could drop a surplus runtime on a caller's thread (an +/// earlier manual `get`/`set` did, panicking when the loser was inside a tokio +/// runtime). The runtime lives for the process and is never dropped, so the +/// "drop a runtime in async context" hazard can't arise. +/// +/// Multi-threaded(1-worker) so concurrent `block_on` from many bridge threads +/// (parallel queries / tests) is safe. No `enable_all()`: the local backend +/// drives I/O via `spawn_blocking`, which needs only the bare runtime; the S3 +/// backend adds `enable_all()` plus the `net`/`io`/`time` tokio features in the +/// slice that introduces it. +fn bridge_runtime() -> Result<&'static Runtime, StoreError> { + static RT: OnceLock> = OnceLock::new(); + match RT.get_or_init(|| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .build() + }) { + Ok(rt) => Ok(rt), + // Build failure is cached (a permanent resource exhaustion); rebuild a + // fresh `io::Error` since it isn't `Clone`. + Err(e) => Err(StoreError::Runtime(std::io::Error::new( + e.kind(), + e.to_string(), + ))), + } +} /// Drive `fut` to completion synchronously — the bridge from the **sync** /// storage API (`Writer`, `Reader`, `compaction`, the manifest) to async /// `object_store` (compaction must reach S3 per RFC0013.3, so a local-only /// `std::fs` shortcut won't do). /// -/// Everything happens on a **fresh OS thread**: a single-threaded runtime is -/// built, drives `fut`, and is dropped, all on a thread that never carries the -/// caller's tokio context. That makes the bridge safe from any call site — -/// including *inside* a runtime (a `#[tokio::test]` that opens the reader, or -/// a future async consumer), where calling `block_on` (or dropping a runtime) -/// on the caller's own thread would panic. [`std::thread::scope`] lets `fut` -/// borrow the caller's `self`/`key` while still running off-thread. +/// `block_on` runs on a **fresh OS thread** (the shared [`bridge_runtime`] is +/// driven from there), not the caller's. A plain thread never carries the +/// caller's tokio context, so this is safe from any call site — including +/// *inside* a runtime (e.g. the querier resolving manifests on its async task, +/// or a `#[tokio::test]`), where `block_on` on the caller's own thread would +/// panic. [`std::thread::scope`] lets `fut` borrow the caller's `self`/`key` +/// while still running off-thread. Reusing the shared runtime keeps the +/// per-call cost to one thread spawn (no per-call runtime build), which matters +/// on the query path (`resolve_live_files` reads one manifest per partition). /// /// `fut` already yields a [`StoreError`] result, returned directly; the extra -/// error modes are building the bridge thread or its runtime +/// error modes are building the bridge thread or runtime /// ([`StoreError::Runtime`]). A panic *inside* `fut` is not swallowed — it is -/// re-raised on the caller's thread via [`std::panic::resume_unwind`]. No -/// `enable_all()`: the local backend drives I/O via `spawn_blocking`, which -/// needs only the bare runtime; the S3 backend adds `enable_all()` and the -/// `net`/`io`/`time` tokio features in the slice that introduces it. +/// re-raised on the caller's thread via [`std::panic::resume_unwind`]. fn block_on_off_runtime( fut: impl Future> + Send, ) -> Result where T: Send, { + let rt = bridge_runtime()?; std::thread::scope(|s| { // `Builder::spawn_scoped` (not `Scope::spawn`) so OS thread-creation // failure surfaces as `StoreError::Runtime` rather than panicking. let handle = std::thread::Builder::new() .name("ourios-store-bridge".into()) - .spawn_scoped(s, || { - let rt = tokio::runtime::Builder::new_current_thread() - .build() - .map_err(StoreError::Runtime)?; - rt.block_on(fut) - }) + .spawn_scoped(s, || rt.block_on(fut)) .map_err(StoreError::Runtime)?; handle .join()