feat(parquet): route reader through the object-store seam (rfc0013 green)

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"] }

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<Self, ReaderError> {
-        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, ReaderError> {
+        Self::from_bytes(bytes, PathBuf::from("<object-store>"))
+    }
+
+    /// 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<Self, ReaderError> {
         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("<object-store>"),
+            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<bytes::Bytes, ReaderError> {
+    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`].

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<T>(
+    fut: impl Future<Output = Result<T, StoreError>> + Send,
+) -> Result<T, StoreError>
+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,13 +112,17 @@ 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 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         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<Vec<u8>, 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<u8>) -> 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"
+        );
+    }
 }