From cd825fa778bb3e6d73506c38e50e91ad85cec36f Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Mon, 15 Jun 2026 19:23:18 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat(parquet):=20rfc0013=20green=20?= =?UTF-8?q?=E2=80=94=20store=20local=20I/O=20surface=20(put/get/delete)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 1 + crates/ourios-parquet/Cargo.toml | 3 ++ crates/ourios-parquet/src/store.rs | 73 +++++++++++++++++++++++++++++- 3 files changed, 76 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index f7c08583..5aab962c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2718,6 +2718,7 @@ dependencies = [ "serde", "serde_json", "tempfile", + "tokio", "uuid", ] diff --git a/crates/ourios-parquet/Cargo.toml b/crates/ourios-parquet/Cargo.toml index 1590ea97..58c5d1ca 100644 --- a/crates/ourios-parquet/Cargo.toml +++ b/crates/ourios-parquet/Cargo.toml @@ -57,6 +57,9 @@ tempfile = "3" # Property tests for the compaction row-conservation invariant # (RFC0009.2 / CLAUDE.md §6.2 — invariants get a property test). proptest = "1" +# Async runtime for the RFC 0013 `Store` round-trip test (the I/O surface +# is async via `object_store`). Already in the workspace tree. +tokio = { version = "1", default-features = false, features = ["rt", "macros"] } [lints] workspace = true diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index 708e78c1..51ffd94f 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -17,9 +17,9 @@ use std::sync::Arc; -use object_store::ObjectStore; use object_store::local::LocalFileSystem; use object_store::path::Path as ObjectPath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload}; /// A handle to the object store backing a tenant store's Parquet + manifest /// objects, addressed by key under `prefix`. Wraps an [`ObjectStore`] so the @@ -139,4 +139,75 @@ impl Store { pub fn prefix(&self) -> &ObjectPath { &self.prefix } + + /// Resolve a `/`-delimited `key` to an absolute object path under the + /// store prefix. At `red` the prefix is empty, so this is just the key; + /// once prefix scoping is wired (RFC0013.5) the prefix segments lead. + fn resolve(&self, key: &str) -> ObjectPath { + self.prefix + .parts() + .chain(ObjectPath::from(key).parts()) + .collect() + } + + /// Write `bytes` to `key`. + /// + /// # Errors + /// [`StoreError::Backend`] if the put fails. + pub async fn put(&self, key: &str, bytes: Vec) -> Result<(), StoreError> { + self.inner + .put(&self.resolve(key), PutPayload::from(bytes)) + .await + .map_err(StoreError::Backend)?; + Ok(()) + } + + /// Read the whole object at `key`. + /// + /// # Errors + /// [`StoreError::Backend`] if the object is missing or the read fails. + pub async fn get(&self, key: &str) -> Result, StoreError> { + let got = self + .inner + .get(&self.resolve(key)) + .await + .map_err(StoreError::Backend)?; + let bytes = got.bytes().await.map_err(StoreError::Backend)?; + Ok(bytes.to_vec()) + } + + /// Delete the object at `key`. + /// + /// # Errors + /// [`StoreError::Backend`] if the delete fails. + pub async fn delete(&self, key: &str) -> Result<(), StoreError> { + self.inner + .delete(&self.resolve(key)) + .await + .map_err(StoreError::Backend) + } +} + +#[cfg(test)] +mod tests { + use super::Store; + + /// A byte object round-trips through the local backend, and a delete + /// removes it. (Foundation for the RFC0013 consumer migration; the §5 + /// scenarios turn green as the writer/reader move onto `Store`.) + #[test] + fn local_store_put_get_delete_round_trip() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .expect("rt"); + rt.block_on(async { + let key = "data/tenant_id=t/year=2026/x.parquet"; + store.put(key, b"hello-ourios".to_vec()).await.expect("put"); + assert_eq!(store.get(key).await.expect("get"), b"hello-ourios"); + store.delete(key).await.expect("delete"); + assert!(store.get(key).await.is_err(), "object gone after delete"); + }); + } } From cf2d6ec9eb3be58025ed3e471c41af5f07692d33 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Mon, 15 Jun 2026 19:38:52 +0200 Subject: [PATCH 2/2] test(parquet): use #[tokio::test] for the store round-trip (copilot) --- crates/ourios-parquet/src/store.rs | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index 51ffd94f..ce839530 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -195,19 +195,14 @@ mod tests { /// A byte object round-trips through the local backend, and a delete /// removes it. (Foundation for the RFC0013 consumer migration; the §5 /// scenarios turn green as the writer/reader move onto `Store`.) - #[test] - fn local_store_put_get_delete_round_trip() { + #[tokio::test(flavor = "current_thread")] + async fn local_store_put_get_delete_round_trip() { let dir = tempfile::TempDir::new().expect("temp dir"); let store = Store::local(dir.path()).expect("local store"); - let rt = tokio::runtime::Builder::new_current_thread() - .build() - .expect("rt"); - rt.block_on(async { - let key = "data/tenant_id=t/year=2026/x.parquet"; - store.put(key, b"hello-ourios".to_vec()).await.expect("put"); - assert_eq!(store.get(key).await.expect("get"), b"hello-ourios"); - store.delete(key).await.expect("delete"); - assert!(store.get(key).await.is_err(), "object gone after delete"); - }); + let key = "data/tenant_id=t/year=2026/x.parquet"; + store.put(key, b"hello-ourios".to_vec()).await.expect("put"); + assert_eq!(store.get(key).await.expect("get"), b"hello-ourios"); + store.delete(key).await.expect("delete"); + assert!(store.get(key).await.is_err(), "object gone after delete"); } }