feat(ingester): migrate the compactor onto the Store seam (RFC 0019 slice 2b) - #297
Merged
Merged
Conversation
Move `compact_partition`, `gc_orphans`, `plan_candidates` and their helpers off raw `std::fs` onto `ourios_parquet::Store`, so data compaction targets a local filesystem or an S3 bucket through one code path (RFC 0019 §3.3). Every directory walk becomes a `Store` listing; inputs are read via `get_blocking` + `Reader::open_partition_bytes` (preserving the RFC0009.5 row-vs-path abort); the consolidated file is written via `Writer::open_in` and addressed by its object key + reported `bytes_written` (no local `stat`). The manifest swap is backend-aware (CLAUDE.md §3.5 / RFC0009.3 — no torn read either way): a conditional-PUT CAS (`publish_cas`) on S3 (RFC0019.4), and an atomic overwrite on local, since `LocalFileSystem` rejects `PutMode::Update` so `publish_cas` cannot commit there — RFC0019.7 keeps the local commit byte-for-byte unchanged. A new `Store::supports_conditional_update()` selects the path. The bootstrap (create-if-absent) is shared by both backends; a lost CAS race backs off as a no-op (the consolidated file is an orphan a later `gc_orphans` reclaims). Partition/tenant enumeration parses canonical zero-padded Hive segments from object keys, requiring a contiguous trailing run (mirrors the querier's `parse_day_partition` fix) so non-canonical layouts are skipped as before. Inline tests adapted to build a `Store::local` and pass `&store`; every assertion preserved. Adds an `#[ignore]`d localstack S3 compaction-CAS test (RFC0019.4). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…019) `Compactor` now holds a `Store` (built by the server from the resolved `StoreConfig`) instead of a `bucket_root` path; `run_sweep` and `tenants` take `&Store` and enumerate tenants from `data/tenant_id=…` object keys via `Store::list_blocking`. `Compactor::run` clones the cheap `Store` handle into each blocking sweep. `main()` drops the blanket s3 fail-fast: it opens the store from `config.store.open()` and runs the compactor on either backend, and the querier already takes the resolved `StoreConfig`. The receiver's RFC 0014 data write path and the `ParquetAuditSink` are still local-only (their Store migrations are later RFC 0019 slices), so the receiver fails fast on s3 and the compaction audit sink is wired only on the local backend (s3 logs the gap and runs without it). A local store root is pre-created before `Store::local` canonicalises it, mirroring the querier role. The compaction bench is updated to drive `compact_partition` through a `Store::local`. Inline compactor tests build a `Store::local`; all assertions preserved. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 7 minutes and 24 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughAll compaction APIs ( ChangesStore-based compaction
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR migrates the ingester compaction path off raw std::fs and onto the ourios_parquet::Store abstraction (RFC 0019 slice 2b), enabling the compactor (and server wiring) to operate on either a local filesystem store or an S3-compatible backend via a single code path.
Changes:
- Refactors Parquet compaction planning, reading, writing, manifest publish, and orphan GC to use
Storelisting/get/put/delete primitives. - Updates the ingester compactor daemon to hold a
Store, enumerate tenants via store keys, and run sweeps against either backend. - Updates the server
mainwiring to open the configured store for the compactor/querier, while still failing fast for the receiver on S3 and gating the local-only audit sink.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/ourios-server/src/main.rs | Opens the configured Store for compactor/querier; keeps receiver local-only and gates the local-only audit sink. |
| crates/ourios-parquet/src/store.rs | Adds a backend capability flag (supports_conditional_update) to gate CAS manifest commits. |
| crates/ourios-parquet/src/compaction.rs | Migrates compaction/GC/planning from filesystem walks to store listings and store-backed reads/writes, including CAS-vs-overwrite manifest commit logic. |
| crates/ourios-ingester/src/compactor.rs | Refactors compactor sweep orchestration to accept/hold a Store and list tenants from keys. |
| crates/ourios-parquet/tests/rfc0013_object_store.rs | Adds an ignored LocalStack S3 integration test for CAS-backed compaction commit behavior. |
| crates/ourios-bench/benches/compaction.rs | Updates benchmarks to call the new compact_partition(&Store, ...) API. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/ourios-parquet/src/compaction.rs (1)
357-366: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRestrict compaction filters to direct partition children.
Storelistings are recursive, butis_committed_parquetonly checks the suffix. A nested key like.../hour=10/sidecar/file.parquetwould be counted as live input, compacted, or deleted as an orphan even though the writer only owns direct<uuid>.parquetchildren.Proposed fix
- let name = basename(&object); + let Some(name) = partition_child_name(&object, &prefix) else { + continue; + }; // `.parquet.tmp` is always a dead interrupted publish. A `.parquet` is @@ - .filter(|(key, _)| is_committed_parquet(key)) + .filter(|(key, _)| is_committed_parquet_in_partition(key, &prefix)) @@ - .filter(|k| is_committed_parquet(k)) + .filter(|k| is_committed_parquet_in_partition(k, &prefix)) .collect()) } @@ -fn is_committed_parquet(key: &str) -> bool { - key.ends_with(".parquet") +fn partition_child_name<'a>(key: &'a str, prefix: &str) -> Option<&'a str> { + let name = key.strip_prefix(prefix)?.strip_prefix('/')?; + (!name.contains('/')).then_some(name) +} + +fn is_committed_parquet_in_partition(key: &str, prefix: &str) -> bool { + partition_child_name(key, prefix).is_some_and(|name| name.ends_with(".parquet")) }Also applies to: 456-467, 551-557, 594-598
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/ourios-parquet/src/compaction.rs` around lines 357 - 366, The compaction path is treating any recursive listing entry with a .parquet suffix as owned input, so nested files under partition subdirectories can be mistakenly compacted or deleted. Update the filtering logic in is_committed_parquet and the call sites in compaction.rs to only accept direct children of the writer’s partition/root scope (for example by validating the basename/prefix shape, not just the suffix), and apply the same restriction wherever keys are collected or pruned for live input, orphan detection, and deletion.
🧹 Nitpick comments (1)
crates/ourios-ingester/src/compactor.rs (1)
590-596: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep the malformed-manifest test on the Store seam.
This test now re-enters the local filesystem via
data_path(bucket.path()). Writing the corrupt manifest withstore.put_blockingkeeps the regression test aligned with the backend abstraction being introduced.♻️ Proposed test-only refactor
- let b_dir = PartitionKey::derive(&rec("b", 1, TS0)) - .expect("derive") - .data_path(bucket.path()); - std::fs::write(b_dir.join(ourios_parquet::MANIFEST_FILENAME), b"not json") - .expect("corrupt b's manifest"); + let b_manifest = format!( + "data/tenant_id=b/year=2026/month=04/day=02/hour=10/{}", + ourios_parquet::MANIFEST_FILENAME, + ); + store + .put_blocking(&b_manifest, b"not json".to_vec()) + .expect("corrupt b's manifest");🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/ourios-ingester/src/compactor.rs` around lines 590 - 596, The malformed-manifest regression test is bypassing the Store abstraction by writing directly through the local filesystem path from PartitionKey::derive(...).data_path(bucket.path()). Update the test in compactor.rs to corrupt the manifest via store.put_blocking instead of std::fs::write, so it stays on the Store seam. Keep the same manifest target identified by PartitionKey and MANIFEST_FILENAME, but route the write through the backend abstraction used elsewhere in the test.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/ourios-bench/benches/compaction.rs`:
- Around line 166-167: Move Store::local out of the timed section in the
compaction benchmarks so backend setup is excluded from the measured compaction
cost. In the benchmark cases around compact_partition, construct the Store
during setup once per temp dir (or before the iter/timed closure) and then only
invoke compact_partition(&store, &part) inside the measurement path; apply the
same change to each affected benchmark block so D2/D3/baseline keep isolating
the compaction hot path.
In `@crates/ourios-ingester/src/compactor.rs`:
- Around line 226-239: The tenant discovery logic in compactor.rs is doing a
full object listing via list_blocking(Some("data")) and then deriving tenants
from every key, which scales with all Parquet objects. Update the tenant
enumeration path in the compactor flow to use a tenant-level source instead,
such as a Store listing with delimiter/common-prefix support, a dedicated tenant
registry, or another index keyed by tenant_id, and keep the filtering/decoding
logic aligned with the existing percent_decode_tenant handling.
In `@crates/ourios-parquet/tests/rfc0013_object_store.rs`:
- Around line 533-538: The consolidation test currently only checks the row
count from Reader::open_partition_bytes/read_all, which can miss duplicate or
dropped records on the S3 path. Strengthen the assertion by comparing a stable
identifier or field set from the returned rows against the original records in
the test, using the existing rows and records values in this test block, so the
compacted object is verified as lossless rather than merely readable.
In `@crates/ourios-server/src/main.rs`:
- Around line 371-372: The compactor store is being opened too late in `main`,
after network roles may already be running, which can skip the graceful shutdown
path if `config.store.open()` fails. Preflight the store open earlier in the
startup flow before binding receiver/querier roles, then pass the opened store
handle into `Compactor::new` so the compactor only starts once the store is
guaranteed available.
---
Outside diff comments:
In `@crates/ourios-parquet/src/compaction.rs`:
- Around line 357-366: The compaction path is treating any recursive listing
entry with a .parquet suffix as owned input, so nested files under partition
subdirectories can be mistakenly compacted or deleted. Update the filtering
logic in is_committed_parquet and the call sites in compaction.rs to only accept
direct children of the writer’s partition/root scope (for example by validating
the basename/prefix shape, not just the suffix), and apply the same restriction
wherever keys are collected or pruned for live input, orphan detection, and
deletion.
---
Nitpick comments:
In `@crates/ourios-ingester/src/compactor.rs`:
- Around line 590-596: The malformed-manifest regression test is bypassing the
Store abstraction by writing directly through the local filesystem path from
PartitionKey::derive(...).data_path(bucket.path()). Update the test in
compactor.rs to corrupt the manifest via store.put_blocking instead of
std::fs::write, so it stays on the Store seam. Keep the same manifest target
identified by PartitionKey and MANIFEST_FILENAME, but route the write through
the backend abstraction used elsewhere in the test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 16b671b5-d141-40aa-86b7-d717ffdaaec0
📒 Files selected for processing (6)
crates/ourios-bench/benches/compaction.rscrates/ourios-ingester/src/compactor.rscrates/ourios-parquet/src/compaction.rscrates/ourios-parquet/src/store.rscrates/ourios-parquet/tests/rfc0013_object_store.rscrates/ourios-server/src/main.rs
Four review fixes on the RFC 0019 compactor-migration branch:
- bench: hoist `Store::local` into the untimed `iter_batched` setup so the D2
/ D3 paths time only `compact_partition`, not store construction.
- tenants: replace the recursive `list_blocking("data")` scan with a new
one-level `Store::list_common_prefixes_blocking` (a `list_with_delimiter`
roll-up over immediate `data/tenant_id=…` common-prefixes), restoring the
original `read_dir(data/)` semantics and avoiding an S3-scale full scan.
Adds a local unit test + an `#[ignore]`d localstack S3 test.
- S3 compaction test: assert row *identity* (value-equal, timestamp-sorted)
against the originals, not just the count, so a drop/dup can't pass.
- main: preflight `config.store.open()?` before binding the receiver/querier
roles, so a store-open failure can't bypass their graceful shutdown; the
opened handle is moved into `Compactor::new`.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Copilot: the Store list* calls are recursive, so the no-manifest glob fallback (live_file_keys) and the candidate size scan (is_candidate) could fold a nested <uuid>.parquet under the partition prefix in as a live input — a regression from the pre-RFC-0019 read_dir, which saw immediate children only. Add is_immediate_child and gate both scans on it. Also reword no_op_outcome's doc: it covers a lost CAS race too, where inputs were read and a consolidated orphan written but nothing was committed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Copilot: the sub-two-file case still does the listing + manifest read before deciding it's a no-op, so 'no I/O happens' was inaccurate; reword to 'no consolidation is performed'. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ive scan `hour_partitions` listed every object under `data/tenant_id=<enc>` and parsed partition tuples from keys — O(N_objects) per sweep on S3 and a regression from the pre-RFC-0019 level-by-level `read_dir` walk. Walk the Hive levels with a delimiter rollup instead: roll up `year=` child prefixes via `Store::list_common_prefixes_blocking`, then `month=`/`day=`/`hour=` under each, so every listing returns only the immediate common-prefixes (cheap). Each segment is parsed in the canonical zero-padded form, dropping a non-canonical intermediate prefix exactly as the old walk dropped non-canonical dirs; the result is identical (sorted chronologically, deduped). Removes the now-unused whole-key parser `parse_hour_partition_key`. Also reword the main.rs store-open preflight comment: `open` validates only local-root existence / config — `Store::s3` does not contact the endpoint, so credential/connectivity errors surface on first request, not at open. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This was referenced Jun 28, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Migrates the compactor off raw
std::fsontoourios_parquet::Store(RFC 0019 slice 2b core), so background compaction runs against a local filesystem or an S3-compatible bucket through one code path. Builds on the Store primitives (#293) and the Writer/Reader Store seam (#294).compaction.rs—plan_candidates, the partition/tenant walks,is_candidate,live_files,compact_partition,gc_orphansnow go entirely throughStore:Store::list_blocking/list_with_sizes_blocking(sizes for the small-file candidate check, no per-filestat);year=/month=/day=/hour=from keys, requiring a contiguous trailing canonical run (mirrors the querier'sparse_day_partitionfix) and skipping non-canonical keys exactly as the old dir-walk skipped non-canonical dirs;get_blocking→Reader::open_partition_bytes(preserves the RFC0009.5 row-vs-path abort on a mis-partitioned input);Writer::open_in, addressed byWrittenFile.key+bytes_written(no local path stat).compactor.rs—Compactorholds aStore(not aPathBuf);new(store, …),run_sweep/tenants(&Store); tenants enumerated fromdata/tenant_id=…keys.with_audit_sinkunchanged.main.rs— removes the blanket s3 fail-fast; opensconfig.store.open()and runs the compactor + querier on either backend; pre-creates the local root beforeStore::local(mirrors the querier role).store.rs— addsStore::supports_conditional_update()(additive backend-capability flag;truefor S3,falseforLocalFileSystem).Manifest commit — backend-aware (sanity-check this)
The spec said "commit via
publish_cas." In practiceobject_store 0.13.2'sLocalFileSystem::put_optsreturnsNotImplementedforPutMode::Update(If-Match), so a CAS commit cannot run on the local backend — which would break RFC0019.7 (local behavior unchanged) and the inline tests. Reconciliation (matching the existingmanifest.rsdocs):Manifest::publish_cas(compare-and-swap on the ETag, RFC0019.4); a lost race aborts the partition as a no-op, next sweep retries.put_blockingoverwrite — the same no-torn-read swapwrite_atomicperformed pre-RFC-0019 (last-writer-wins, RFC0019.7 unchanged).PutMode::Create) is shared — it works on both backends — preserving the RFC0009.3 "reader is manifest-authoritative before the consolidated file appears, no torn read" guarantee.Gated by the new
Store::supports_conditional_update().Invariants
Reader::open_partition_bytes.compaction_conserves_every_rowproptest passes.rfc0009_4_*) pass;delete_blockingtoleratesis_not_found(local) / idempotent delete (S3).Known follow-ups (flagged in code; not silent breakage)
These are separate seams, deliberately out of this slice (the compactor data path is the scope here):
main()fails fast if the receiver is enabled on s3 (clear error).ParquetAuditSinkis still local-only → on s3 the compactor runs with a no-op audit sink and logs the gap (rather than mis-placing audit Parquet locally).Both must land (plus the localstack e2e, slice 3) before RFC 0019 flips to
green.Tests
All existing compaction/compactor suites pass, adapted to construct via
Store::local(assertions unchanged). Added an#[ignore]d localstack S3 testcompact_partition_consolidates_on_s3_via_casfor the RFC0019.4 CAS commit path.Local gate (run firsthand on the branch)
cargo fmt --all --check,cargo clippy --workspace --all-targets --all-features -D warnings,cargo test -p ourios-parquet -p ourios-ingester -p ourios-server(0 failed; 14 ignored = localstack + deferred stub),RUSTDOCFLAGS=-D warnings cargo doc— all green.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Tests