feat(parquet)!: rfc 0025 green a — body_kind ordinal 2 for absent bodies

crates/ourios-miner/src/cluster.rs

@@ -1153,10 +1153,10 @@ impl MinerCluster {
                 // record with `BodyKind::Absent` and the
                 // template-id sentinel; tokenize/mask didn't
                 // run, so there's no separator / param info to
-                // carry. `lossy_flag = true` because there is no
-                // template, so reconstruction is not possible.
-                let mut rec = Self::record_envelope(record, BodyKind::Absent);
-                rec.lossy_flag = true;
+                // carry. `lossy_flag = false` per RFC 0025 §3.1:
+                // absence is not loss — reconstruction is defined
+                // and total (it renders nothing).
+                let rec = Self::record_envelope(record, BodyKind::Absent);
                 self.emit_record(rec, service);
                 NO_TEMPLATE
             }
@@ -4331,7 +4331,10 @@ mod tests {
         assert_eq!(rec.tenant_id, t);
         assert_eq!(rec.template_id, NO_TEMPLATE);
         assert_eq!(rec.body_kind, BodyKind::Absent);
-        assert!(rec.lossy_flag, "Body::None records are lossy (no template)");
+        assert!(
+            !rec.lossy_flag,
+            "absence is not loss (RFC 0025 §3.1) — reconstruction renders nothing, exactly"
+        );
         assert!(rec.separators.is_empty());
         assert!(rec.params.is_empty());
         assert!(rec.body.is_none());

crates/ourios-parquet/src/reader.rs

@@ -682,9 +682,13 @@ fn decode_body_kind(ord: u8) -> Result<BodyKind, ReaderError> {
     match ord {
         0 => Ok(BodyKind::String),
         1 => Ok(BodyKind::Structured),
+        2 => Ok(BodyKind::Absent),
         other => Err(ReaderError::Conversion {
             column: columns::BODY_KIND,
-            detail: format!("unknown ordinal {other} (RFC 0005 §3.2 pins 0=String, 1=Structured)"),
+            detail: format!(
+                "unknown ordinal {other} (RFC 0005 §3.2 pins 0=String, 1=Structured, \
+                 2=Absent per RFC 0025)"
+            ),
         }),
     }
 }

crates/ourios-parquet/src/record_batch.rs

@@ -166,16 +166,12 @@ pub enum BatchError {
         count: usize,
         source: ourios_core::otlp::canonical::CanonicalJsonError,
     },
-    /// A record carried [`BodyKind::Absent`] (the in-memory
-    /// "wire delivered no body" variant). RFC 0005 §3.2's
-    /// `body_kind` column pins exactly two ordinals (`0 = String,
-    /// 1 = Structured`); silently mapping `Absent` to one of
-    /// them would misclassify wire-absent rows. Until a future
-    /// RFC 0005 amendment either adds a third ordinal or adds a
-    /// separate `body_present` boolean column, the writer
-    /// rejects these records rather than corrupting the
-    /// `body_kind` semantics.
-    UnsupportedAbsentBody,
+    /// A `body_kind = Absent` record carried body bytes. RFC 0025
+    /// §3.1 pins the on-disk contract as ordinal 2 **with a `NULL`
+    /// `body` cell**; silently writing the bytes would smuggle an
+    /// undefined state into the schema, so the writer rejects the
+    /// producer bug loudly.
+    NonNullBodyForAbsent,
     /// A clean-attach `body_kind = String` record had too few
     /// `separators` entries to satisfy the RFC 0005 §3.2
     /// invariant ("`tokens.len() + 1` elements when
@@ -226,13 +222,6 @@ impl fmt::Display for BatchError {
                  `with-serde` derives are infallible on every spec-compliant `AnyValue`; \
                  this means an `opentelemetry-proto` upgrade broke that contract)",
             ),
-            Self::UnsupportedAbsentBody => write!(
-                f,
-                "record carries BodyKind::Absent (wire-absent body), which RFC 0005 §3.2's \
-                 body_kind column does not yet encode (the column pins ordinals 0=String, \
-                 1=Structured); a future RFC 0005 amendment is required to represent this \
-                 in the schema",
-            ),
             Self::InvalidSeparatorsForString {
                 expected_at_least,
                 actual,
@@ -250,6 +239,12 @@ impl fmt::Display for BatchError {
                  reconstruction path returns the retained body verbatim — without one, the \
                  record is unreconstructable on read",
             ),
+            Self::NonNullBodyForAbsent => write!(
+                f,
+                "body_kind = Absent record carries body bytes, but RFC 0025 §3.1 pins the \
+                 Absent on-disk contract as a NULL body cell — a producer bug, rejected \
+                 rather than written",
+            ),
             Self::Arrow(e) => write!(f, "arrow rejected RecordBatch: {e}"),
         }
     }
@@ -259,9 +254,9 @@ impl std::error::Error for BatchError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             Self::TimestampOverflow { .. }
-            | Self::UnsupportedAbsentBody
             | Self::InvalidSeparatorsForString { .. }
-            | Self::MissingBodyForLossyString => None,
+            | Self::MissingBodyForLossyString
+            | Self::NonNullBodyForAbsent => None,
             Self::AttributeEncode { source, .. } => Some(source),
             Self::Arrow(e) => Some(e),
         }
@@ -421,7 +416,7 @@ impl Builders {
         self.flags.append_value(r.flags);
         append_option_str(&mut self.event_name, r.event_name.as_deref());
 
-        self.body_kind.append_value(body_kind_ordinal(r.body_kind)?);
+        self.body_kind.append_value(body_kind_ordinal(r.body_kind));
         // RFC 0005 §3.3: when `body_kind = Structured`, the
         // body column carries Ourios-canonical JSON — the bytes
         // the miner has already encoded via
@@ -432,6 +427,9 @@ impl Builders {
         // is the retained line bytes on the §6.6 lossy path
         // (or `None` on the clean-attach path, reconstructed
         // from `template + params + separators` by the reader).
+        if r.body_kind == BodyKind::Absent && r.body.is_some() {
+            return Err(BatchError::NonNullBodyForAbsent);
+        }
         match r.body.as_deref() {
             Some(s) => self.body.append_value(s.as_bytes()),
             None => self.body.append_null(),
@@ -534,16 +532,14 @@ fn append_option_str(b: &mut StringBuilder, v: Option<&str>) {
 }
 
 /// Map an in-memory [`BodyKind`] to the §3.2 on-disk `body_kind`
-/// ordinal. The schema pins exactly two ordinals (`0 = String,
-/// 1 = Structured`); `BodyKind::Absent` has no on-disk
-/// representation today and the writer rejects records carrying
-/// it via [`BatchError::UnsupportedAbsentBody`] rather than
-/// silently misclassifying them.
-fn body_kind_ordinal(k: BodyKind) -> Result<u8, BatchError> {
+/// ordinal (`0 = String, 1 = Structured, 2 = Absent` — the third
+/// ordinal is the RFC 0025 §3.1 amendment; wire-absent rows carry a
+/// `NULL` body cell).
+fn body_kind_ordinal(k: BodyKind) -> u8 {
     match k {
-        BodyKind::String => Ok(0),
-        BodyKind::Structured => Ok(1),
-        BodyKind::Absent => Err(BatchError::UnsupportedAbsentBody),
+        BodyKind::String => 0,
+        BodyKind::Structured => 1,
+        BodyKind::Absent => 2,
     }
 }
 
@@ -840,18 +836,42 @@ mod tests {
         assert_eq!(stored.value(0), canonical.as_bytes());
     }
 
-    /// `BodyKind::Absent` is not representable in the §3.2
-    /// `body_kind` column today (the ordinals pin to
-    /// `0 = String, 1 = Structured`). The writer rejects such
-    /// records rather than silently lumping them with String.
+    /// RFC 0025 §3.1: `BodyKind::Absent` writes ordinal 2 with a
+    /// `NULL` body cell — the contract change that retired the old
+    /// `UnsupportedAbsentBody` rejection (approved via RFC 0025;
+    /// wire-absent bodies are spec-legal and must persist).
+    #[test]
+    fn absent_body_kind_writes_ordinal_two_with_null_body() {
+        let mut rec = empty_record();
+        rec.body_kind = BodyKind::Absent;
+        rec.body = None;
+        let batch = mined_records_to_batch(&[rec]).expect("Absent body must write");
+        let kind_idx = batch.schema().index_of(crate::columns::BODY_KIND).unwrap();
+        let kinds = batch
+            .column(kind_idx)
+            .as_any()
+            .downcast_ref::<arrow_array::UInt8Array>()
+            .expect("body_kind is UInt8");
+        assert_eq!(kinds.value(0), 2, "Absent is ordinal 2 (RFC 0025 §3.1)");
+        let body_idx = batch.schema().index_of(crate::columns::BODY).unwrap();
+        assert!(
+            batch.column(body_idx).is_null(0),
+            "the body cell is NULL for Absent rows",
+        );
+    }
+
+    /// The §3.1 contract's other direction: an Absent record
+    /// carrying body bytes is a producer bug and must be rejected
+    /// loudly, never written.
     #[test]
-    fn absent_body_kind_returns_unsupported_error() {
+    fn absent_body_kind_with_bytes_is_rejected() {
         let mut rec = empty_record();
         rec.body_kind = BodyKind::Absent;
-        let err = mined_records_to_batch(&[rec]).expect_err("Absent body must error");
+        rec.body = Some("stray bytes".to_string());
+        let err = mined_records_to_batch(&[rec]).expect_err("must reject");
         assert!(
-            matches!(err, BatchError::UnsupportedAbsentBody),
-            "expected UnsupportedAbsentBody, got {err:?}",
+            matches!(err, BatchError::NonNullBodyForAbsent),
+            "expected NonNullBodyForAbsent, got {err:?}",
         );
     }
 

crates/ourios-parquet/tests/rfc0024_properties.rs

@@ -78,7 +78,8 @@ fn fail(what: &str, e: impl std::fmt::Display) -> TestCaseError {
 /// Mine `batch`, store every writable row, read it back, and assert
 /// fidelity. Adversarial timestamps can exceed `i64::MAX`, which the
 /// writer *rejects by contract* (RFC 0005 §3.2 timestamp overflow) —
-/// those rows are asserted rejected, everything else must round-trip.
+/// those rows are asserted rejected, everything else (absent bodies
+/// included, per RFC 0025) must round-trip.
 fn assert_round_trip(batch: &[OtlpLogRecord]) -> Result<(), TestCaseError> {
     let sink = SharedRecordSink::new();
     let mut cluster =
@@ -97,24 +98,17 @@ fn assert_round_trip(batch: &[OtlpLogRecord]) -> Result<(), TestCaseError> {
     let mut groups: Vec<(PartitionKey, Vec<(usize, MinedRecord)>)> = Vec::new();
     let mut writable = vec![false; batch.len()];
     for (i, mined) in emitted.into_iter().enumerate() {
-        let absent_body = mined.body_kind == ourios_core::record::BodyKind::Absent;
         let ts_overflow = !fits_i64(mined.time_unix_nano)
             || mined.observed_time_unix_nano.is_some_and(|t| !fits_i64(t));
-        if absent_body || ts_overflow {
-            // - Absent body: KNOWN GAP (#362, found by this suite) —
-            //   no §3.2 on-disk representation until the RFC 0005
-            //   amendment lands; this arm then turns into a
-            //   round-trip.
-            // - Timestamp overflow: the §3.2 u64→i64 contract, on
-            //   *either* timestamp column.
-            // Both must be loud rejections, never silent drops.
+        if ts_overflow {
+            // Timestamp overflow: the §3.2 u64→i64 contract, on
+            // *either* timestamp column — the writer's one remaining
+            // documented loud rejection. (Absent bodies round-trip
+            // since RFC 0025 §3.1 gave them ordinal 2.)
             let err = mined_records_to_batch(std::slice::from_ref(&mined))
                 .expect_err("the writer must reject this record, not silently map it");
             prop_assert!(
-                matches!(
-                    err,
-                    BatchError::UnsupportedAbsentBody | BatchError::TimestampOverflow { .. }
-                ),
+                matches!(err, BatchError::TimestampOverflow { .. }),
                 "record {}: rejected for an undocumented reason: {}",
                 i,
                 err