From a827da028a71be0e8b995f25b350b27c17fc6a1e Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Sun, 12 Jul 2026 02:41:23 +0200 Subject: [PATCH] =?UTF-8?q?perf(querier):=20late=20materialization=20?= =?UTF-8?q?=E2=80=94=20page-selective=20reads=20on=20the=20materialize=20s?= =?UTF-8?q?can?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The RFC 0031 run #8 decomposition showed row materialization dominating per-query object-storage reads: rendering even ONE matching row re-read the hot partition's page-index-matched window of every column chunk whole (severity query, 1 row: 3,146,731 B materialize vs 609,498 B count scan on otel-demo-v8). Enable DataFusion's parquet filter pushdown (pushdown_filters + reorder_filters, off by default in DF 54) on the materialize scan in Querier::collect_records: the predicate now evaluates during Parquet decode, and the writer's offset index lets the reader fetch only the heavy-column (body / params / attributes) pages the selected rows live in. Results are bit-identical (the regression test pins the exact LineKey; the full querier + bench suites pass unmodified). Scoped to the materialize scan only. Enabled on the count scan too, pushdown lets statistics-fully-matched row groups answer with zero bytes scanned (correct rows, bytes_scanned=0) — a further win, but it conflicts with honest_total_bytes_breaks_down_additively's "count_scan_bytes > 0" assertion and would hollow out the stats.bytes_read semantics the B1/B2 gates assert on, so the count scan keeps the default (conflict reported, test untouched). Empirical basis (new deterministic 70,000-row / 3-hour-partition store, hot hour 60,000 rows, 1-row `severity >= 17 | limit 10` query, regression-pinned in ourios-bench): before: count_scan=148,392 materialize=1,664,654 registry=8,972 after: count_scan=148,392 materialize= 742,036 registry=8,972 2.24x fewer materialize bytes; the regression test asserts the component stays under a 1 MiB ceiling the whole-window behavior violates by ~1.6x while the page-selective scan clears it with ~1.4x headroom. (An earlier 20,000-row hot hour showed only 690,181 -> 623,049: parquet-rs's 20,000-row default page limit made each chunk a single page, so the fixture is sized past it — page-selective reads need >1 page per chunk to be observable.) Hazard §4.6: no DataFusion type crosses the public surface — the knob is internal to the scan session. Invariant §3.3 unaffected: rendering still goes through the read-time registry; the regression test asserts the returned body byte-for-byte. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y --- crates/ourios-bench/src/comparative.rs | 192 +++++++++++++++++++++++++ crates/ourios-querier/src/lib.rs | 27 ++++ 2 files changed, 219 insertions(+) diff --git a/crates/ourios-bench/src/comparative.rs b/crates/ourios-bench/src/comparative.rs index 02e51b52..5b75b4a9 100644 --- a/crates/ourios-bench/src/comparative.rs +++ b/crates/ourios-bench/src/comparative.rs @@ -1139,6 +1139,198 @@ mod tests { ); } + /// Hour width and shape of the late-materialization corpus, shared + /// between the generator and the test's expected-row computation. + const LM_HOUR_NS: u64 = 3_600_000_000_000; + const LM_HOT_ROWS: u64 = 60_000; + const LM_FILLER_ROWS: u64 = 5_000; + const LM_ERROR_INDEX: u64 = 13_337; + + /// The per-record pseudo-random ids (an LCG keeps the corpus + /// deterministic and dependency-free) whose entropy makes the heavy + /// columns compress poorly, the way real request logs do. + fn synthetic_ids(n: u64) -> (u64, u64) { + let a = n + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + let b = a.wrapping_mul(6_364_136_223_846_793_005) >> 32; + (a, b) + } + + fn synthetic_body(n: u64, error: bool) -> String { + let (a, b) = synthetic_ids(n); + let verb = if error { "failed" } else { "handled" }; + format!("{verb} request {a:016x} for user {b:08x} path /api/v1/orders/{n}") + } + + /// One `LogRecord` of the late-materialization corpus. + fn synthetic_record( + time_unix_nano: u64, + n: u64, + error: bool, + ) -> opentelemetry_proto::tonic::logs::v1::LogRecord { + use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value}; + use opentelemetry_proto::tonic::logs::v1::LogRecord; + + let (a, b) = synthetic_ids(n); + let (severity_number, severity_text) = if error { (17, "ERROR") } else { (9, "INFO") }; + let body = synthetic_body(n, error); + let attr = |key: &str, value: String| KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue(value)), + }), + ..KeyValue::default() + }; + LogRecord { + time_unix_nano, + severity_number, + severity_text: severity_text.to_string(), + body: Some(AnyValue { + value: Some(any_value::Value::StringValue(body)), + }), + attributes: vec![ + attr("http.request.id", format!("{a:016x}")), + attr("session.id", format!("{a:016x}{b:016x}")), + attr("url.path", format!("/api/v1/orders/{n}/items/{b:x}")), + attr( + "client.address", + format!("10.{}.{}.{}", a % 256, b % 256, n % 256), + ), + attr( + "user_agent.original", + format!("Mozilla/5.0 (build {b:08x}) ourios-bench/{}", n % 97), + ), + ], + ..LogRecord::default() + } + } + + /// The synthetic multi-hour store the late-materialization regression + /// measures: a hot hour of 60 000 rows holding exactly one ERROR row, + /// plus two 5 000-row INFO-only filler hours. Hour-aligned so the hot + /// hour lands in a single partition; sized past parquet-rs's 20 000-row + /// default page limit so each hot-hour column chunk spans several pages + /// (page-granular reads need >1 page per chunk to be observable). + fn late_materialization_corpus() -> String { + use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value}; + use opentelemetry_proto::tonic::logs::v1::{ResourceLogs, ScopeLogs}; + use opentelemetry_proto::tonic::resource::v1::Resource; + + let base_ns = (crate::corpus::TIME_BASELINE_NS / LM_HOUR_NS) * LM_HOUR_NS; + + let mut lines = Vec::new(); + for (hour, rows) in [ + (0u64, LM_HOT_ROWS), + (1, LM_FILLER_ROWS), + (2, LM_FILLER_ROWS), + ] { + let hour_base = base_ns + hour * LM_HOUR_NS; + let spacing = LM_HOUR_NS / (rows + 1); + let log_records = (0..rows) + .map(|i| { + let error = hour == 0 && i == LM_ERROR_INDEX; + synthetic_record(hour_base + i * spacing, hour * 1_000_000 + i, error) + }) + .collect(); + let logs = LogsData { + resource_logs: vec![ResourceLogs { + resource: Some(Resource { + attributes: vec![KeyValue { + key: "service.name".to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue( + FIXTURE_SERVICE.to_string(), + )), + }), + ..KeyValue::default() + }], + ..Resource::default() + }), + scope_logs: vec![ScopeLogs { + log_records, + ..ScopeLogs::default() + }], + ..ResourceLogs::default() + }], + }; + lines.push(serde_json::to_string(&logs).expect("synthetic LogsData serializes")); + } + lines.join("\n") + } + + /// The late-materialization regression guard (RFC 0031 run #8 + /// decomposition): rendering even ONE matching row used to re-read the + /// hot partition's page-index-matched window of every column chunk + /// whole — with filter pushdown on the querier's scans, the + /// materialize pass fetches only the pages the selected row needs. + /// + /// Empirical basis for the ceiling (this exact store, measured + /// 2026-07-12, `DataFusion` 54 / parquet 58): the pre-pushdown + /// whole-window materialize scan read 1 664 654 B; the page-selective + /// scan reads 742 036 B. The 1 MiB ceiling sits between with + /// comfortable margin both ways — the old behavior violates it by + /// ~1.6×, the new behavior clears it with ~1.4× headroom. The store is + /// fully deterministic (LCG corpus, fixed timestamps), so drift toward + /// the ceiling means a real regression in scan selectivity, not noise. + #[test] + fn one_row_materialization_reads_pages_not_whole_chunks() { + const MATERIALIZE_CEILING_BYTES: u64 = 1_048_576; + + let corpus = tempfile::TempDir::new().expect("corpus dir"); + std::fs::write( + corpus.path().join("synthetic.jsonl"), + late_materialization_corpus(), + ) + .expect("write corpus"); + let bucket = tempfile::TempDir::new().expect("bucket dir"); + let built = + crate::build_comparative_store(corpus.path(), bucket.path(), crate::TxtSeverity::Fixed) + .expect("build store"); + assert_eq!( + built.rows, + LM_HOT_ROWS + 2 * LM_FILLER_ROWS, + "the whole synthetic corpus is stored", + ); + + let tenant = TenantId::new(built.tenant); + let now = built.max_effective_time_unix_nano + 1; + let window = built.max_effective_time_unix_nano - built.min_effective_time_unix_nano + 2; + let answer = ourios_query_answer( + bucket.path(), + &tenant, + "severity >= 17 | limit 10", + now, + window, + ) + .expect("query answer"); + + // Correctness first: pushdown must not change the answer. The one + // ERROR row comes back with its exact timestamp and bit-identical + // reconstructed body. + let base_ns = (crate::corpus::TIME_BASELINE_NS / LM_HOUR_NS) * LM_HOUR_NS; + let expected = LineKey { + timestamp_unix_nanos: base_ns + LM_ERROR_INDEX * (LM_HOUR_NS / (LM_HOT_ROWS + 1)), + body: synthetic_body(LM_ERROR_INDEX, true).into_bytes(), + }; + assert_eq!( + answer.lines, + vec![expected], + "exactly the one ERROR row matches, rendered bit-identically", + ); + + assert!( + answer.materialize_bytes < MATERIALIZE_CEILING_BYTES, + "materializing 1 row must read pages, not the whole page-index \ + window of every chunk: materialize={} (ceiling {}, count_scan={}, \ + registry={})", + answer.materialize_bytes, + MATERIALIZE_CEILING_BYTES, + answer.count_scan_bytes, + answer.registry_bytes, + ); + } + #[test] fn body_preview_bounds_large_bodies() { // Small body is shown whole. diff --git a/crates/ourios-querier/src/lib.rs b/crates/ourios-querier/src/lib.rs index 5244e8b4..adcbb64f 100644 --- a/crates/ourios-querier/src/lib.rs +++ b/crates/ourios-querier/src/lib.rs @@ -1038,6 +1038,33 @@ impl Querier { tenant: &TenantId, task_ctx: Arc, ) -> Result { + // Filter pushdown ("late materialization", off by default in + // DataFusion 54): the scan evaluates the predicate during Parquet + // decode and — via the writer's offset index — fetches only the + // heavy-column (`body` / `params` / `attributes`) pages the selected + // rows live in, instead of the whole page-index-matched window of + // every unpruned chunk. This keeps the RFC 0031 §3.6 materialization + // component proportional to the result size, not the partition size + // (regression-pinned in `ourios-bench`'s + // `one_row_materialization_reads_pages_not_whole_chunks`); the + // session-level option reaches the directly-constructed + // `ParquetFormat` because `ParquetSource::try_pushdown_filters` + // honours table *or* session config. Scoped to this scan only: on + // the count scan, pushdown lets statistics-fully-matched row groups + // answer with **zero** bytes scanned, which would hollow out the + // `stats.bytes_read` figure the B1/B2 gates and the RFC 0031 honest + // total assert on — the count scan keeps the default. + let (state, logical_plan) = df.into_parts(); + let mut config = state.config().clone(); + { + let parquet = &mut config.options_mut().execution.parquet; + parquet.pushdown_filters = true; + parquet.reorder_filters = true; + } + let state = datafusion::execution::SessionStateBuilder::new_from_existing(state) + .with_config(config) + .build(); + let df = datafusion::dataframe::DataFrame::new(state, logical_plan); let limited = df.limit(0, Some(limit)).map_err(storage_err)?; // Plan + collect by hand (rather than `DataFrame::collect`) so this // scan's metrics can be read off the retained plan. The caller folds