feat(querier): partition-level time pruning (RFC 0007)

crates/ourios-parquet/src/lib.rs

@@ -46,7 +46,8 @@ pub use compaction::{
 };
 pub use manifest::{MANIFEST_FILENAME, Manifest, ManifestError};
 pub use partition::{
-    PartitionKey, TimestampOverflowError, percent_decode_tenant, percent_encode_tenant,
+    PartitionKey, TimestampOverflowError, hour_partition_in_window, percent_decode_tenant,
+    percent_encode_tenant,
 };
 pub use reader::{Reader, ReaderError};
 pub use record_batch::{BatchError, mined_records_to_batch};

crates/ourios-parquet/src/partition.rs

@@ -18,9 +18,12 @@
 use std::fmt;
 use std::path::{Path, PathBuf};
 
-use chrono::{DateTime, Datelike, Timelike, Utc};
+use chrono::{DateTime, Datelike, NaiveDate, Timelike, Utc};
 use ourios_core::record::MinedRecord;
 
+/// One hour in nanoseconds — the span a `…/hour=HH/` partition covers.
+const HOUR_NANOS: u64 = 3_600_000_000_000;
+
 /// Partition key for the on-disk Hive-style layout.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct PartitionKey {
@@ -126,6 +129,59 @@ impl PartitionKey {
     }
 }
 
+/// Whether the hour partition at `partition_dir` *could* hold a row in
+/// the half-open window `[start_ns, end_ns)` — i.e. its
+/// `[hour_start, hour_start + 1h)` UTC span overlaps the window.
+///
+/// `partition_dir` is a `…/year=YYYY/month=MM/day=DD/hour=HH` leaf. This
+/// is a **conservative** pruning predicate for the querier: it returns
+/// `true` (do *not* prune — read the partition) whenever the trailing
+/// Hive segments don't parse or the hour isn't a real UTC instant, so a
+/// query can never drop in-window data on an unrecognised layout. The
+/// row-level time filter stays the caller's column predicate; this only
+/// lets it skip footers that are *certain* to fall outside the window
+/// (RFC 0007's deferred partition-level time pruning).
+#[must_use]
+pub fn hour_partition_in_window(partition_dir: &Path, start_ns: u64, end_ns: u64) -> bool {
+    let Some((year, month, day, hour)) = parse_hour_partition(partition_dir) else {
+        return true;
+    };
+    let Some((lo, hi)) = hour_span_ns(year, month, day, hour) else {
+        return true;
+    };
+    // Half-open overlap: [lo, hi) ∩ [start, end) ≠ ∅.
+    lo < end_ns && start_ns < hi
+}
+
+/// Parse the `(year, month, day, hour)` from the trailing four Hive
+/// segments of a partition directory path. `None` if the deepest four
+/// components aren't `hour=`, `day=`, `month=`, `year=` with parseable
+/// numbers (e.g. a non-leaf dir or a foreign path).
+fn parse_hour_partition(dir: &Path) -> Option<(i32, u32, u32, u32)> {
+    let mut segments = dir.components().rev().filter_map(|c| match c {
+        std::path::Component::Normal(s) => s.to_str(),
+        _ => None,
+    });
+    let hour = segments.next()?.strip_prefix("hour=")?.parse().ok()?;
+    let day = segments.next()?.strip_prefix("day=")?.parse().ok()?;
+    let month = segments.next()?.strip_prefix("month=")?.parse().ok()?;
+    let year = segments.next()?.strip_prefix("year=")?.parse().ok()?;
+    Some((year, month, day, hour))
+}
+
+/// The `[start, end)` UTC-nanosecond span of the hour partition
+/// `(year, month, day, hour)`. `None` if it isn't a real UTC instant or
+/// predates the 1970 epoch (no `u64`-nanos row can land there), so the
+/// caller treats it as non-prunable.
+fn hour_span_ns(year: i32, month: u32, day: u32, hour: u32) -> Option<(u64, u64)> {
+    let start = NaiveDate::from_ymd_opt(year, month, day)?
+        .and_hms_opt(hour, 0, 0)?
+        .and_utc()
+        .timestamp_nanos_opt()?;
+    let lo = u64::try_from(start).ok()?;
+    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
@@ -423,6 +479,77 @@ mod tests {
         assert_eq!(path, expected);
     }
 
+    /// `hour=10` on 2026-04-02 covers [10:00, 11:00) UTC.
+    const HOUR10_START: u64 = 1_775_124_000_000_000_000; // 2026-04-02T10:00:00Z
+
+    fn hour10_dir() -> PathBuf {
+        [
+            "bucket",
+            "data",
+            "tenant_id=t",
+            "year=2026",
+            "month=04",
+            "day=02",
+            "hour=10",
+        ]
+        .iter()
+        .collect()
+    }
+
+    #[test]
+    fn hour_partition_in_window_overlap_cases() {
+        let dir = hour10_dir();
+        // A window fully inside the hour overlaps.
+        assert!(hour_partition_in_window(
+            &dir,
+            HOUR10_START + 60_000_000_000,
+            HOUR10_START + 120_000_000_000,
+        ));
+        // A window touching the hour's start (half-open, inclusive lo).
+        assert!(hour_partition_in_window(
+            &dir,
+            HOUR10_START,
+            HOUR10_START + 1
+        ));
+        // A window entirely before the hour does not overlap → prune.
+        assert!(!hour_partition_in_window(
+            &dir,
+            HOUR10_START - 120_000_000_000,
+            HOUR10_START - 60_000_000_000,
+        ));
+        // A window starting exactly at the hour's end is excluded
+        // (half-open upper bound) → prune.
+        assert!(!hour_partition_in_window(
+            &dir,
+            HOUR10_START + HOUR_NANOS,
+            HOUR10_START + HOUR_NANOS + 1,
+        ));
+    }
+
+    #[test]
+    fn hour_partition_in_window_is_conservative_on_unparseable_paths() {
+        // A non-leaf / foreign path can't be proven out of range, so it
+        // is never pruned (returns true) — pruning must not drop data.
+        let day_dir: PathBuf = [
+            "bucket",
+            "data",
+            "tenant_id=t",
+            "year=2026",
+            "month=04",
+            "day=02",
+        ]
+        .iter()
+        .collect();
+        assert!(hour_partition_in_window(&day_dir, 0, 1));
+        let foreign: PathBuf = ["some", "other", "dir"].iter().collect();
+        assert!(hour_partition_in_window(&foreign, 0, 1));
+        // A non-canonical hour value (not a real instant) → not pruned.
+        let bad_hour: PathBuf = ["year=2026", "month=04", "day=02", "hour=99"]
+            .iter()
+            .collect();
+        assert!(hour_partition_in_window(&bad_hour, 0, 1));
+    }
+
     #[test]
     fn audit_path_stops_at_day() {
         let key = PartitionKey {

crates/ourios-querier/src/lib.rs

@@ -20,11 +20,13 @@
 //! §4.6). It reads the shipped RFC 0005 store; it needs neither
 //! the WAL nor the receiver.
 //!
-//! (Partition-level *time* pruning — deriving `year/month/day/hour`
-//! path bounds from the time range so whole directories are
-//! skipped — is a later refinement; today the time bound is a
-//! column predicate, and the row-group skipping it enables is what
-//! slice 2 / B1 measures.)
+//! Partition-level *time* pruning is live: a query with a time range
+//! skips whole `year/month/day/hour` partitions whose span can't
+//! overlap the window (`hour_partition_in_window`) before `DataFusion`
+//! opens any footer, so scanned row groups stay flat as the corpus's
+//! time span grows. It layers on the `time_unix_nano` column predicate
+//! (still the row-level correctness authority); the pruning is
+//! conservative and never drops an in-window partition.
 //!
 //! **Throwaway query surface.** [`QueryRequest`] is intentionally
 //! minimal — just the predicates B1/B2 need. The real logs DSL
@@ -52,6 +54,7 @@ use datafusion::prelude::{SessionContext, col, lit};
 use ourios_core::tenant::TenantId;
 use ourios_parquet::Manifest;
 use ourios_parquet::columns;
+use ourios_parquet::hour_partition_in_window;
 use ourios_parquet::percent_encode_tenant;
 
 /// A logs query to execute. **Throwaway surface** while the query
@@ -168,7 +171,10 @@ impl std::error::Error for QueryError {}
 /// denied, transient failure) is propagated as [`QueryError::Storage`]
 /// rather than silently masked as "no data" — a wrong zero-row answer
 /// is worse than a surfaced error.
-fn resolve_live_files(dir: &std::path::Path) -> Result<Vec<PathBuf>, QueryError> {
+fn resolve_live_files(
+    dir: &std::path::Path,
+    window: Option<(u64, u64)>,
+) -> Result<Vec<PathBuf>, QueryError> {
     let io_err = |op: &str, p: &std::path::Path, e: &std::io::Error| QueryError::Storage {
         detail: format!("{op} {}: {e}", p.display()),
     };
@@ -194,13 +200,25 @@ fn resolve_live_files(dir: &std::path::Path) -> Result<Vec<PathBuf>, QueryError>
                 Err(e) => return Err(io_err("file_type", &path, &e)),
             }
         }
-        match Manifest::read(&d).map_err(|e| QueryError::Storage {
-            detail: format!("manifest in {}: {e}", d.display()),
-        })? {
-            // Manifest is authoritative: only its named files are live.
-            Some(manifest) => files.extend(manifest.files.into_iter().map(|name| d.join(name))),
-            // No manifest → glob fallback for this partition.
-            None => files.append(&mut parquets),
+        // Partition-level time pruning (RFC 0007): when the query has a
+        // time range, skip a leaf partition whose `hour=HH` span can't
+        // overlap it — so DataFusion never opens those footers. This is
+        // a pure optimisation layered on the row-level time column
+        // predicate (which stays the correctness authority);
+        // `hour_partition_in_window` is conservative, never pruning a
+        // path it can't prove out of range, so no in-window data is lost.
+        let keep = window.is_none_or(|(start, end)| hour_partition_in_window(&d, start, end));
+        if keep {
+            match Manifest::read(&d).map_err(|e| QueryError::Storage {
+                detail: format!("manifest in {}: {e}", d.display()),
+            })? {
+                // Manifest is authoritative: only its named files are live.
+                Some(manifest) => {
+                    files.extend(manifest.files.into_iter().map(|name| d.join(name)));
+                }
+                // No manifest → glob fallback for this partition.
+                None => files.append(&mut parquets),
+            }
         }
         stack.extend(subdirs);
     }
@@ -343,7 +361,7 @@ impl Querier {
         // holding only `*.parquet.tmp` (a poisoned/crashed writer) —
         // where building a table over zero files would otherwise
         // error and wrongly fail the query.
-        let live_files = resolve_live_files(&tenant_dir)?;
+        let live_files = resolve_live_files(&tenant_dir, request.time_range)?;
         if live_files.is_empty() {
             return Ok(QueryResult::default());
         }
@@ -632,7 +650,7 @@ mod tests {
         let ghost = tmp.path().join("data/tenant_id=ghost");
 
         // Act
-        let files = resolve_live_files(&ghost).expect("resolve");
+        let files = resolve_live_files(&ghost, None).expect("resolve");
 
         // Assert
         assert!(files.is_empty());
@@ -646,7 +664,7 @@ mod tests {
         std::fs::write(partition.join("x.parquet.tmp"), b"partial").expect("write tmp");
 
         // Act
-        let files = resolve_live_files(&tenant).expect("resolve");
+        let files = resolve_live_files(&tenant, None).expect("resolve");
 
         // Assert
         assert!(files.is_empty(), "uncommitted .tmp files are not live");
@@ -661,7 +679,7 @@ mod tests {
         std::fs::write(partition.join("b.parquet"), b"b").expect("write b");
 
         // Act
-        let files = resolve_live_files(&tenant).expect("resolve");
+        let files = resolve_live_files(&tenant, None).expect("resolve");
 
         // Assert
         assert_eq!(
@@ -689,7 +707,7 @@ mod tests {
         .expect("write manifest");
 
         // Act
-        let files = resolve_live_files(&tenant).expect("resolve");
+        let files = resolve_live_files(&tenant, None).expect("resolve");
 
         // Assert
         assert_eq!(files.len(), 1, "only the manifest's file is live");
@@ -709,7 +727,7 @@ mod tests {
         .expect("write manifest");
 
         // Act
-        let result = resolve_live_files(&tenant);
+        let result = resolve_live_files(&tenant, None);
 
         // Assert
         assert!(matches!(result, Err(QueryError::Storage { .. })));

crates/ourios-querier/tests/execution.rs

@@ -365,6 +365,84 @@ async fn rfc0007_2_template_exact_work_scales_with_result_not_corpus() {
     );
 }
 
+/// Lay down a target file in hour 10 (3 rows of template 1) plus one
+/// filler file in each of the next `filler_hours` hours (distinct
+/// templates, distinct partitions).
+fn corpus_over_hours(bucket: &Path, filler_hours: u64) {
+    write_all(
+        bucket,
+        &[
+            rec("a", 1, TS0),
+            rec("a", 1, TS0 + 1_000_000),
+            rec("a", 1, TS0 + 2_000_000),
+        ],
+    );
+    for k in 1..=filler_hours {
+        write_all(bucket, &[rec("a", 100 + k, TS0 + k * HOUR_NS)]);
+    }
+}
+
+/// RFC0007.2 (B2) — partition-level **time** pruning. A time-windowed
+/// query skips whole `hour=` partitions outside the window *before*
+/// `DataFusion` opens their footers, so the work tracks the window, not
+/// the corpus's time span. Two corpora share the same in-window hour-10
+/// data but differ 10× in time span (2 vs 20 filler hours). A query
+/// windowed to hour 10 returns the same 3 rows and — crucially — hands
+/// `DataFusion` the *same* (tiny) set of row groups in both, because the
+/// out-of-window partitions are pruned at the directory level. This is
+/// the piece the `otel-demo` run showed is missing for real logs, where
+/// a template recurs across every partition.
+#[tokio::test]
+async fn rfc0007_2_time_window_prunes_whole_partitions() {
+    let small = tempfile::TempDir::new().expect("temp small");
+    let large = tempfile::TempDir::new().expect("temp large");
+    corpus_over_hours(small.path(), 2); //  3 partitions
+    corpus_over_hours(large.path(), 20); // 21 partitions (10× the time span)
+
+    // A ~1s window inside hour 10 — covers only the target partition.
+    let window = Some((TS0, TS0 + 1_000_000_000));
+    let s = Querier::new(small.path())
+        .run(req("a", window, None))
+        .await
+        .expect("small query");
+    let l = Querier::new(large.path())
+        .run(req("a", window, None))
+        .await
+        .expect("large query");
+
+    // Same in-window result regardless of how many out-of-window hours exist.
+    assert_eq!(s.rows, 3);
+    assert_eq!(
+        l.rows, 3,
+        "the window result is fixed across corpus time span"
+    );
+
+    // The work handed to DataFusion is FLAT across a 10× larger time
+    // span — only hour 10's partition reaches the engine in both.
+    assert_eq!(
+        s.stats.row_groups_scanned, l.stats.row_groups_scanned,
+        "scanned row groups track the window, not the corpus; small={:?} large={:?}",
+        s.stats, l.stats,
+    );
+    assert_eq!(
+        s.stats.bytes_read, l.stats.bytes_read,
+        "bytes read track the window, not the corpus; small={:?} large={:?}",
+        s.stats, l.stats,
+    );
+
+    // Directory-level pruning means the out-of-window partitions never
+    // reach DataFusion at all: the total row groups it sees (scanned +
+    // statistics-pruned) stays tiny even at 20 filler hours — it does
+    // NOT grow with the corpus (which it would if every partition's
+    // footer were opened and pruned by the column predicate instead).
+    let l_total = l.stats.row_groups_scanned + l.stats.row_groups_pruned;
+    assert!(
+        l_total <= 2,
+        "out-of-window partitions are pruned before DataFusion; it saw {l_total} row groups (stats={:?})",
+        l.stats,
+    );
+}
+
 /// A `bucket_root` whose path contains a space still resolves —
 /// the URL is built from the canonical path (`DataFusion`
 /// URI-encodes it), not a raw `file://{display}` string that would