feat(querier): add severity_text predicate for the B1 level filter

crates/ourios-bench/benches/b2.rs

@@ -105,6 +105,7 @@ fn template_exact(template_id: u64) -> QueryRequest {
         tenant: TenantId::new("a"),
         time_range: None,
         template_id: Some(template_id),
+        severity_text: None,
     }
 }
 

crates/ourios-querier/src/lib.rs

@@ -2,9 +2,10 @@
 //!
 //! **Status: execution slice 3.** [`Querier::run`] executes a
 //! minimal query — tenant scope + optional time range + optional
-//! template-exact id — against the RFC 0005 Parquet store via
-//! `DataFusion`, returning a matching-row count **and the scan's
-//! row-group pruning stats** ([`QueryStats`]). Tenant isolation
+//! template-exact id + optional `severity_text` (the B1 `level='ERROR'`
+//! filter) — against the RFC 0005 Parquet store via `DataFusion`,
+//! returning a matching-row count **and the scan's row-group pruning
+//! stats** ([`QueryStats`]). Tenant isolation
 //! (RFC0007.5), B1 pruning (RFC0007.1 — a selective query provably
 //! skips row groups via statistics) and B2 (RFC0007.2 — the work
 //! the engine does tracks the result size, not the corpus size;
@@ -57,9 +58,10 @@ use ourios_parquet::percent_encode_tenant;
 /// thesis (B1/B2) is unproven — per the maintainer decision, DSL
 /// contracts (RFC 0002) are deferred until B1/B2 say the querier
 /// is worth a stable language. So this carries only the minimal
-/// predicates B1/B2 need: tenant scope, optional time bounds, and
-/// optional template-exact id — exactly the RFC 0005 §3.3
-/// pushdown keys.
+/// predicates B1/B2 need: tenant scope, optional time bounds,
+/// optional template-exact id, and an optional `severity_text`
+/// equality (the B1 `level='ERROR'` filter) — exactly the RFC 0005
+/// §3.3 pushdown keys.
 #[derive(Debug, Clone)]
 pub struct QueryRequest {
     /// Tenant whose data the query is scoped to. Enforced
@@ -70,6 +72,11 @@ pub struct QueryRequest {
     pub time_range: Option<(u64, u64)>,
     /// Optional template-exact filter (B2 — `template_id` equality).
     pub template_id: Option<u64>,
+    /// Optional `severity_text` equality filter — the B1 `level='ERROR'`
+    /// query shape (RFC 0005 §3.2 `severity_text` column). The
+    /// structured counterpart to the B1 reference's `grep ERROR`: rows
+    /// whose severity is null or anything else don't match.
+    pub severity_text: Option<String>,
 }
 
 /// Pruning / IO accounting for one query, surfaced so B1
@@ -423,6 +430,28 @@ impl Querier {
                 .filter(col(columns::TEMPLATE_ID).eq(lit(template_id)))
                 .map_err(storage_err)?;
         }
+        if let Some(severity_text) = &request.severity_text {
+            // `severity_text` is OPTIONAL (RFC 0005 §3.2). If a tenant's
+            // entire file set predates the column, the inferred union
+            // schema omits it — and filtering an unknown column would
+            // fail planning (surfacing as a generic Storage error). An
+            // absent OPTIONAL column reads as all-NULL (RFC 0005 §3.9 /
+            // RFC0007.4), so `severity_text = X` matches nothing: return
+            // an empty result rather than erroring. (Per-file drift,
+            // where *some* file has the column, is handled by DataFusion's
+            // schema union — see tests/forward_compat.rs.)
+            let has_severity = df
+                .schema()
+                .fields()
+                .iter()
+                .any(|f| f.name() == columns::SEVERITY_TEXT);
+            if !has_severity {
+                return Ok(QueryResult::default());
+            }
+            df = df
+                .filter(col(columns::SEVERITY_TEXT).eq(lit(severity_text.as_str())))
+                .map_err(storage_err)?;
+        }
 
         // Count via an aggregate so the heavy `attributes` /
         // `params` / `body` columns are never materialised

crates/ourios-querier/tests/boundary.rs

@@ -50,6 +50,7 @@ async fn rfc0007_3_real_engine_error_does_not_leak() {
             tenant: TenantId::new("a"),
             time_range: None,
             template_id: Some(1),
+            severity_text: None,
         })
         .await
         .expect_err("a corrupt parquet must surface as an error");

crates/ourios-querier/tests/execution.rs

@@ -98,6 +98,29 @@ fn req(tenant: &str, time_range: Option<(u64, u64)>, template_id: Option<u64>) -
         tenant: TenantId::new(tenant),
         time_range,
         template_id,
+        severity_text: None,
+    }
+}
+
+/// A record with an explicit severity (for the B1 `level='ERROR'`
+/// query shape) — `severity_text` and the canonical OTLP
+/// `severity_number` are set coherently, so a fixture keyed off
+/// either column agrees — otherwise identical to [`rec`].
+fn rec_sev(tenant: &str, template_id: u64, ts_ns: u64, severity: &str) -> MinedRecord {
+    // OTLP severity-number ranges (lower bound of each band).
+    let severity_number: u8 = match severity {
+        "TRACE" => 1,
+        "DEBUG" => 5,
+        "INFO" => 9,
+        "WARN" => 13,
+        "ERROR" => 17,
+        "FATAL" => 21,
+        _ => 0, // UNSPECIFIED
+    };
+    MinedRecord {
+        severity_text: Some(severity.to_string()),
+        severity_number,
+        ..rec(tenant, template_id, ts_ns)
     }
 }
 
@@ -241,6 +264,47 @@ async fn rfc0007_1_pushdown_prunes_row_groups() {
     );
 }
 
+/// RFC0007.1 (B1) — the `level='ERROR'` query shape. A
+/// `severity_text = 'ERROR'` filter both counts correctly and
+/// prunes via Parquet statistics: an INFO-only file in another hour
+/// has a `severity_text` min/max that can't satisfy `= 'ERROR'`, so
+/// its row group is skipped. This is the structured predicate that
+/// the B1 reference (`zstdcat | grep ERROR`) does by scanning.
+#[tokio::test]
+async fn rfc0007_1_severity_filter_counts_and_prunes() {
+    let bucket = tempfile::TempDir::new().expect("temp");
+    // hour 10: two ERROR rows + one INFO row (mixed file).
+    write_all(
+        bucket.path(),
+        &[
+            rec_sev("a", 1, TS0, "ERROR"),
+            rec_sev("a", 1, TS0 + 1_000_000, "ERROR"),
+            rec_sev("a", 1, TS0 + 2_000_000, "INFO"),
+        ],
+    );
+    // hour 11: an INFO-only file ⇒ its row group's severity_text
+    // min/max is INFO..INFO, which `= 'ERROR'` can prune.
+    write_all(bucket.path(), &[rec_sev("a", 1, TS0 + HOUR_NS, "INFO")]);
+
+    let q = Querier::new(bucket.path());
+    let r = q
+        .run(QueryRequest {
+            tenant: TenantId::new("a"),
+            time_range: None,
+            template_id: None,
+            severity_text: Some("ERROR".to_string()),
+        })
+        .await
+        .expect("severity query");
+
+    assert_eq!(r.rows, 2, "only the two ERROR rows match");
+    assert!(
+        r.stats.row_groups_pruned >= 1,
+        "the INFO-only file's row group is pruned by severity_text stats; stats={:?}",
+        r.stats,
+    );
+}
+
 /// RFC0007.2 (B2) — the inverted-index-collapse claim, measured
 /// structurally instead of by wall clock: for a fixed-result
 /// template-exact query, the *work the engine does* tracks the

crates/ourios-querier/tests/forward_compat.rs

@@ -82,6 +82,7 @@ fn req(time_range: Option<(u64, u64)>, template_id: Option<u64>) -> QueryRequest
         tenant: TenantId::new("a"),
         time_range,
         template_id,
+        severity_text: None,
     }
 }
 
@@ -201,3 +202,33 @@ async fn rfc0007_4_heterogeneous_schemas_stay_queryable() {
     let all = q.run(req(None, None)).await.expect("unfiltered query");
     assert_eq!(all.rows, 3, "all three heterogeneous files are read");
 }
+
+/// RFC0007.4 / §3.9 rule 2 — a B1 `severity_text='ERROR'` query against
+/// a tenant whose *entire* file set predates the OPTIONAL `severity_text`
+/// column. The inferred union schema omits the column, so a naive filter
+/// would fail planning; instead the absent OPTIONAL column reads as
+/// all-NULL and the equality matches nothing — an empty result, not a
+/// `QueryError`.
+#[tokio::test]
+async fn rfc0007_4_severity_filter_on_column_absent_everywhere_is_empty() {
+    let bucket = tempfile::TempDir::new().expect("temp");
+    // The only file omits severity_text (an old writer's output).
+    let old = rec(1, TS0);
+    write_raw_at(bucket.path(), &old, &old_schema_batch(&old));
+
+    let q = Querier::new(bucket.path());
+    let r = q
+        .run(QueryRequest {
+            tenant: TenantId::new("a"),
+            time_range: None,
+            template_id: None,
+            severity_text: Some("ERROR".to_string()),
+        })
+        .await
+        .expect("severity filter on an absent column must not error");
+
+    assert_eq!(
+        r.rows, 0,
+        "an absent OPTIONAL severity_text matches nothing (all-NULL)",
+    );
+}

crates/ourios-querier/tests/manifest.rs

@@ -54,6 +54,7 @@ fn req(template_id: Option<u64>) -> QueryRequest {
         tenant: TenantId::new("a"),
         time_range: None,
         template_id,
+        severity_text: None,
     }
 }