diff --git a/crates/ourios-parquet/src/store.rs b/crates/ourios-parquet/src/store.rs index 5f7a92b4..09ea1500 100644 --- a/crates/ourios-parquet/src/store.rs +++ b/crates/ourios-parquet/src/store.rs @@ -408,6 +408,25 @@ impl Store { .map_err(StoreError::Backend) } + /// Blocking [`Self::delete`] for the **sync** storage call sites (the + /// compactor's orphan GC and post-commit input reclaim). Safe to call from + /// inside a tokio runtime (see [`Self::get_blocking`]). + /// + /// **Missing-key behaviour is backend-dependent** (this bridge adds no + /// existence check): `LocalFileSystem` maps an absent key to a + /// [`is_not_found`](StoreError::is_not_found) error, while S3 DELETE is + /// idempotent and returns success. The compactor's GC loops treat *both* as + /// "already reclaimed" — they match `is_not_found` and otherwise count a + /// failure — so the difference is invisible to them; do not rely on a + /// uniform not-found for an absent key. + /// + /// # Errors + /// [`StoreError::Runtime`] if the bridge runtime can't be built; + /// otherwise as [`Self::delete`] (and see the missing-key note above). + pub fn delete_blocking(&self, key: &str) -> Result<(), StoreError> { + block_on_off_runtime(self.delete(key)) + } + /// 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. @@ -431,6 +450,22 @@ impl Store { /// not inherited from the backend (neither `LocalFileSystem` nor S3 /// guarantees stream order), so the contract is deterministic. async fn list(&self, prefix: Option<&str>) -> Result, StoreError> { + Ok(self + .list_entries(prefix) + .await? + .into_iter() + .map(|(key, _size)| key) + .collect()) + } + + /// List every object under `prefix` (store-relative) as `(key, size)` pairs, + /// recursively, in **lexicographic key order** — the size-bearing core of + /// [`Self::list`]. The compactor's small-file candidate check needs each + /// object's byte length, which the backend already reports in the listing + /// (`ObjectMeta::size`), so it comes for free here rather than via a + /// per-object `head`. Same tenant-isolation gating and key normalisation as + /// [`Self::list`]. + async fn list_entries(&self, prefix: Option<&str>) -> Result, StoreError> { let scoped = prefix.map_or_else(|| self.prefix.clone(), |p| self.resolve(p)); let metas: Vec = self .inner @@ -439,7 +474,7 @@ impl Store { .await .map_err(StoreError::Backend)?; let root = &self.prefix; - let mut keys: Vec = metas + let mut entries: Vec<(String, u64)> = metas .into_iter() .filter_map(|m| { // The backend's `list` does **string**-prefix matching, so S3 @@ -455,16 +490,15 @@ impl Store { // `#[must_use]` value used. let _ = m.location.prefix_match(&scoped)?; let parts = m.location.prefix_match(root)?; - Some( - parts - .map(|p| p.as_ref().to_owned()) - .collect::>() - .join("/"), - ) + let key = parts + .map(|p| p.as_ref().to_owned()) + .collect::>() + .join("/"); + Some((key, m.size)) }) .collect(); - keys.sort(); - Ok(keys) + entries.sort_by(|a, b| a.0.cmp(&b.0)); + Ok(entries) } /// Blocking recursive key listing for the **sync** storage call sites — the @@ -478,6 +512,21 @@ impl Store { block_on_off_runtime(self.list(prefix)) } + /// Blocking `(key, size)` listing for the **sync** storage call sites — the + /// bridge over the internal async `list_entries`, used by the compactor to + /// size small-file candidates without a per-object `head`. Same order + + /// isolation contract as [`Self::list_blocking`]. + /// + /// # Errors + /// [`StoreError::Runtime`] if the bridge runtime can't be built; + /// [`StoreError::Backend`] on a listing failure. + pub fn list_with_sizes_blocking( + &self, + prefix: Option<&str>, + ) -> Result, StoreError> { + block_on_off_runtime(self.list_entries(prefix)) + } + /// Blocking [`Self::put`] for the **sync** storage call sites (`Writer`, /// compaction). Safe to call from inside a tokio runtime (see /// [`Self::get_blocking`]). @@ -717,6 +766,58 @@ mod tests { ); } + /// `list_with_sizes_blocking` reports each object's byte length alongside + /// the key, in the same lexicographic-by-key order and with the same + /// segment-wise tenant isolation as `list_blocking` — the compactor sizes + /// small-file candidates from this rather than a per-object `head`. + #[test] + fn list_with_sizes_reports_byte_lengths_in_key_order() { + let dir = tempfile::TempDir::new().expect("temp dir"); + let store = Store::local(dir.path()).expect("local store"); + // Distinct lengths so a size mismatch is visible; the `tenant_id=ab` + // sibling must be excluded when scoping to `tenant_id=a`. + store + .put_blocking("data/tenant_id=a/year=2026/h0.parquet", vec![0u8; 3]) + .expect("put"); + store + .put_blocking("data/tenant_id=a/year=2026/h1.parquet", vec![0u8; 7]) + .expect("put"); + store + .put_blocking("data/tenant_id=ab/year=2026/h0.parquet", vec![0u8; 11]) + .expect("put"); + assert_eq!( + store + .list_with_sizes_blocking(Some("data/tenant_id=a")) + .expect("list a"), + vec![ + ("data/tenant_id=a/year=2026/h0.parquet".to_string(), 3), + ("data/tenant_id=a/year=2026/h1.parquet".to_string(), 7), + ], + ); + } + + /// `delete_blocking` removes an object (the compactor's orphan/input GC). On + /// the **local** backend a missing key surfaces as a `is_not_found` error + /// (S3 DELETE is idempotent instead — see the method doc); the compactor's + /// GC treats either as already-reclaimed, the same way it tolerates + /// `ErrorKind::NotFound` on `std::fs::remove_file`. + #[test] + fn delete_blocking_removes_and_local_missing_is_not_found() { + 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"x".to_vec()).expect("put"); + store.delete_blocking(key).expect("delete"); + assert_eq!(store.get_blocking_opt(key).expect("get_opt"), None); + let err = store + .delete_blocking(key) + .expect_err("local backend: absent key is a not-found error"); + assert!( + err.is_not_found(), + "absent delete maps to not-found: {err:?}" + ); + } + /// 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 diff --git a/crates/ourios-parquet/tests/rfc0013_object_store.rs b/crates/ourios-parquet/tests/rfc0013_object_store.rs index 5eb22d08..534a26fa 100644 --- a/crates/ourios-parquet/tests/rfc0013_object_store.rs +++ b/crates/ourios-parquet/tests/rfc0013_object_store.rs @@ -456,3 +456,72 @@ async fn store_list_enumerates_keys_on_s3() { "no prefix lists the whole bucket (incl. the sibling), ordered", ); } + +/// `Store::list_with_sizes_blocking` reports each object's byte length on the +/// real `AmazonS3` backend — the compactor's small-file candidate check reads +/// sizes from the listing rather than a per-object `head` (RFC 0019 §3.3). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "RFC 0019 — S3 integration; run via the `s3-integration` CI job (needs Docker + AWS_* env)"] +async fn store_list_with_sizes_reports_byte_lengths_on_s3() { + let (_node, s3) = localstack_s3("ourios-it-sizes").await; + s3.put("data/tenant_id=a/year=2026/h0.parquet", vec![0u8; 3]) + .await + .expect("s3 put"); + s3.put("data/tenant_id=a/year=2026/h1.parquet", vec![0u8; 7]) + .await + .expect("s3 put"); + + let entries = { + let s3 = s3.clone(); + tokio::task::spawn_blocking(move || s3.list_with_sizes_blocking(Some("data/tenant_id=a"))) + .await + .expect("join") + .expect("list a") + }; + assert_eq!( + entries, + vec![ + ("data/tenant_id=a/year=2026/h0.parquet".to_string(), 3), + ("data/tenant_id=a/year=2026/h1.parquet".to_string(), 7), + ], + "S3 listing carries byte sizes, key-ordered", + ); +} + +/// `Store::delete_blocking` removes an object on the real `AmazonS3` backend +/// (the compactor's orphan/input GC). S3 DELETE is idempotent, so a redundant +/// delete of an absent key is tolerated (success or `is_not_found`) — the GC +/// loops treat either as already-reclaimed. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "RFC 0019 — S3 integration; run via the `s3-integration` CI job (needs Docker + AWS_* env)"] +async fn store_delete_blocking_removes_on_s3() { + let (_node, s3) = localstack_s3("ourios-it-delete").await; + let key = "data/tenant_id=t/year=2026/x.parquet"; + s3.put(key, b"x".to_vec()).await.expect("s3 put"); + + let s3b = s3.clone(); + tokio::task::spawn_blocking(move || s3b.delete_blocking(key)) + .await + .expect("join") + .expect("delete"); + assert!( + s3.get(key) + .await + .expect_err("gone after delete") + .is_not_found(), + "object is removed", + ); + + // S3 DELETE is idempotent: a redundant delete of an absent key returns + // success (unlike the local backend's not-found). Assert only what the + // compactor's GC relies on — the redundant delete is *tolerable*: either + // success or a not-found, never a hard error the GC can't classify. + let s3c = s3.clone(); + let redundant = tokio::task::spawn_blocking(move || s3c.delete_blocking(key)) + .await + .expect("join"); + assert!( + redundant.map_or_else(|e| e.is_not_found(), |()| true), + "redundant S3 delete is tolerated (success or not-found)", + ); +}