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..ce839530 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,70 @@ 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`.) + #[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 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"); + } }