feat(ingester): rfc0014 green pt1 — ParquetRecordSink + flush policy

[jensholdgaard]

RFC 0014 green (part 1) — the buffering production data write path. ParquetRecordSink replaces what NoOpRecordSink couldn't do: carry mined records to Parquet objects on the store.

What

ParquetRecordSink implements RecordSink, buffering mined records per partition (HashMap<PartitionKey, …>) and flushing each to a Parquet object via the RFC 0013 seam (encode_records_to_parquet + Store::put_blocking, UUIDv7-named per RFC 0005 §3.4). Hybrid flush policy, per the specified decisions:

• Size (RFC0014.1) — the emit crossing target_bytes flushes the partition.
• Age (RFC0014.2) — flush_aged (batch-window tick) flushes partitions whose oldest record reached max_buffer_age (inclusive).
• Rotation (RFC0014.3) — flush_all force-flushes every partition, including sub-threshold ones.
• Hard ceiling (RFC0014.4) — emit flushes the largest partition inline before it would exceed ceiling_bytes; buffered bytes never exceed it.
• Tenant isolation (RFC0014.6) — buffers keyed by PartitionKey (carries tenant_id), so a flush only ever writes one tenant's rows.

A flush failure retains the buffer (counted) rather than dropping — the WAL is the durability of record (CLAUDE.md §3.4). The sink takes a Store (local or S3), so it's S3-ready.

Tests

RFC0014.1/.2/.3/.4/.6 un-#[ignore]d — driven against a LocalFileSystem Store, reading the flushed objects back to prove no-loss + per-file single-tenant isolation. 5 pass, 1 ignored. Full workspace green; clippy -D warnings + fmt clean.

Deferred — green part 2

RFC0014.5 (crash: no acknowledged-data loss) stays #[ignore]d; part 2 wires the sink into the ingest pipeline (miner emit + WAL rotation hook, replacing NoOpRecordSink) and extends the RFC 0008 crash harness. RFC 0014 stays red until .5 is green. Then the follow-on server-Store-wiring slice greens RFC0013.6.

🤖 Generated with Claude Code

Summary by CodeRabbit

Release Notes

• New Features

  • Added record sinking capability with configurable flush policies: size-based, age-based, and manual triggers
  • Supports partitioned data buffering with memory-bounded operations
  • Includes tenant isolation support

• Tests

  • Enabled RFC 0014 ingest write-path acceptance tests verifying flushing behavior, data integrity, and multi-tenant isolation

[coderabbitai]

Warning

Review limit reached

@jensholdgaard, we couldn't start this review because you've reached your PR review rate limit.

More reviews will be available in 35 minutes and 42 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 @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

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 refill rate.

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, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly.

Please see our Fair Usage Limits Policy for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 24204e38-7d2e-45df-9de1-dd5437aadab1

📥 Commits

Reviewing files that changed from the base of the PR and between 2cf3002 and 52291f4.

📒 Files selected for processing (2)

• crates/ourios-ingester/src/record_sink.rs
• crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs

📝 Walkthrough

Walkthrough

Introduces ParquetRecordSink, a RecordSink implementation that buffers MinedRecords per PartitionKey and flushes Parquet files to a Store using a hybrid policy (size threshold, age timeout, forced rotation). A hard ceiling_bytes limit is enforced inline during emit. RFC0014 acceptance scenarios 1–4 and 6 are implemented and enabled; scenario 5 (crash recovery) remains ignored.

Changes

ParquetRecordSink: buffered flush to Parquet with RFC0014 acceptance tests

Layer / File(s)	Summary
Public contracts, struct definitions, and module wiring crates/ourios-ingester/Cargo.toml, crates/ourios-ingester/src/lib.rs, crates/ourios-ingester/src/record_sink.rs	Adds uuid v7 feature, exposes record_sink as pub mod, documents the buffering/flush policy, and defines FlushConfig (3 knobs), FlushError (encode/store variants with Display/Error), PartitionBuffer, and ParquetRecordSink struct fields and counters.
Internal helpers, constructor, and read-only metrics crates/ourios-ingester/src/record_sink.rs	Implements estimate_bytes (conservative per-record estimator), object_key (UUID v7-based Parquet key from partition path), ParquetRecordSink::new, and all public read-only accessors (buffered_bytes, buffered_partitions, flushes, records_flushed, buffered_records).
Flush logic and RecordSink::emit with ceiling enforcement crates/ourios-ingester/src/record_sink.rs	Implements flush_all, flush_aged, flush_partition (encode+write+counter updates on success, buffer retained on failure), flush_partition_swallow, flush_largest (ceiling loop control), and RecordSink::emit (derive, ceiling loop, buffer append, per-partition size trigger).
RFC0014 acceptance tests crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs	Adds rec_for helper, parquet file reader, and sink constructor; implements and enables RFC0014.1 (size trigger), RFC0014.2 (age trigger via flush_aged), RFC0014.3 (rotation via flush_all across 3 partitions), RFC0014.4 (ceiling never exceeded during emits), and RFC0014.6 (per-object tenant isolation); RFC0014.5 (crash recovery) stays #[ignore].

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• jensholdgaard/ourios#44: ParquetRecordSink::flush_partition calls encode_records_to_parquet and derives PartitionKey, both of which are implemented in this earlier PR's Parquet writer and partition key derivation code.
• jensholdgaard/ourios#231: ParquetRecordSink writes encoded bytes via the Store::put interface first introduced in this PR's local backend implementation.
• jensholdgaard/ourios#242: This PR implements and unignores the same RFC0014.1–.6 test stubs (todo!() bodies) added as placeholders in that prior PR.

Poem

🐇 Hoppity-hop, the records flow in,
Buffered by partition, ready to win!
A ceiling so firm, no bytes overflow,
Flush by size, age, or rotation's glow.
Parquet files bloom in the store so bright —
RFC 0014 shines in green light! ✨

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title directly and specifically describes the main change: implementing ParquetRecordSink and its flush policy for RFC 0014 part 1.
Description check	✅ Passed	The PR description comprehensively covers the implementation (What), the hybrid flush policy mechanisms, tenant isolation, testing status, and deferred work, though the checklist items are not explicitly marked.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch rfc0014-green-record-sink

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[Copilot]

Pull request overview

Adds the first “green” slice of RFC 0014’s ingest write path by introducing a production RecordSink implementation that buffers mined records by partition and flushes them to Parquet objects via the RFC 0013 Store seam, along with acceptance tests for the flush policy scenarios.

Changes:

• Introduces ParquetRecordSink + FlushConfig implementing hybrid size/age/rotation flush triggers and parquet object writes.
• Converts RFC0014 acceptance tests .1–.4/.6 from ignored stubs into real round-trip tests against a local filesystem Store.
• Enables UUIDv7 support for deterministic RFC0005-style object naming.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs	Implements RFC0014 scenario tests exercising sink flush behavior and tenant isolation via read-back.
crates/ourios-ingester/src/record_sink.rs	Adds ParquetRecordSink buffering + flush policy and Parquet object publication via Store::put_blocking.
crates/ourios-ingester/src/lib.rs	Exposes the new record_sink module.
crates/ourios-ingester/Cargo.toml	Enables uuid feature v7 for UUIDv7 object naming.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

crates/ourios-ingester/src/record_sink.rs (1)

82-91: ⚡ Quick win

Expose flush_errors and derive_errors for external observability.

These error counters are tracked internally but have no public getters, preventing operators from instrumenting them for alerting. Following the crate's CompactionMetrics pattern, consider adding:

#[must_use]
pub fn flush_errors(&self) -> u64 {
    self.flush_errors
}

#[must_use]
pub fn derive_errors(&self) -> u64 {
    self.derive_errors
}

This enables the metrics layer to export these counters to Prometheus when the sink is wired into the pipeline. As per coding guidelines: "Use Prometheus metrics for every subsystem."

🤖 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/record_sink.rs` around lines 82 - 91, Add public
getter methods to the ParquetRecordSink struct to expose the internal error
counters for external observability. Implement two public methods,
`flush_errors()` and `derive_errors()`, each annotated with the `#[must_use]`
attribute, that return the corresponding u64 values from the struct fields
flush_errors and derive_errors respectively. This follows the same pattern used
in CompactionMetrics and allows the metrics layer to access these counters for
Prometheus export without direct field access.

Source: Coding guidelines

🤖 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-ingester/tests/rfc0014_ingest_write_path.rs`:
- Around line 249-270: The test currently only validates per-file single-tenant
constraints and specific row counts for tenant-x and tenant-y, but does not
prevent extra rows from unexpected tenants from being present. After the
existing assertion on the (x, y) tuple, add two additional assertions: first,
verify that the total count of all rows across all files equals exactly x plus y
(ensuring no extra rows), and second, collect and assert that the complete set
of unique tenants across all files equals exactly the expected set of tenant-x
and tenant-y with no additional tenants present.

---

Nitpick comments:
In `@crates/ourios-ingester/src/record_sink.rs`:
- Around line 82-91: Add public getter methods to the ParquetRecordSink struct
to expose the internal error counters for external observability. Implement two
public methods, `flush_errors()` and `derive_errors()`, each annotated with the
`#[must_use]` attribute, that return the corresponding u64 values from the
struct fields flush_errors and derive_errors respectively. This follows the same
pattern used in CompactionMetrics and allows the metrics layer to access these
counters for Prometheus export without direct field access.

🪄 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: b0c92b3b-2343-44fb-b30d-a2b8696b3911

📥 Commits

Reviewing files that changed from the base of the PR and between e92c7af and 2cf3002.

📒 Files selected for processing (4)

• crates/ourios-ingester/Cargo.toml
• crates/ourios-ingester/src/lib.rs
• crates/ourios-ingester/src/record_sink.rs
• crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.