diff --git a/crates/ourios-bench/benches/b1.rs b/crates/ourios-bench/benches/b1.rs index 5f09b817b..2fb0a4fd7 100644 --- a/crates/ourios-bench/benches/b1.rs +++ b/crates/ourios-bench/benches/b1.rs @@ -25,8 +25,10 @@ //! reference (` ` per record — the flat file a //! traditional logger would have written, one zstd block per hour) //! and picks the severity to query (`ERROR`, or the busiest -//! error-band text). The query window is the corpus's full timestamp -//! span, so the reference scans the same in-window set. **OTLP +//! error-band text). The query window is the corpus's full +//! *effective*-timestamp span (RFC 0005 §3.2 — `timeUnixNano`, else +//! `observedTimeUnixNano`), so the reference scans the same in-window +//! set and observed-only corpora are B1-eligible. **OTLP //! corpora only**: the RFC 0006 §3.3 plain-text loader fixes every //! line at severity `9` / `INFO`, so a severity predicate over a //! plain-text corpus has no selectivity — such dirs are skipped with @@ -272,7 +274,9 @@ fn real_corpus(c: &mut Criterion) { } /// The B1 query for a built real-corpus store: the chosen error-band -/// severity over the corpus's full timestamp span, half-open +/// severity over the corpus's full *effective*-timestamp span +/// (RFC 0005 §3.2 amendment 2026-06-11 — observed-only corpora are +/// B1-eligible; only genuinely timeless rows disqualify), half-open /// (`max + 1` keeps the last record in-window; clamped like b2's /// windowed arm so a near-2262 corpus can't push the bound past the /// querier's i64 range). `None` (with a skip note) when the corpus @@ -291,22 +295,23 @@ fn severity_query(built: &ourios_bench::B1Store, dir: &str) -> Option<(String, Q eprintln!("b1/real-corpus: {dir} has no error-band severity_text rows — skipping"); return None; }; - if built.min_time_unix_nano == 0 || built.zero_ts_rows > 0 { + if built.min_effective_time_unix_nano == 0 || built.zero_effective_ts_rows > 0 { eprintln!( - "b1/real-corpus: {dir} — no usable timestamp span ({} zero-ts rows); the B1 \ - query needs a real time window — skipping", - built.zero_ts_rows, + "b1/real-corpus: {dir} — no usable effective-timestamp span ({} \ + zero-effective-ts rows, i.e. neither timeUnixNano nor \ + observedTimeUnixNano); the B1 query needs a real time window — skipping", + built.zero_effective_ts_rows, ); return None; } #[allow(clippy::cast_sign_loss)] // i64::MAX as u64 is exact let window_end = built - .max_time_unix_nano + .max_effective_time_unix_nano .saturating_add(1) .min(i64::MAX as u64); let query = QueryRequest { tenant: TenantId::new(built.tenant), - time_range: Some((built.min_time_unix_nano, window_end)), + time_range: Some((built.min_effective_time_unix_nano, window_end)), template_id: None, severity_text: Some(severity.clone()), }; diff --git a/crates/ourios-bench/benches/b2.rs b/crates/ourios-bench/benches/b2.rs index 5c79322b6..1688f32d1 100644 --- a/crates/ourios-bench/benches/b2.rs +++ b/crates/ourios-bench/benches/b2.rs @@ -272,13 +272,14 @@ fn first_hour_window( dir: &str, baseline_row_groups: u64, ) -> Option { - if built.min_time_unix_nano == 0 || built.files < 2 { + if built.min_effective_time_unix_nano == 0 || built.files < 2 { eprintln!( "b2/real-corpus: {dir} — single-partition or no timestamp span; skipping windowed arm" ); return None; } - let hour_start = built.min_time_unix_nano - (built.min_time_unix_nano % HOUR_NS); + let hour_start = + built.min_effective_time_unix_nano - (built.min_effective_time_unix_nano % HOUR_NS); // Clamp the window end into the querier's i64-nanosecond range, so a // corpus near the year-2262 boundary can't push the bound past // i64::MAX (which the querier rejects as InvalidQuery, panicking the diff --git a/crates/ourios-bench/src/store.rs b/crates/ourios-bench/src/store.rs index bf3909a61..686628a75 100644 --- a/crates/ourios-bench/src/store.rs +++ b/crates/ourios-bench/src/store.rs @@ -45,12 +45,16 @@ pub struct BuiltStore { /// How many rows that busiest template has (the result size a /// `template_id = busiest_template_id` query returns). pub busiest_template_rows: u64, - /// Smallest non-zero `time_unix_nano` written (`0` if none) — the - /// start of the corpus's time span, for picking a B2 query window. - pub min_time_unix_nano: u64, - /// Largest `time_unix_nano` written (`0` if none) — the end of the - /// corpus's time span. - pub max_time_unix_nano: u64, + /// Smallest non-zero **effective** timestamp written (`0` if none) — + /// the start of the corpus's time span, for picking a B2 query + /// window. Effective per RFC 0005 §3.2 (amendment 2026-06-11): + /// `time_unix_nano`, else `observed_time_unix_nano` — the value the + /// query window actually filters, derived via the same + /// `ourios_parquet::effective_time_unix_nano` the writer stores. + pub min_effective_time_unix_nano: u64, + /// Largest effective timestamp written (`0` if none) — the end of + /// the corpus's time span. + pub max_effective_time_unix_nano: u64, } /// Load the corpus at `corpus_dir`, mine it, and write the emitted @@ -72,7 +76,7 @@ pub struct BuiltStore { pub fn build_query_store(corpus_dir: &Path, bucket_root: &Path) -> Result { let mut counts: HashMap = HashMap::new(); - let core = build_store(corpus_dir, bucket_root, |_input, emitted| { + let core = build_store(corpus_dir, bucket_root, |_input, emitted, _effective| { *counts.entry(emitted.template_id).or_insert(0) += 1; Ok(()) })?; @@ -86,8 +90,8 @@ pub fn build_query_store(corpus_dir: &Path, bucket_root: &Path) -> Result, /// The `zstdcat | grep` baseline input: every record with a - /// non-zero timestamp rendered as the flat-text line a + /// non-zero effective timestamp rendered as the flat-text line a /// traditional logger would have written /// (` `), compressed one block per hour — /// the hour granularity mirrors the store's partitioning, i.e. - /// the `*.zst` segments `files_in_range.zst` would name. Zero-ts - /// rows are excluded: they sit outside any window and the bench - /// skips such corpora. + /// the `*.zst` segments `files_in_range.zst` would name. + /// Zero-effective-ts rows are excluded: they sit outside any + /// window and the bench skips such corpora. pub reference: ReferenceCorpus, } @@ -157,25 +167,27 @@ pub fn build_b1_store( let mut spool = HourSpool::new().map_err(|e| BenchError::Pipeline { detail: format!("create B1 reference spool: {e}"), })?; - let mut zero_ts_rows = 0u64; + let mut zero_effective_ts_rows = 0u64; - let core = build_store(corpus_dir, bucket_root, |input, emitted| { + let core = build_store(corpus_dir, bucket_root, |input, emitted, effective| { if let Some(text) = &emitted.severity_text { *severity_rows.entry(text.clone()).or_insert(0) += 1; if ERROR_BAND.contains(&emitted.severity_number) { *error_band_rows.entry(text.clone()).or_insert(0) += 1; } } - if emitted.time_unix_nano == 0 { - // Out-of-window by definition (the B1 arm skips any corpus - // carrying zero-ts rows); keep the reference strictly - // in-window rather than spooling lines no query scans. - zero_ts_rows += 1; + if effective == 0 { + // Genuinely timeless (neither wire timestamp set) — + // out-of-window by definition (the B1 arm skips any corpus + // carrying zero-effective-ts rows); keep the reference + // strictly in-window rather than spooling lines no query + // scans. + zero_effective_ts_rows += 1; return Ok(()); } let line = reference_line(input)?; spool - .append(emitted.time_unix_nano / HOUR_NS, &line) + .append(effective / HOUR_NS, &line) .map_err(|e| BenchError::Pipeline { detail: format!("spool B1 reference line: {e}"), })?; @@ -216,9 +228,9 @@ pub fn build_b1_store( tenant: crate::corpus::BENCH_TENANT, rows: core.rows, files: core.files, - min_time_unix_nano: core.min_time_unix_nano, - max_time_unix_nano: core.max_time_unix_nano, - zero_ts_rows, + min_effective_time_unix_nano: core.min_effective_time_unix_nano, + max_effective_time_unix_nano: core.max_effective_time_unix_nano, + zero_effective_ts_rows, distinct_severities: severity_rows.len(), query_severity, reference, @@ -309,19 +321,21 @@ impl HourSpool { struct StoreCore { rows: u64, files: u64, - min_time_unix_nano: u64, - max_time_unix_nano: u64, + min_effective_time_unix_nano: u64, + max_effective_time_unix_nano: u64, } /// The shared load → mine → write pipeline behind /// [`build_query_store`] and [`build_b1_store`]. `observe` runs once -/// per successfully-appended record; its first error aborts the -/// build (surfaced after the harness loop, same stash pattern as -/// `a1::A1Accumulator`). +/// per successfully-appended record — its third argument is the +/// record's effective timestamp (the shared writer/partition +/// derivation, so the bookkeeping can never disagree with the +/// store) — and its first error aborts the build (surfaced after the +/// harness loop, same stash pattern as `a1::A1Accumulator`). fn build_store( corpus_dir: &Path, bucket_root: &Path, - mut observe: impl FnMut(&OtlpLogRecord, &MinedRecord) -> Result<(), BenchError>, + mut observe: impl FnMut(&OtlpLogRecord, &MinedRecord, u64) -> Result<(), BenchError>, ) -> Result { // A reused bucket would let the querier enumerate a prior run's // Parquet too, mixing corpora and skewing both the row counts @@ -342,10 +356,11 @@ fn build_store( let mut writers: HashMap = HashMap::new(); let mut rows: u64 = 0; - // Track the corpus's `time_unix_nano` span so the benches can pick - // a real time window. Only non-zero timestamps count (a `0` falls - // back to observed/epoch for partitioning — not a meaningful - // window bound). + // Track the corpus's *effective*-timestamp span so the benches can + // pick a real time window — the query window filters the effective + // column (RFC 0002 §6.2 / RFC 0005 §3.2). Only non-zero values + // count (`0` means genuinely timeless: the epoch partition, not a + // meaningful window bound). let mut min_ts = u64::MAX; let mut max_ts = 0u64; // The harness callback returns `()`, so a write/observe error is @@ -357,14 +372,17 @@ fn build_store( if first_err.is_some() { return; } - let appended = append_record(&mut writers, bucket_root, emitted) - .and_then(|()| observe(input, emitted)); + let appended = effective_nanos(emitted).and_then(|effective| { + append_record(&mut writers, bucket_root, emitted)?; + observe(input, emitted, effective)?; + Ok(effective) + }); match appended { - Ok(()) => { + Ok(effective) => { rows += 1; - if emitted.time_unix_nano != 0 { - min_ts = min_ts.min(emitted.time_unix_nano); - max_ts = max_ts.max(emitted.time_unix_nano); + if effective != 0 { + min_ts = min_ts.min(effective); + max_ts = max_ts.max(effective); } } Err(e) => first_err = Some(e), @@ -382,8 +400,8 @@ fn build_store( })?; } - // No non-zero timestamp seen ⇒ no meaningful span (report 0, 0). - let (min_time_unix_nano, max_time_unix_nano) = if min_ts == u64::MAX { + // No non-zero effective timestamp ⇒ no meaningful span (0, 0). + let (min_effective_time_unix_nano, max_effective_time_unix_nano) = if min_ts == u64::MAX { (0, 0) } else { (min_ts, max_ts) @@ -392,8 +410,26 @@ fn build_store( Ok(StoreCore { rows, files, - min_time_unix_nano, - max_time_unix_nano, + min_effective_time_unix_nano, + max_effective_time_unix_nano, + }) +} + +/// The record's RFC 0005 §3.2 effective timestamp in the `u64` wire +/// domain — `ourios_parquet::effective_time_unix_nano`, the same +/// derivation the writer stores and the partition tuple uses, so the +/// bench's span / eligibility bookkeeping can never disagree with +/// what the query window filters. +fn effective_nanos(emitted: &MinedRecord) -> Result { + let effective = + ourios_parquet::effective_time_unix_nano(emitted).map_err(|e| BenchError::Pipeline { + detail: format!("effective timestamp derive failed: {e}"), + })?; + // The derivation validates both candidates against the u64→i64 + // overflow contract, so the i64 is never negative; keep the + // conversion total anyway rather than panicking. + u64::try_from(effective).map_err(|_| BenchError::Pipeline { + detail: format!("effective timestamp {effective} is negative"), }) } @@ -468,9 +504,12 @@ mod tests { let built = build_query_store(corpus.path(), bucket.path()).expect("build"); assert_eq!(built.rows, 3, "one record per line"); - assert_eq!(built.min_time_unix_nano, TIME_BASELINE_NS, "span start"); assert_eq!( - built.max_time_unix_nano, + built.min_effective_time_unix_nano, TIME_BASELINE_NS, + "span start" + ); + assert_eq!( + built.max_effective_time_unix_nano, TIME_BASELINE_NS + 2 * TIME_INCREMENT_NS, "span end (3rd line)", ); @@ -496,6 +535,79 @@ mod tests { ) } + /// Like [`logs_data_line`], but **observed-only**: `timeUnixNano` + /// is absent from the wire (the OTLP "source timestamp unknown" + /// case), `observedTimeUnixNano` carries `base + i` ns. + fn observed_only_logs_data_line(n: usize, text: &str, number: u8, base: u64) -> String { + let records: Vec = (0..n) + .map(|i| { + format!( + "{{\"observedTimeUnixNano\":\"{}\",\"severityNumber\":{number},\ + \"severityText\":\"{text}\",\ + \"body\":{{\"stringValue\":\"{text} event {i}\"}}}}", + base + u64::try_from(i).expect("usize fits in u64"), + ) + }) + .collect(); + format!( + "{{\"resourceLogs\":[{{\"scopeLogs\":[{{\"logRecords\":[{}]}}]}}]}}", + records.join(","), + ) + } + + /// RFC 0005 §3.2 rule 7 (the RFC0005.13 bench follow-up) — an + /// observed-only corpus (`timeUnixNano` absent, ~15 % of real + /// OTel-Demo records) is **B1-eligible**: the bookkeeping keys + /// off the effective timestamp, so `zero_effective_ts_rows` + /// stays 0, the span derives from the observed values, and every + /// line lands in the reference corpus. These are exactly the + /// outputs the `benches/b1.rs` `severity_query` guard checks. + #[test] + fn b1_store_with_observed_only_rows_is_eligible() { + let corpus = tempfile::TempDir::new().expect("corpus dir"); + let base = crate::corpus::TIME_BASELINE_NS; + let jsonl = format!( + "{}\n{}\n", + observed_only_logs_data_line(5, "INFO", 9, base), + observed_only_logs_data_line(3, "ERROR", 17, base + 1_000), + ); + std::fs::write(corpus.path().join("c.jsonl"), jsonl).expect("write corpus"); + let bucket = tempfile::TempDir::new().expect("bucket dir"); + + let built = build_b1_store(corpus.path(), bucket.path(), 3).expect("build"); + + // The b1 eligibility guard: a usable span and no + // zero-effective-ts rows. + assert_eq!( + built.zero_effective_ts_rows, 0, + "observed-only rows have a non-zero effective timestamp", + ); + assert_eq!( + built.min_effective_time_unix_nano, base, + "span start derives from the observed fallback", + ); + assert_eq!( + built.max_effective_time_unix_nano, + base + 1_002, + "span end is the last ERROR record's observed instant", + ); + // The query predicate and the reference corpus both see the + // full row set — nothing was dropped as out-of-window. + assert_eq!( + built.query_severity, + Some(("ERROR".to_string(), 3)), + "the B1 predicate is unaffected by the timestamp source", + ); + assert_eq!( + built + .reference + .count_lines_containing("ERROR") + .expect("reference grep"), + 3, + "observed-only rows are spooled into the reference", + ); + } + /// B1 store over an OTLP corpus with a real severity mix: the /// "ERROR" text is preferred for the query predicate, its row /// count is exact, the severity distribution is visible (the @@ -517,7 +629,7 @@ mod tests { let built = build_b1_store(corpus.path(), bucket.path(), 3).expect("build"); assert_eq!(built.rows, 8); - assert_eq!(built.zero_ts_rows, 0); + assert_eq!(built.zero_effective_ts_rows, 0); assert_eq!(built.distinct_severities, 2, "INFO + ERROR"); assert_eq!( built.query_severity, @@ -532,7 +644,7 @@ mod tests { 3, "every ERROR record's reference line carries the token", ); - assert_eq!(built.min_time_unix_nano, base, "span start"); + assert_eq!(built.min_effective_time_unix_nano, base, "span start"); } /// Without a literal "ERROR" text, the busiest error-band @@ -589,7 +701,7 @@ mod tests { /// corpora carrying them), so they must not leak into the /// reference corpus either. #[test] - fn b1_reference_excludes_zero_ts_rows() { + fn b1_reference_excludes_zero_effective_ts_rows() { let corpus = tempfile::TempDir::new().expect("corpus dir"); let base = crate::corpus::TIME_BASELINE_NS; let jsonl = format!( @@ -602,7 +714,7 @@ mod tests { let built = build_b1_store(corpus.path(), bucket.path(), 3).expect("build"); - assert_eq!(built.zero_ts_rows, 1); + assert_eq!(built.zero_effective_ts_rows, 1); assert_eq!( built .reference @@ -662,7 +774,13 @@ mod tests { let built = build_query_store(corpus.path(), bucket.path()).expect("build"); assert_eq!(built.rows, 1, "the one record is written"); - assert_eq!(built.min_time_unix_nano, 0, "no non-zero timestamp → 0"); - assert_eq!(built.max_time_unix_nano, 0, "no non-zero timestamp → 0"); + assert_eq!( + built.min_effective_time_unix_nano, 0, + "no non-zero timestamp → 0" + ); + assert_eq!( + built.max_effective_time_unix_nano, 0, + "no non-zero timestamp → 0" + ); } } diff --git a/crates/ourios-parquet/src/lib.rs b/crates/ourios-parquet/src/lib.rs index 474bc94d5..cd1f17ff0 100644 --- a/crates/ourios-parquet/src/lib.rs +++ b/crates/ourios-parquet/src/lib.rs @@ -46,8 +46,8 @@ pub use compaction::{ }; pub use manifest::{MANIFEST_FILENAME, Manifest, ManifestError}; pub use partition::{ - PartitionKey, TimestampOverflowError, hour_partition_in_window, percent_decode_tenant, - percent_encode_tenant, + PartitionKey, TimestampOverflowError, effective_time_unix_nano, hour_partition_in_window, + percent_decode_tenant, percent_encode_tenant, }; pub use reader::{Reader, ReaderError}; pub use record_batch::{BatchError, mined_records_to_batch}; @@ -66,6 +66,7 @@ pub mod columns { pub const TEMPLATE_VERSION: &str = "template_version"; pub const TIME_UNIX_NANO: &str = "time_unix_nano"; pub const OBSERVED_TIME_UNIX_NANO: &str = "observed_time_unix_nano"; + pub const EFFECTIVE_TIME_UNIX_NANO: &str = "effective_time_unix_nano"; pub const SEVERITY_NUMBER: &str = "severity_number"; pub const SEVERITY_TEXT: &str = "severity_text"; pub const SCOPE_NAME: &str = "scope_name"; @@ -144,6 +145,15 @@ pub fn data_schema() -> SchemaRef { ), Field::new( columns::OBSERVED_TIME_UNIX_NANO, + DataType::Timestamp(TimeUnit::Nanosecond, Some(utc.clone())), + true, + ), + // OPTIONAL per §3.8 rule 1 (additive amendment 2026-06-11); + // the writer always populates it, NULL appears only in + // pre-amendment files (the §3.9 rule-2 read default is the + // row's `time_unix_nano`, not `None`). + Field::new( + columns::EFFECTIVE_TIME_UNIX_NANO, DataType::Timestamp(TimeUnit::Nanosecond, Some(utc)), true, ), diff --git a/crates/ourios-parquet/src/partition.rs b/crates/ourios-parquet/src/partition.rs index bc88943ab..2c7c3f697 100644 --- a/crates/ourios-parquet/src/partition.rs +++ b/crates/ourios-parquet/src/partition.rs @@ -76,7 +76,7 @@ impl PartitionKey { /// timestamp exceeds `i64::MAX` — the writer-rejects-overflow /// contract from RFC 0005 §3.2. pub fn derive(record: &MinedRecord) -> Result { - let chosen = choose_partition_timestamp(record)?; + let chosen = effective_time_unix_nano(record)?; // chosen is already i64-safe (checked above); `chrono:: // DateTime::from_timestamp_nanos` accepts i64. let dt = DateTime::::from_timestamp_nanos(chosen); @@ -182,12 +182,23 @@ fn hour_span_ns(year: i32, month: u32, day: u32, hour: u32) -> Option<(u64, u64) Some((lo, lo.saturating_add(HOUR_NANOS))) } -/// Choose the nanosecond timestamp for partition derivation per -/// §3.4: prefer `time_unix_nano` if non-zero, else -/// `observed_time_unix_nano` if non-zero, else the 1970 epoch -/// (returned as `0_i64`). Each candidate is checked against the -/// `u64`→`i64` overflow contract before being adopted. -fn choose_partition_timestamp(record: &MinedRecord) -> Result { +/// The RFC 0005 §3.2 **effective timestamp** (amendment +/// 2026-06-11): `time_unix_nano` if non-zero, else +/// `observed_time_unix_nano` if non-zero, else `0` (the 1970 +/// epoch). Each candidate is checked against the §3.2 `u64`→`i64` +/// overflow contract before being adopted. +/// +/// This single function feeds both the §3.4 partition derivation +/// ([`PartitionKey::derive`]) and the stored +/// `effective_time_unix_nano` column (the record-batch writer), so +/// the partition tuple and the column can never disagree — the +/// §3.4 "never disagree" rule is structural, not a convention. +/// +/// # Errors +/// +/// Returns [`TimestampOverflowError`] if the chosen nanosecond +/// timestamp exceeds `i64::MAX`. +pub fn effective_time_unix_nano(record: &MinedRecord) -> Result { if record.time_unix_nano != 0 { return i64::try_from(record.time_unix_nano).map_err(|_| TimestampOverflowError { field: "time_unix_nano", diff --git a/crates/ourios-parquet/src/reader.rs b/crates/ourios-parquet/src/reader.rs index 00e75c383..990c2486f 100644 --- a/crates/ourios-parquet/src/reader.rs +++ b/crates/ourios-parquet/src/reader.rs @@ -1054,6 +1054,13 @@ mod tests { b.append_null(); Arc::new(b.finish()) } + "effective_time_unix_nano" => { + // Equals the row's non-zero `time_unix_nano` + // (the §3.2 derivation with observed = None). + let mut b = TimestampNanosecondBuilder::new().with_timezone("UTC"); + b.append_value(1_775_127_480_000_000_000); + Arc::new(b.finish()) + } "severity_number" => { let mut b = UInt8Builder::new(); b.append_value(9); diff --git a/crates/ourios-parquet/src/record_batch.rs b/crates/ourios-parquet/src/record_batch.rs index 9ecf1c3b5..bc8b03398 100644 --- a/crates/ourios-parquet/src/record_batch.rs +++ b/crates/ourios-parquet/src/record_batch.rs @@ -36,7 +36,7 @@ use ourios_core::audit::ParamType; use ourios_core::otlp::KeyValue; use ourios_core::record::{BodyKind, MinedRecord}; -use crate::partition::TimestampOverflowError; +use crate::partition::{TimestampOverflowError, effective_time_unix_nano}; use crate::{columns, data_schema}; /// Build an Arrow `RecordBatch` matching `data_schema()` from a @@ -208,6 +208,7 @@ struct Builders { template_version: UInt32Builder, time_unix_nano: TimestampNanosecondBuilder, observed_time_unix_nano: TimestampNanosecondBuilder, + effective_time_unix_nano: TimestampNanosecondBuilder, severity_number: UInt8Builder, severity_text: StringBuilder, scope_name: StringBuilder, @@ -246,6 +247,8 @@ impl Builders { time_unix_nano: TimestampNanosecondBuilder::with_capacity(cap).with_timezone("UTC"), observed_time_unix_nano: TimestampNanosecondBuilder::with_capacity(cap) .with_timezone("UTC"), + effective_time_unix_nano: TimestampNanosecondBuilder::with_capacity(cap) + .with_timezone("UTC"), severity_number: UInt8Builder::with_capacity(cap), severity_text: StringBuilder::with_capacity(cap, 0), scope_name: StringBuilder::with_capacity(cap, 0), @@ -302,6 +305,13 @@ impl Builders { None => self.observed_time_unix_nano.append_null(), } + // Derived via the same function the §3.4 partition tuple + // uses (RFC 0005 §3.2 amendment 2026-06-11), so the stored + // column and the partition bucket can never disagree. Always + // populated — NULL appears only in pre-amendment files. + self.effective_time_unix_nano + .append_value(effective_time_unix_nano(r)?); + self.severity_number.append_value(r.severity_number); append_option_str(&mut self.severity_text, r.severity_text.as_deref()); append_option_str(&mut self.scope_name, r.scope_name.as_deref()); @@ -393,6 +403,7 @@ impl Builders { Arc::new(self.template_version.finish()), Arc::new(self.time_unix_nano.finish()), Arc::new(self.observed_time_unix_nano.finish()), + Arc::new(self.effective_time_unix_nano.finish()), Arc::new(self.severity_number.finish()), Arc::new(self.severity_text.finish()), Arc::new(self.scope_name.finish()), diff --git a/crates/ourios-parquet/src/writer.rs b/crates/ourios-parquet/src/writer.rs index ffa3c8fdc..985a91831 100644 --- a/crates/ourios-parquet/src/writer.rs +++ b/crates/ourios-parquet/src/writer.rs @@ -635,10 +635,11 @@ fn writer_properties(zstd: ZstdLevel) -> WriterProperties { // (`attributes`, `trace_id`, `span_id`) or non-text numeric // columns where dict-encoding adds overhead without payoff // (`time_unix_nano`, `observed_time_unix_nano`, - // `confidence`). + // `effective_time_unix_nano`, `confidence`). for no_dict_col in [ crate::columns::TIME_UNIX_NANO, crate::columns::OBSERVED_TIME_UNIX_NANO, + crate::columns::EFFECTIVE_TIME_UNIX_NANO, crate::columns::ATTRIBUTES, crate::columns::TRACE_ID, crate::columns::SPAN_ID, diff --git a/crates/ourios-parquet/tests/effective_timestamp.rs b/crates/ourios-parquet/tests/effective_timestamp.rs new file mode 100644 index 000000000..e14c8715f --- /dev/null +++ b/crates/ourios-parquet/tests/effective_timestamp.rs @@ -0,0 +1,171 @@ +//! Scenario RFC0005.13 (storage half) — effective-timestamp +//! fallback, amendment 2026-06-11. +//! See `docs/rfcs/0005-parquet-storage.md` §3.2 / §5. +//! +//! A record with `time_unix_nano = 0` and a non-zero +//! `observed_time_unix_nano = T`: +//! +//! - stores `effective_time_unix_nano = T` (writer-derived); +//! - lands under the partition tuple derived from `T` (§3.4 — the +//! partition tuple and the stored column never disagree); +//! - keeps the wire `time_unix_nano = 0` verbatim (RFC0001.10 — +//! derived, never overwriting). +//! +//! The query half (the RFC 0002 §6.2 window over the column, plus +//! the §3.9 pre-amendment-file fallback) lives in +//! `crates/ourios-querier/tests/rfc0005_13.rs`. + +use std::fs::File; +use std::path::{Path, PathBuf}; + +use arrow_array::Array; +use arrow_array::cast::AsArray; +use arrow_array::types::TimestampNanosecondType; +use ourios_core::record::{BodyKind, MinedRecord}; +use ourios_core::tenant::TenantId; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + +use ourios_parquet::{PartitionKey, Writer, columns, effective_time_unix_nano}; + +/// 2026-04-02T10:58:00 UTC (hour=10) — same anchor as the other +/// storage tests. +const T: u64 = 1_775_127_480_000_000_000; + +fn rec(time_unix_nano: u64, observed: Option) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("a"), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: None, + scope_name: None, + scope_version: None, + time_unix_nano, + observed_time_unix_nano: observed, + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0, + event_name: None, + body_kind: BodyKind::String, + params: Vec::new(), + separators: vec![String::new()], + body: None, + confidence: 1.0, + lossy_flag: false, + } +} + +/// Write `record` through the production `Writer` and return the +/// single emitted `*.parquet` path. +fn write_one(bucket: &Path, record: &MinedRecord) -> PathBuf { + let part = PartitionKey::derive(record).expect("derive partition"); + let mut w = Writer::open(bucket, part.clone()).expect("open writer"); + w.append_records(std::slice::from_ref(record)) + .expect("append"); + w.close().expect("close"); + + let dir = part.data_path(bucket); + let mut parquets: Vec = std::fs::read_dir(&dir) + .expect("read partition dir") + .map(|e| e.expect("dir entry").path()) + .filter(|p| p.extension().is_some_and(|x| x == "parquet")) + .collect(); + assert_eq!(parquets.len(), 1, "one file per flush"); + parquets.pop().expect("one parquet file") +} + +/// The single row's `(time_unix_nano, effective_time_unix_nano)` +/// as stored, read raw through the `parquet` crate (not the +/// project reader — `MinedRecord` deliberately has no effective +/// field, so the column is only visible at the Parquet level). +fn stored_timestamps(file: &Path) -> (i64, Option) { + let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(file).expect("open file")) + .expect("parquet builder") + .build() + .expect("parquet reader"); + let batches: Vec<_> = reader.collect::>().expect("read batches"); + assert_eq!(batches.len(), 1, "single tiny row group"); + let batch = &batches[0]; + assert_eq!(batch.num_rows(), 1, "single row"); + + let time_idx = batch + .schema() + .index_of(columns::TIME_UNIX_NANO) + .expect("time column present"); + let time = batch + .column(time_idx) + .as_primitive::() + .value(0); + + let eff_idx = batch + .schema() + .index_of(columns::EFFECTIVE_TIME_UNIX_NANO) + .expect("effective column present in post-amendment files"); + let eff_col = batch + .column(eff_idx) + .as_primitive::(); + let eff = (!eff_col.is_null(0)).then(|| eff_col.value(0)); + + (time, eff) +} + +/// RFC0005.13 — observed-only record: the stored effective column +/// equals `T`, the partition tuple derives from `T`, and the wire +/// `time_unix_nano` stays `0` verbatim. +#[test] +fn rfc0005_13_observed_only_record_stores_effective_and_keeps_wire_zero() { + // Arrange + let bucket = tempfile::TempDir::new().expect("temp"); + let record = rec(0, Some(T)); + + // Act + let file = write_one(bucket.path(), &record); + + // Assert — partition tuple derived from T (2026-04-02T10 UTC), + // not the 1970 epoch the zero wire value would land under. + let part = PartitionKey::derive(&record).expect("derive"); + assert_eq!( + (part.year, part.month, part.day, part.hour), + (2026, 4, 2, 10), + "partition derives from the observed fallback", + ); + + // Assert — stored column: wire zero verbatim, effective = T. + let (time, eff) = stored_timestamps(&file); + assert_eq!(time, 0, "wire time_unix_nano is never overwritten"); + assert_eq!( + eff, + Some(i64::try_from(T).expect("T fits i64")), + "effective_time_unix_nano stores the observed fallback", + ); +} + +/// The stored column always equals what the §3.4 partition +/// derivation chose — both run [`effective_time_unix_nano`], so +/// asserting the stored value against the shared function pins the +/// "never disagree" rule across all three derivation shapes. +#[test] +fn stored_effective_always_matches_the_partition_derivation() { + for record in [rec(T, Some(T + 1_000)), rec(0, Some(T)), rec(0, None)] { + // Arrange + let bucket = tempfile::TempDir::new().expect("temp"); + let expected = effective_time_unix_nano(&record).expect("derive effective"); + + // Act + let file = write_one(bucket.path(), &record); + + // Assert + let (_, eff) = stored_timestamps(&file); + assert_eq!( + eff, + Some(expected), + "stored column equals the shared derivation for \ + time={} observed={:?}", + record.time_unix_nano, + record.observed_time_unix_nano, + ); + } +} diff --git a/crates/ourios-parquet/tests/schema_pin.rs b/crates/ourios-parquet/tests/schema_pin.rs index 504e3e41a..30124c1e6 100644 --- a/crates/ourios-parquet/tests/schema_pin.rs +++ b/crates/ourios-parquet/tests/schema_pin.rs @@ -114,6 +114,13 @@ fn rfc0005_10_data_schema_matches_pinned_field_list() { DataType::Timestamp(TimeUnit::Nanosecond, Some(utc())), true, ), + // OPTIONAL, writer-derived (RFC 0005 §3.2 amendment + // 2026-06-11); absent only in pre-amendment files. + Field::new( + "effective_time_unix_nano", + DataType::Timestamp(TimeUnit::Nanosecond, Some(utc())), + true, + ), Field::new("severity_number", DataType::UInt8, false), Field::new("severity_text", DataType::Utf8, true), Field::new("scope_name", DataType::Utf8, true), diff --git a/crates/ourios-querier/src/compile.rs b/crates/ourios-querier/src/compile.rs index b8a1f2c90..e98dd860c 100644 --- a/crates/ourios-querier/src/compile.rs +++ b/crates/ourios-querier/src/compile.rs @@ -25,6 +25,12 @@ //! | `lossy` | `lossy_flag` | `Boolean` | //! | `flags` | `flags` | `UInt32` | //! +//! The `range(...)` time window is **not** the bare `ts` field: it compiles +//! against the derived `effective_time_unix_nano` column (RFC 0002 §6.2 +//! amendment 2026-06-11) via [`crate::time_window_filter`], with the +//! RFC 0005 §3.9 `effective := time_unix_nano` fallback for files that +//! predate the column. +//! //! `service`, `resource.`, and `attr.` have **no dedicated column** in //! the RFC 0005 schema: resource/log attributes are stored as a single //! OTLP-canonical-JSON `Utf8` column (`resource_attributes` / `attributes`). @@ -173,13 +179,11 @@ pub(crate) fn apply(df: DataFrame, plan: Plan) -> Result, Quer alias_classes, limit, } = plan; - let mut df = df - .filter( - col(columns::TIME_UNIX_NANO) - .gt_eq(lit(time_bound_scalar(start)?)) - .and(col(columns::TIME_UNIX_NANO).lt(lit(time_bound_scalar(end)?))), - ) - .map_err(crate::storage_err)?; + // The window filters the *effective* timestamp (RFC 0002 §6.2 amendment + // 2026-06-11), with the RFC 0005 §3.9 fallback for pre-amendment files; + // the bare `ts` field stays `time_unix_nano`, the verbatim wire value. + let window_filter = crate::time_window_filter(&df, start, end)?; + let mut df = df.filter(window_filter).map_err(crate::storage_err)?; match compile_predicate(&predicate, &df, &alias_classes)? { // `true` ⇒ match-all ⇒ no predicate filter (window only). diff --git a/crates/ourios-querier/src/lib.rs b/crates/ourios-querier/src/lib.rs index f9f0b2143..630d49b9e 100644 --- a/crates/ourios-querier/src/lib.rs +++ b/crates/ourios-querier/src/lib.rs @@ -78,7 +78,9 @@ pub struct QueryRequest { /// structurally — the querier only ever reads under this /// tenant's partition directory (`CLAUDE.md` §3.7; RFC0007.5). pub tenant: TenantId, - /// Optional `[start, end)` `time_unix_nano` bounds. + /// Optional `[start, end)` bounds over the **effective** timestamp + /// (`effective_time_unix_nano`, falling back to `time_unix_nano` for + /// pre-amendment files — RFC 0005 §3.2 / §3.9, amendment 2026-06-11). pub time_range: Option<(u64, u64)>, /// Optional template-exact filter (B2 — `template_id` equality). pub template_id: Option, @@ -347,6 +349,47 @@ fn has_column(df: &datafusion::dataframe::DataFrame, column: &str) -> bool { df.schema().fields().iter().any(|f| f.name() == column) } +/// The row-level time-window filter `[start, end)` over the **effective** +/// timestamp (RFC 0002 §6.2 / RFC 0005 §3.2, amendment 2026-06-11), with the +/// §3.9 rule-2 carve-out for files that predate the +/// `effective_time_unix_nano` column. Shared by the `QueryRequest` path and +/// the DSL compiler so both windows have identical semantics. +/// +/// The carve-out is the explicit exception to the +/// absent-OPTIONAL-column ⇒ predicate-false convention (RFC0007.4): for +/// pre-amendment files the window applies `effective := time_unix_nano` — +/// exactly the pre-amendment behaviour — because compiling the window to +/// `false` would silently hide every old file from every query. +/// +/// - Column absent from the (post-union) schema ⇒ every file predates the +/// amendment ⇒ filter `time_unix_nano` directly (prunable, as before). +/// - Column present ⇒ a *mixed* scan is still possible: `DataFusion` fills +/// the column with NULL for files that lack it, and NULL fails both window +/// comparisons — the forbidden silent-hiding outcome. Post-amendment +/// writers always populate the column (§3.2: NULL appears only in +/// pre-amendment files), so `IS NULL` identifies exactly the rows needing +/// the `time_unix_nano` fallback. The `OR` shape (rather than a +/// `coalesce`) keeps the predicate inside `DataFusion`'s pruning grammar: +/// min/max statistics prune the effective branch and null counts collapse +/// the fallback branch on post-amendment row groups — the B1 mechanism +/// (RFC 0005 §3.2 rule 3). +fn time_window_filter( + df: &datafusion::dataframe::DataFrame, + start: u64, + end: u64, +) -> Result { + let lo = lit(time_bound_scalar(start)?); + let hi = lit(time_bound_scalar(end)?); + let ts = || col(columns::TIME_UNIX_NANO); + let ts_window = ts().gt_eq(lo.clone()).and(ts().lt(hi.clone())); + if !has_column(df, columns::EFFECTIVE_TIME_UNIX_NANO) { + return Ok(ts_window); + } + let eff = || col(columns::EFFECTIVE_TIME_UNIX_NANO); + let eff_window = eff().gt_eq(lo).and(eff().lt(hi)); + Ok(eff_window.or(eff().is_null().and(ts_window))) +} + /// Apply the [`QueryRequest`] predicate set as `DataFusion` filters. Returns /// `Ok(None)` when a `severity_text` filter targets an absent OPTIONAL column /// (provably empty — short-circuit). @@ -355,13 +398,8 @@ fn apply_request_filters( request: &QueryRequest, ) -> Result, QueryError> { if let Some((start, end)) = request.time_range { - df = df - .filter( - col(columns::TIME_UNIX_NANO) - .gt_eq(lit(time_bound_scalar(start)?)) - .and(col(columns::TIME_UNIX_NANO).lt(lit(time_bound_scalar(end)?))), - ) - .map_err(storage_err)?; + let window = time_window_filter(&df, start, end)?; + df = df.filter(window).map_err(storage_err)?; } if let Some(template_id) = request.template_id { df = df diff --git a/crates/ourios-querier/tests/rfc0005_13.rs b/crates/ourios-querier/tests/rfc0005_13.rs new file mode 100644 index 000000000..7aa4e87f5 --- /dev/null +++ b/crates/ourios-querier/tests/rfc0005_13.rs @@ -0,0 +1,236 @@ +//! Scenario RFC0005.13 (query half) — the time window filters the +//! effective timestamp, with the §3.9 pre-amendment fallback. +//! See `docs/rfcs/0005-parquet-storage.md` §3.2 / §3.9 / §5 and +//! `docs/rfcs/0002-query-dsl.md` §6.2 (amendment 2026-06-11). +//! +//! Three obligations, each through the real querier path: +//! +//! 1. A record with `time_unix_nano = 0` and +//! `observed_time_unix_nano = T` is returned by a `range(...)` +//! window containing `T` — observed-only records are addressable +//! by time (the B1 unblock). +//! 2. A **pre-amendment** file (no `effective_time_unix_nano` +//! column, built with the raw `ArrowWriter` per the RFC0007.4 +//! pattern) answers the same window as `effective := +//! time_unix_nano` — old rows are still found, alone *and* mixed +//! with post-amendment files (where `DataFusion`'s schema union +//! fills the column with NULL, which would otherwise fail both +//! window bounds and silently hide the file — the §3.9-forbidden +//! outcome). +//! 3. The window stays *prunable* on the stored column: one file +//! with two row groups in one hour partition (so neither the +//! directory prune nor plan-time file pruning can hide the skip), +//! and a sub-hour window skips the out-of-window row group via +//! statistics (RFC 0005 §3.2 rule 3 — the B1 mechanism). +//! +//! The storage half (stored column value, partition tuple, verbatim +//! wire zero) lives in +//! `crates/ourios-parquet/tests/effective_timestamp.rs`. + +mod common; + +use std::fs::File; +use std::path::Path; + +use arrow_array::RecordBatch; +use ourios_core::record::MinedRecord; +use ourios_core::tenant::TenantId; +use parquet::arrow::ArrowWriter; + +use common::{DEFAULT_WINDOW_NS, HOUR_NS, NOW, TS0, no_aliases, simple, write_all}; +use ourios_parquet::{PartitionKey, columns, mined_records_to_batch}; +use ourios_querier::Querier; + +/// Render nanoseconds-since-epoch as the RFC 3339 instant the DSL +/// `time` grammar takes (nanosecond precision, so bounds are exact). +fn rfc3339(ns: u64) -> String { + chrono::DateTime::::from_timestamp_nanos(i64::try_from(ns).expect("fits i64")) + .to_rfc3339_opts(chrono::SecondsFormat::Nanos, true) +} + +/// Run `range(, )` (half-open, nanosecond bounds) for tenant +/// "a" and return the matching row count. +async fn rows_in_window(bucket: &Path, lo: u64, hi: u64) -> u64 { + let query = + ourios_querier::dsl::parse(&format!("true | range({}, {})", rfc3339(lo), rfc3339(hi))) + .expect("parse"); + Querier::new(bucket) + .run_query( + &query, + &TenantId::new("a"), + NOW, + DEFAULT_WINDOW_NS, + &no_aliases(), + ) + .await + .expect("run_query") + .rows +} + +/// A `simple` fixture row reshaped to the RFC0005.13 trigger: wire +/// `time_unix_nano = 0`, `observed_time_unix_nano = Some(ts_ns)`. +fn observed_only(ts_ns: u64) -> MinedRecord { + MinedRecord { + time_unix_nano: 0, + observed_time_unix_nano: Some(ts_ns), + ..simple("a", 1, 0) + } +} + +/// Write `record` as a committed `*.parquet` in **pre-amendment** +/// shape: the writer's batch with the `effective_time_unix_nano` +/// column projected away, laid down with the raw `ArrowWriter` at +/// the record's RFC 0005 partition directory (the RFC0007.4 / +/// RFC0005.2 old-writer pattern). +fn write_pre_amendment(bucket: &Path, record: &MinedRecord) { + let base = mined_records_to_batch(std::slice::from_ref(record)).expect("base batch"); + let keep: Vec = base + .schema() + .fields() + .iter() + .enumerate() + .filter(|(_, f)| f.name() != columns::EFFECTIVE_TIME_UNIX_NANO) + .map(|(i, _)| i) + .collect(); + let old: RecordBatch = base.project(&keep).expect("project out effective column"); + + let dir = PartitionKey::derive(record) + .expect("derive partition") + .data_path(bucket); + std::fs::create_dir_all(&dir).expect("mkdir partition"); + let file = File::create(dir.join("pre_amendment.parquet")).expect("create parquet"); + let mut w = ArrowWriter::try_new(file, old.schema(), None).expect("arrow writer"); + w.write(&old).expect("write batch"); + w.close().expect("close writer"); +} + +/// RFC0005.13 — a `range(...)` window containing `T` returns the +/// observed-only record (`time_unix_nano = 0`, +/// `observed_time_unix_nano = T`), and a window over the epoch (where +/// the zero wire value would sit) does NOT — the window filters the +/// effective timestamp, not the wire one. +#[tokio::test] +async fn rfc0005_13_window_returns_observed_only_record() { + // Arrange + let bucket = tempfile::TempDir::new().expect("temp"); + write_all(bucket.path(), &[observed_only(TS0)]); + + // Act / Assert — a window around T finds the row. + assert_eq!( + rows_in_window(bucket.path(), TS0 - 1_000, TS0 + 1_000).await, + 1, + "the observed-only record is addressable by time via its effective timestamp", + ); + // A window over the epoch (covering the wire `0`) finds nothing: + // the effective value replaced the zero for windowing purposes. + assert_eq!( + rows_in_window(bucket.path(), 0, 1_000).await, + 0, + "the wire zero is not what the window filters", + ); +} + +/// RFC0005.13 (second half) — a pre-amendment file (no +/// `effective_time_unix_nano` column) answers a time window as +/// `effective := time_unix_nano` (RFC 0005 §3.9 rule 2): the old row +/// is found both when the file stands alone and when it is mixed with +/// a post-amendment file in the same scan (the schema-union NULL +/// case), and an out-of-window query still excludes it. +#[tokio::test] +async fn rfc0005_13_pre_amendment_file_windows_on_time_unix_nano() { + // Arrange — hour 10: a pre-amendment file alone in the store. + let bucket = tempfile::TempDir::new().expect("temp"); + write_pre_amendment(bucket.path(), &simple("a", 1, TS0)); + + // Act / Assert — alone: the union schema lacks the column, the + // window compiles over `time_unix_nano` exactly as before. + assert_eq!( + rows_in_window(bucket.path(), TS0 - 1_000, TS0 + 1_000).await, + 1, + "a pre-amendment-only store answers the window unchanged", + ); + assert_eq!( + rows_in_window(bucket.path(), TS0 + 1_000, TS0 + 2_000).await, + 0, + "out-of-window rows in a pre-amendment file stay excluded", + ); + + // Arrange — hour 11: add a post-amendment file, making the scan + // mixed: the union schema now has the column and the + // pre-amendment file's rows read it as NULL. + write_all(bucket.path(), &[simple("a", 2, TS0 + HOUR_NS)]); + + // Act / Assert — a window covering both hours returns both rows; + // the NULL-filled old row is NOT silently hidden (§3.9 rule 2's + // explicit carve-out from absent-OPTIONAL ⇒ predicate-false). + assert_eq!( + rows_in_window(bucket.path(), TS0 - 1_000, TS0 + HOUR_NS + 1_000).await, + 2, + "a mixed scan returns pre- and post-amendment rows alike", + ); + // And a window covering only the old row still finds exactly it. + assert_eq!( + rows_in_window(bucket.path(), TS0 - 1_000, TS0 + 1_000).await, + 1, + "the pre-amendment row is individually addressable in a mixed scan", + ); +} + +/// The effective-timestamp window still prunes row groups via the +/// stored column's statistics (RFC 0005 §3.2 rule 3 — a real column, +/// not a query-time fallback expression, is what keeps B1's pruning +/// mechanism alive). One file with two row groups (so neither the +/// directory-level prune nor `DataFusion`'s plan-time file-level +/// prune can hide the skip from the row-group metrics): a sub-hour +/// window must skip the out-of-window row group by min/max +/// statistics. +#[tokio::test] +async fn rfc0005_13_effective_window_prunes_row_groups() { + // Arrange — one hour-10 file holding two single-row row groups: + // one at the start of the hour, one 30 minutes in. Written with + // the raw `ArrowWriter` (the production writer rotates row + // groups by size, far above two rows) from the production + // batch, so the effective column and its statistics are real. + let bucket = tempfile::TempDir::new().expect("temp"); + let records = [observed_only(TS0), observed_only(TS0 + HOUR_NS / 2)]; + let batch = mined_records_to_batch(&records).expect("batch"); + let dir = PartitionKey::derive(&records[0]) + .expect("derive partition") + .data_path(bucket.path()); + std::fs::create_dir_all(&dir).expect("mkdir partition"); + let file = File::create(dir.join("two_row_groups.parquet")).expect("create parquet"); + let props = parquet::file::properties::WriterProperties::builder() + .set_max_row_group_size(1) + .build(); + let mut w = ArrowWriter::try_new(file, batch.schema(), Some(props)).expect("arrow writer"); + w.write(&batch).expect("write batch"); + w.close().expect("close writer"); + + let query = ourios_querier::dsl::parse(&format!( + "true | range({}, {})", + rfc3339(TS0 - 1_000), + rfc3339(TS0 + 1_000) + )) + .expect("parse"); + // Act — a window covering only the first file's instant. + let result = Querier::new(bucket.path()) + .run_query( + &query, + &TenantId::new("a"), + NOW, + DEFAULT_WINDOW_NS, + &no_aliases(), + ) + .await + .expect("run_query"); + + // Assert — only the in-window row matches, and the other file's + // row group was pruned by statistics, not scanned. + assert_eq!(result.rows, 1, "only the in-window row matches"); + assert!( + result.stats.row_groups_pruned >= 1, + "the out-of-window row group must be pruned via the effective \ + column's statistics; stats={:?}", + result.stats, + ); +}