diff --git a/crates/ourios-parquet/Cargo.toml b/crates/ourios-parquet/Cargo.toml index bf2ab2d02..a5b614d97 100644 --- a/crates/ourios-parquet/Cargo.toml +++ b/crates/ourios-parquet/Cargo.toml @@ -31,6 +31,13 @@ object_store = { version = "0.13" } # the RFC 0013 buffer-and-put read path (`Reader::open_bytes`). Already in the # 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"] } # 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/reader.rs b/crates/ourios-parquet/src/reader.rs index 647b21dba..908d385a1 100644 --- a/crates/ourios-parquet/src/reader.rs +++ b/crates/ourios-parquet/src/reader.rs @@ -28,7 +28,6 @@ //! surface records rather than making up defaults. use std::fmt; -use std::fs::File; use std::io; use std::path::{Path, PathBuf}; @@ -45,6 +44,7 @@ use parquet::errors::ParquetError; use crate::columns; use crate::partition::{PartitionKey, TimestampOverflowError}; +use crate::store::{Store, StoreError}; /// Streaming Parquet reader for one data file. /// @@ -91,22 +91,8 @@ impl Reader { /// /// Same set as [`Self::open_partition`]. pub fn open_file(path: &Path) -> Result { - let file = File::open(path).map_err(|source| ReaderError::Io { - op: "open", - path: path.to_path_buf(), - source, - })?; - - let builder = - ParquetRecordBatchReaderBuilder::try_new(file).map_err(ReaderError::Parquet)?; - require_baseline_columns(builder.schema())?; - let inner = builder.build().map_err(ReaderError::Parquet)?; - - Ok(Self { - inner, - partition: None, - file_path: path.to_path_buf(), - }) + let bytes = read_object(path)?; + Self::from_bytes(bytes, path.to_path_buf()) } /// Open a reader over in-memory Parquet `bytes` — the RFC 0013 @@ -120,6 +106,14 @@ impl Reader { /// [`ReaderError::MissingRequiredColumn`] if a baseline REQUIRED column /// is absent (§3.9). pub fn open_bytes(bytes: bytes::Bytes) -> Result { + Self::from_bytes(bytes, PathBuf::from("")) + } + + /// Build a reader over in-memory Parquet `bytes`, recording `file_path` + /// for diagnostics (the real path on the [`Self::open_file`] / + /// [`Self::open_partition`] seam, a sentinel on [`Self::open_bytes`]). + /// Applies the §3.9 baseline-column check; leaves `partition` unset. + fn from_bytes(bytes: bytes::Bytes, file_path: PathBuf) -> Result { let builder = ParquetRecordBatchReaderBuilder::try_new(bytes).map_err(ReaderError::Parquet)?; require_baseline_columns(builder.schema())?; @@ -128,7 +122,7 @@ impl Reader { Ok(Self { inner, partition: None, - file_path: PathBuf::from(""), + file_path, }) } @@ -180,7 +174,8 @@ impl Reader { /// Errors produced by [`Reader`]. #[derive(Debug)] pub enum ReaderError { - /// Filesystem I/O failure (file open, footer read). + /// Object-store read I/O failure (store open, object fetch, or key + /// derivation). `op` names which step failed. Io { op: &'static str, path: PathBuf, @@ -231,7 +226,7 @@ impl fmt::Display for ReaderError { match self { Self::Io { op, path, source } => write!( f, - "filesystem I/O on `{op}` at {}: {source}", + "object-store read I/O on `{op}` at {}: {source}", path.display(), ), Self::Parquet(e) => write!(f, "parquet reader: {e}"), @@ -313,6 +308,46 @@ fn validate_row_vs_partition( Ok(()) } +/// Read a data file's bytes through the object-storage [`Store`] seam +/// (RFC 0013): a `LocalFileSystem`-backed store rooted at the file's parent +/// directory, keyed by the file name. The sync read path (compaction, tests) +/// thus goes through the same seam the S3 backend will, while keeping the +/// `&Path` API — `Store.get` is async, so this uses the blocking bridge. +fn read_object(path: &Path) -> Result { + let parent = path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let name = path + .file_name() + .and_then(|n| n.to_str()) + .ok_or_else(|| ReaderError::Io { + op: "derive object key", + path: path.to_path_buf(), + source: io::Error::new( + io::ErrorKind::InvalidInput, + "file name is empty or not valid UTF-8", + ), + })?; + let store = Store::local(parent).map_err(|e| store_io_err("open store", path, e))?; + let bytes = store + .get_blocking(name) + .map_err(|e| store_io_err("fetch object", path, e))?; + Ok(bytes::Bytes::from(bytes)) +} + +/// Map a [`StoreError`] from the read seam onto [`ReaderError::Io`], preserving +/// the file path and op for diagnostics. The `StoreError` is wrapped as the +/// `io::Error`'s source (not stringified), so the backend cause stays +/// inspectable through the error chain. +fn store_io_err(op: &'static str, path: &Path, err: StoreError) -> ReaderError { + ReaderError::Io { + op, + path: path.to_path_buf(), + source: io::Error::other(err), + } +} + /// RFC 0005 §3.9: every baseline REQUIRED (non-nullable) column must be /// present in the file's schema, else a hard error. Shared by /// [`Reader::open_file`] and [`Reader::open_bytes`]. diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index ce8395301..66c5860cb 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -15,12 +15,57 @@ //! (not a new crate): `ourios-querier`, `-ingester`, and `-server` already //! depend on this crate, so the type is visible to every storage consumer. +use std::future::Future; use std::sync::Arc; use object_store::local::LocalFileSystem; use object_store::path::Path as ObjectPath; use object_store::{ObjectStore, ObjectStoreExt, PutPayload}; +/// 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. +/// +/// `fut` already yields a [`StoreError`] result, returned directly; the extra +/// error modes are building the bridge thread or its 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. +fn block_on_off_runtime( + fut: impl Future> + Send, +) -> Result +where + T: Send, +{ + 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) + }) + .map_err(StoreError::Runtime)?; + handle + .join() + .unwrap_or_else(|payload| std::panic::resume_unwind(payload)) + }) +} + /// A handle to the object store backing a tenant store's Parquet + manifest /// objects, addressed by key under `prefix`. Wraps an [`ObjectStore`] so the /// same code path targets `LocalFileSystem` or `AmazonS3` / S3-compatible. @@ -67,6 +112,9 @@ pub enum StoreError { /// Returned (rather than panicking) so an accidental call fails /// gracefully while `red`. Unimplemented(&'static str), + /// The sync→async bridge thread or runtime could not be built (resource + /// exhaustion). Surfaced by the `*_blocking` methods. + Runtime(std::io::Error), } impl std::fmt::Display for StoreError { @@ -74,6 +122,7 @@ impl std::fmt::Display for StoreError { match self { Self::Backend(e) => write!(f, "object-store backend: {e}"), Self::Unimplemented(what) => write!(f, "not implemented (RFC 0013 red): {what}"), + Self::Runtime(e) => write!(f, "object-store bridge runtime: {e}"), } } } @@ -83,6 +132,7 @@ impl std::error::Error for StoreError { match self { Self::Backend(e) => Some(e), Self::Unimplemented(_) => None, + Self::Runtime(e) => Some(e), } } } @@ -186,6 +236,28 @@ impl Store { .await .map_err(StoreError::Backend) } + + /// Blocking [`Self::get`] for the **sync** storage call sites (`Reader`, + /// compaction). Safe to call from any thread, including inside a tokio + /// runtime — the `block_on` runs off the caller's thread. + /// + /// # Errors + /// [`StoreError::Runtime`] if the bridge runtime can't be built; + /// otherwise as [`Self::get`]. + pub fn get_blocking(&self, key: &str) -> Result, StoreError> { + block_on_off_runtime(self.get(key)) + } + + /// Blocking [`Self::put`] for the **sync** storage call sites (`Writer`, + /// compaction). Safe to call from inside a tokio runtime (see + /// [`Self::get_blocking`]). + /// + /// # Errors + /// [`StoreError::Runtime`] if the bridge runtime can't be built; + /// otherwise as [`Self::put`]. + pub fn put_blocking(&self, key: &str, bytes: Vec) -> Result<(), StoreError> { + block_on_off_runtime(self.put(key, bytes)) + } } #[cfg(test)] @@ -205,4 +277,39 @@ mod tests { store.delete(key).await.expect("delete"); assert!(store.get(key).await.is_err(), "object gone after delete"); } + + /// The sync `*_blocking` bridge round-trips a byte object — the path the + /// sync `Writer` / `Reader` / compaction take onto `Store`. Runs on a + /// plain test thread (no ambient runtime), exercising `block_on`. + #[test] + fn blocking_bridge_put_get_round_trip() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + let key = "data/tenant_id=t/year=2026/x.parquet"; + store + .put_blocking(key, b"hello-blocking".to_vec()) + .expect("put_blocking"); + assert_eq!( + store.get_blocking(key).expect("get_blocking"), + b"hello-blocking" + ); + } + + /// The `*_blocking` bridge is safe to call from *within* a tokio runtime — + /// some consumers (e.g. a `#[tokio::test]` that reads back via `Reader`) + /// do exactly that. The `block_on` runs off the caller's thread, so it + /// must not panic "runtime within a runtime". + #[tokio::test(flavor = "current_thread")] + async fn blocking_bridge_is_safe_inside_a_runtime() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + let key = "data/tenant_id=t/year=2026/x.parquet"; + store + .put_blocking(key, b"inside-runtime".to_vec()) + .expect("put_blocking"); + assert_eq!( + store.get_blocking(key).expect("get_blocking"), + b"inside-runtime" + ); + } }