feat(parquet): Writer + partition derivation + RecordBatch builder (PR-E2)

[jensholdgaard]

Summary

Lands the Writer half of RFC 0005 §3.10's crate-shape plan. Reader and audit stream are still pending (PR-F, PR-G).

• partition::PartitionKey — derives the §3.4 partition tuple (tenant_id + year/month/day/hour) via the time-fallback algorithm (time_unix_nano → observed_time_unix_nano → 1970 epoch). Same algorithm the reader's row-vs-path validation will use (§3.9). percent_encode_tenant implements §3.4's RFC 3986 percent-encoding with explicit overrides.
• record_batch::mined_records_to_batch — column-by-column Arrow RecordBatch builder matching data_schema(). Per-record validation enforces RFC 0005 §3.2 / RFC 0001 §6.6 invariants: timestamp overflow rejection (u64→i64), BodyKind::Absent rejection (§3.2 ordinals pin to 0/1), BodyKind::Structured rejection until the canonicalisation PR replaces the miner's interim Debug rendering, attributes/resource_attributes non-empty rejection until the same canonicalisation PR, separators.len() >= params.len() + 1 for clean-attach String rows, body.is_some() for lossy rows. attributes/resource_attributes serialise to literal "[]" for empty Vec<KeyValue> per §3.2's round-trip rule.
• writer::Writer — opens a UUIDv7-named Parquet file at the partition path, validates each record's derived partition against the writer's open partition (fails fast with PartitionMismatch so the §3.9 row-vs-path contract is enforced at write time too). Internal chunking of records into 1024-row sub-batches plus per-sub-batch flush-on-threshold keeps row-group sizes bounded to roughly 128 MiB + (per-record bytes × 1024) — well under §3.5's 1 GiB upper bound for log-scale per-record sizes. Applies the §3.6 encoding policy: ZSTD-3 across all columns; explicit Dictionary = no on time_unix_nano/observed_time_unix_nano/attributes/trace_id/span_id/body/confidence/params.list.element.{type_tag,value} (both leaves per the §3.6 "(list values)" reading); per-column EnabledStatistics::Chunk (page-index off) on tenant_id/attributes/resource_attributes/body/params.list.element.{type_tag,value}/separators.list.element; bloom filter on template_id for B2 predicate-pushdown.
• Atomic publish — writes to <uuid>.parquet.tmp, Writer::close renames to <uuid>.parquet after the footer is on disk, Drop removes the .tmp if close wasn't called. Satisfies RFC 0005 §7's atomic-publish open-question item. WriterError::Io carries {op, path, source_path, source} so a failed rename leaves a .parquet.tmp for diagnosis and the error message names exactly which file.

Invariant coverage

• RFC0005.5 (tests/partition_layout.rs) — multi-tenant, multi-hour files land at the expected partition paths with UUIDv7 filenames; non-ASCII tenant id percent-encodes to %C3%A5; cross-partition records rejected at write time; atomic-publish round-trip (drop without close → temp gone, no final; close → final exists, no temp).
• RFC0005.8 (tests/no_body_dict.rs) — writes 200 unique high-entropy bodies; Parquet footer shows the body column with ZSTD compression, no PLAIN_DICTIONARY/RLE_DICTIONARY encoding, and no dictionary_page_offset on disk. A sibling test exercises the §3.6 params.list.element.{value, type_tag} no-dict contract over both leaves.
• Six unit tests in record_batch.rs pin the empty-attributes "[]" serialisation, the non-empty AttributesNotYetEncoded errors on both attributes columns, the UnsupportedAbsentBody rejection, the StructuredBodyNotYetCanonical rejection, the InvalidSeparatorsForString { expected_at_least, actual } lower-bound check on clean-attach String rows, and the MissingBodyForLossyString rejection on the lossy carve-out.

Deferred to PR-F (reader) and PR-G (audit stream): RFC0005.1 (round-trip), RFC0005.2/3/4 (forward-compat), RFC0005.6 (#[ignore]-d row-group sizing), RFC0005.7 (audit emission), RFC0005.9 (unknown ParamType ordinal), RFC0005.11 (row-vs-path validation — needs reader).

Test plan

• cargo build -p ourios-parquet — clean
• cargo test --all-features -p ourios-parquet — 25 pass
• cargo fmt --all --check — clean
• cargo clippy --all-targets --all-features -- -D warnings — clean
• CI green

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Added Parquet file writing capabilities with RFC 0005–compliant partitioning
  • Implemented atomic file publishing and row-level partition validation
  • Added support for non-ASCII tenant identifiers with RFC 3986 percent-encoding
  • Introduced comprehensive error handling for timestamp overflow, batch validation, and I/O operations

• Chores

  • Updated project dependencies to support Parquet and Arrow functionality

[Copilot]

Pull request overview

Implements the “writer half” of RFC 0005’s Parquet storage plan in ourios-parquet: partition-path derivation, Arrow RecordBatch construction for mined records, and a Parquet writer configured for the RFC’s row-group sizing and encoding policies, plus RFC-scenario tests.

Changes:

• Add partition::PartitionKey and tenant percent-encoding to derive Hive-style partition directories from MinedRecord timestamps.
• Add record_batch::mined_records_to_batch to build Arrow arrays/batches matching data_schema().
• Add writer::Writer to write Parquet files (UUIDv7 filenames), apply compression/dictionary/bloom settings, and flush row groups based on in-progress size; add RFC0005.5 and RFC0005.8 tests.

Reviewed changes

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

Show a summary per file

File	Description
crates/ourios-parquet/src/lib.rs	Exposes new partition, batch-building, and writer modules from the crate API.
crates/ourios-parquet/src/partition.rs	Derives (tenant_id, year, month, day, hour) partition keys and percent-encodes tenant IDs for paths.
crates/ourios-parquet/src/record_batch.rs	Builds Arrow RecordBatches from MinedRecord slices in RFC schema order.
crates/ourios-parquet/src/writer.rs	Implements Parquet writer + encoding configuration + row-group flushing.
crates/ourios-parquet/tests/partition_layout.rs	RFC0005.5 scenario test for partition layout + UUIDv7 filenames + tenant percent-encoding.
crates/ourios-parquet/tests/no_body_dict.rs	RFC0005.8 scenario test asserting body has no dictionary encoding in Parquet metadata.
crates/ourios-parquet/Cargo.toml	Adds Arrow array + Parquet + UUID + chrono dependencies needed by writer/batch code and tests.
Cargo.lock	Locks new transitive dependencies introduced by Arrow/Parquet/uuid/chrono/tempfile additions.

Comments suppressed due to low confidence (3)

crates/ourios-parquet/src/writer.rs:214

• RFC 0005 §3.6 specifies params (list values) must have dictionary encoding disabled, but writer_properties() only disables dictionary for body, attributes, trace_id, and span_id. With set_dictionary_enabled(true) globally, the params column (and in particular its nested value field) will still be dictionary-encoded unless explicitly opted out, which violates the encoding policy and can reintroduce the high-cardinality blow-up the RFC calls out.

    // §3.6: NO dictionary on the high-entropy attribute / id
    // columns (`attributes`, `trace_id`, `span_id`, `params`
    // values). Page index stays on for `trace_id` / `span_id`
    // per the §3.6 table; on `attributes` it's `no`/`no`/`no`.
    for high_entropy in [
        crate::columns::ATTRIBUTES,
        crate::columns::TRACE_ID,
        crate::columns::SPAN_ID,
    ] {
        builder = builder
            .set_column_dictionary_enabled(ColumnPath::new(vec![high_entropy.to_string()]), false);
    }

crates/ourios-parquet/src/writer.rs:196

• The RFC 0005 §3.6 table has per-column Page index on/off settings, but the implementation only calls set_statistics_enabled(EnabledStatistics::Page) globally and does not apply any per-column page-index overrides (and the comment conflates page index with statistics). This risks producing files that don’t match the RFC’s page-index policy; please align the WriterProperties configuration with the §3.6 page-index column (either via an explicit page-index setting or by documenting exactly how EnabledStatistics::Page maps to page indexes in parquet-rs).

        // Dictionary on globally by default (most columns benefit
        // per §3.6); we opt out per-column below for the high-
        // entropy ones.
        .set_dictionary_enabled(true)
        // Page-index ON by default — §3.6's "page index" column
        // is `yes` for most columns; cheaper to enable globally
        // and skip per-column overrides.
        .set_statistics_enabled(EnabledStatistics::Page);

crates/ourios-parquet/src/writer.rs:108

• RFC 0005 §3.5 requires row groups never exceed 1 GiB (the row that would cross the ceiling starts a new row group). append_records() only flushes once after writing the whole RecordBatch, so a single large records slice (or one very large batch) can produce an oversized row group before the post-write flush runs. To meet the 1 GiB ceiling, consider chunking/splitting input batches and flushing in a loop based on in_progress_size, or configuring an explicit max row-group size in the Parquet writer if parquet-rs supports it.

    pub fn append_records(&mut self, records: &[MinedRecord]) -> Result<(), WriterError> {
        if records.is_empty() {
            return Ok(());
        }
        let batch = mined_records_to_batch(records).map_err(WriterError::Batch)?;
        self.inner.write(&batch).map_err(WriterError::Parquet)?;
        if self.inner.in_progress_size() >= ROW_GROUP_FLUSH_BYTES {
            self.inner.flush().map_err(WriterError::Parquet)?;
        }
        Ok(())

---

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

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (2)

crates/ourios-parquet/src/record_batch.rs:351

• encode_attributes panics for non-empty attribute lists (unimplemented!). This is reachable with real MinedRecords (the type allows non-empty attributes/resource_attributes) and will crash the writer instead of returning a structured BatchError. Consider changing this to return an error (e.g., a new BatchError variant) or implementing a temporary encoding that preserves forward progress until the canonical-JSON PR lands.

fn encode_attributes(attrs: &[KeyValue]) -> String {
    if attrs.is_empty() {
        return "[]".to_string();
    }
    unimplemented!(
        "ourios-parquet: canonical JSON encoding of non-empty KeyValue lists is deferred to \
         the RFC 0005 §3.3 canonicalisation PR (see the PR-E1 breadcrumb on \
         ourios_core::otlp::Body::Structured). Got {} entries — corpus / bench inputs \
         today carry empty attributes; the RFC 0003 receiver is what populates them.",
        attrs.len(),
    );
}

crates/ourios-parquet/src/writer.rs:267

• The RFC 0005 §3.6 encoding table specifies dictionary encoding should be OFF for params list values, but writer_properties currently only disables dictionary for body, attributes, trace_id, and span_id. With set_dictionary_enabled(true) globally, the nested params leaf columns will likely still be dictionary-encoded unless explicitly disabled via their ColumnPaths.

    // §3.6: NO dictionary on the high-entropy attribute / id
    // columns (`attributes`, `trace_id`, `span_id`, `params`
    // values). Page index stays on for `trace_id` / `span_id`
    // per the §3.6 table; on `attributes` it's `no`/`no`/`no`.
    for high_entropy in [
        crate::columns::ATTRIBUTES,
        crate::columns::TRACE_ID,
        crate::columns::SPAN_ID,
    ] {
        builder = builder
            .set_column_dictionary_enabled(ColumnPath::new(vec![high_entropy.to_string()]), false);
    }

[Copilot]

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated 6 comments.

Comments suppressed due to low confidence (2)

crates/ourios-parquet/src/writer.rs:267

• The comment says dictionary is disabled for high-entropy columns including params values, but the implementation only disables it for attributes, trace_id, and span_id (and body above). Either extend the per-column overrides to include params (if intended by RFC §3.6) or adjust the comment to avoid misleading future changes.

    // §3.6: NO dictionary on the high-entropy attribute / id
    // columns (`attributes`, `trace_id`, `span_id`, `params`
    // values). Page index stays on for `trace_id` / `span_id`
    // per the §3.6 table; on `attributes` it's `no`/`no`/`no`.
    for high_entropy in [

crates/ourios-parquet/src/partition.rs:305

• This unit test compares PathBuf via to_str() against a hard-coded Unix path string. This is OS-specific (path separators / prefix handling). Prefer comparing PathBuf values (or components()) built with Path::join so the test remains portable.

        let bucket = Path::new("/tmp/bucket");
        let path = key.audit_path(bucket);
        assert_eq!(
            path.to_str().unwrap(),
            "/tmp/bucket/audit/tenant_id=tenant-x/year=2026/month=04/day=02"
        );

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated 1 comment.