feat(querier): rfc 0022 green pt2 — promoted predicate compile (RFC0022.3/.4/.6)

crates/ourios-parquet/src/promoted.rs

@@ -16,10 +16,13 @@ use ourios_core::otlp::{KeyValue, any_value};
 /// resource entity, surfaced in the DSL as the bare `service` field.
 pub const SERVICE_NAME_KEY: &str = "service.name";
 
-/// Column-name prefix for promoted resource-attribute keys.
-const RESOURCE_PREFIX: &str = "resource.";
-/// Column-name prefix for promoted log-attribute keys.
-const ATTR_PREFIX: &str = "attr.";
+/// Column-name prefix for promoted resource-attribute keys. Public because
+/// the query-side compile (RFC 0022 §3.3) derives promoted column names from
+/// the same prefixes the writer declares.
+pub const RESOURCE_PREFIX: &str = "resource.";
+/// Column-name prefix for promoted log-attribute keys (see
+/// [`RESOURCE_PREFIX`]).
+pub const ATTR_PREFIX: &str = "attr.";
 
 /// The effective promoted attribute key set (RFC 0022 §3.1/§3.2).
 ///

crates/ourios-querier/src/compile.rs

@@ -31,15 +31,20 @@
 //! RFC 0005 §3.9 `effective := time_unix_nano` fallback for files that
 //! predate the column.
 //!
-//! `service`, `resource.<k>`, and `attr.<k>` have **no dedicated column** in
-//! the RFC 0005 schema: resource/log attributes are stored as a single
-//! Ourios-canonical-JSON `Utf8` column (`resource_attributes` / `attributes`).
-//! They compile to a substring/`LIKE` match against that JSON column using a
-//! needle built from the canonical `{"key":…,"value":{"stringValue":…}}`
-//! shape — honest about the storage, not a column that doesn't exist. This is
-//! a `Filter` with no row-group-pruning claim (RFC 0002 §5 RFC0002.6), and is
-//! limited to string equality / string calls; ordering comparisons on a
-//! JSON-encoded attribute are out of scope for this slice and rejected.
+//! `service`, `resource.<k>`, and `attr.<k>` are attribute-backed:
+//! resource/log attributes are stored as a single Ourios-canonical-JSON
+//! `Utf8` column (`resource_attributes` / `attributes`), plus — for keys in
+//! the RFC 0022 promoted set — a dedicated `OPTIONAL Utf8` column named
+//! after the DSL path (`resource.<k>` / `attr.<k>`). When the scanned union
+//! schema carries a key's promoted column, [`attr_match`] compiles the full
+//! `cmp_op` set against it (§3.3's two-arm form for `==`/`!=`, typed-arm
+//! only for ordering/regex) and the typed arm prunes row groups. Otherwise
+//! the key compiles to a substring/`LIKE` match against the JSON column
+//! using a needle built from the canonical
+//! `{"key":…,"value":{"stringValue":…}}` shape — honest about the storage,
+//! a `Filter` with no row-group-pruning claim (RFC 0002 §5 RFC0002.6),
+//! limited to string equality; ordering/regex on a non-promoted
+//! (JSON-encoded) attribute stay rejected.
 //!
 //! ## Absent OPTIONAL columns (RFC 0005 §3.9 / RFC0007.4)
 //!
@@ -52,6 +57,7 @@
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 
+use datafusion::common::Column;
 use datafusion::dataframe::DataFrame;
 use datafusion::functions::expr_fn::{regexp_like, starts_with};
 use datafusion::logical_expr::{Expr, not};
@@ -64,7 +70,7 @@ use crate::dsl::ir::{
     Call, CmpOp, Field, OrdOp, Predicate, Query, SeverityValue, Stage, Time, Value,
 };
 use crate::{QueryError, has_column, time_bound_scalar};
-use ourios_parquet::columns;
+use ourios_parquet::{columns, promoted};
 
 /// A compiled query: the resolved time window (drives both the
 /// directory-level partition pruning and the row-level time filter) and the
@@ -594,13 +600,27 @@ fn field_name(field: &Field) -> String {
     }
 }
 
-/// Compile an attribute equality (`service`/`resource.k`/`attr.k`) to a
-/// substring `LIKE` over the Ourios-canonical-JSON column. Only `==`/`!=` on a
-/// string value is supported in this slice; the canonical encoding stores
-/// string values as `{"key":"<k>","value":{"stringValue":"<v>"}}`, so an
-/// exact key+string-value pair is matched by that JSON fragment as a `LIKE`
-/// substring. Ordering / regex / non-string comparisons over a JSON-encoded
-/// attribute are rejected (out of scope until attributes are columned).
+/// Compile an attribute comparison (`service`/`resource.k`/`attr.k`).
+///
+/// When the scanned union schema carries the key's RFC 0022 promoted column
+/// (`resource.<k>` / `attr.<k>` — §3.4's compile rule), the operator set is
+/// the full `cmp_op` (§3.3):
+///
+/// - `==`/`!=` compile to the two-arm form — the typed column arm (prunable)
+///   `OR` a `P IS NULL AND <JSON arm>` fallback covering pre-amendment files
+///   and non-string values.
+/// - Ordering and regex compile against the typed arm only; the JSON arm
+///   cannot express them, so rows whose promoted cell is `NULL`
+///   (pre-amendment files, non-string values) never match — §3.3's
+///   documented silent non-match, consistent with the DSL's missing-field
+///   rule.
+///
+/// Without the promoted column, `==`/`!=` on a string value keep the #146
+/// substring `LIKE` over the Ourios-canonical-JSON column (the canonical
+/// encoding stores string values as
+/// `{"key":"<k>","value":{"stringValue":"<v>"}}`, so an exact
+/// key+string-value pair is matched by that JSON fragment), and every other
+/// operator is rejected — unchanged pre-RFC 0022 behaviour.
 fn attr_match(
     column: &str,
     key: &str,
@@ -618,13 +638,28 @@ fn attr_match(
             detail: "attribute comparisons take a string value in this query surface".to_string(),
         });
     };
-    let ord = match op {
+    let promoted_name = promoted_column_name(column, key);
+    // `col()` parses dotted names as qualified references, so the promoted
+    // column (literally named `resource.<k>` / `attr.<k>`) must be addressed
+    // as an unqualified `Column` built directly.
+    let promoted = has_column(df, &promoted_name)
+        .then(|| Expr::Column(Column::new_unqualified(promoted_name)));
+    let eq = match op {
         CmpOp::Ord(OrdOp::Eq) => true,
         CmpOp::Ord(OrdOp::Ne) => false,
-        _ => {
-            return Err(QueryError::InvalidQuery {
-                detail: "attributes support only == / != in this query surface".to_string(),
-            });
+        op => {
+            let Some(p) = promoted else {
+                return Err(QueryError::InvalidQuery {
+                    detail: "non-promoted attributes support only == / != in this query surface"
+                        .to_string(),
+                });
+            };
+            let expr = match op {
+                CmpOp::Ord(ord) => ord_expr(p, ord, lit(v.clone())),
+                CmpOp::Match => regexp_like(p, lit(v.clone()), None),
+                CmpOp::NotMatch => not(regexp_like(p, lit(v.clone()), None)),
+            };
+            return Ok(PredExpr::Filter(expr));
         }
     };
     // The canonical JSON fragment for this key/value pair. `serde_json`'s
@@ -638,18 +673,45 @@ fn attr_match(
     })?;
     let fragment = format!("{{\"key\":{needle_key},\"value\":{{\"stringValue\":{needle_value}}}}}");
     let value_match = col(column).like(lit(format!("%{}%", like_escape(&fragment))));
-    if ord {
+    let json = if eq {
         // `==` matches when the key is present with this exact string value.
-        return Ok(PredExpr::Filter(value_match));
+        value_match
+    } else {
+        // `!=` must require the key PRESENT with a *different* value: a row
+        // missing the key does not match. The presence guard matches the key
+        // with any string value, then we exclude the exact value above.
+        // Without the guard, `NOT LIKE` is also true for absent keys, which
+        // diverges from the missing-field "no match" semantics used
+        // everywhere else.
+        let key_present = format!("{{\"key\":{needle_key},\"value\":{{\"stringValue\":");
+        let presence = col(column).like(lit(format!("%{}%", like_escape(&key_present))));
+        presence.and(not(value_match))
+    };
+    let Some(p) = promoted else {
+        return Ok(PredExpr::Filter(json));
+    };
+    // §3.3's two-arm form. The `!=` typed arm keeps the presence check
+    // explicit (`P IS NOT NULL AND P != v`) rather than leaning on 3-valued
+    // logic, mirroring the JSON arm's presence guard.
+    let expr = if eq {
+        p.clone().eq(lit(v.clone())).or(p.is_null().and(json))
+    } else {
+        p.clone()
+            .is_not_null()
+            .and(p.clone().not_eq(lit(v.clone())))
+            .or(p.is_null().and(json))
+    };
+    Ok(PredExpr::Filter(expr))
+}
+
+/// The RFC 0022 promoted column name for an attribute key: the literal DSL
+/// path (`resource.<k>` / `attr.<k>`, §3.1), derived from the same prefixes
+/// the writer's [`ourios_parquet::promoted`] module declares.
+fn promoted_column_name(column: &str, key: &str) -> String {
+    match column {
+        columns::RESOURCE_ATTRIBUTES => format!("{}{key}", promoted::RESOURCE_PREFIX),
+        _ => format!("{}{key}", promoted::ATTR_PREFIX),
     }
-    // `!=` must require the key PRESENT with a *different* value: a row
-    // missing the key does not match. The presence guard matches the key with
-    // any string value, then we exclude the exact value above. Without the
-    // guard, `NOT LIKE` is also true for absent keys, which diverges from the
-    // missing-field "no match" semantics used everywhere else.
-    let key_present = format!("{{\"key\":{needle_key},\"value\":{{\"stringValue\":");
-    let presence = col(column).like(lit(format!("%{}%", like_escape(&key_present))));
-    Ok(PredExpr::Filter(presence.and(not(value_match))))
 }
 
 /// Escape the `%` / `_` / `\` wildcards in a `LIKE` pattern literal so the

crates/ourios-querier/src/lib.rs

@@ -797,14 +797,32 @@ impl Querier {
         // `schema_force_view_types` override is needed.
         let options =
             ListingOptions::new(Arc::new(ParquetFormat::default())).with_file_extension(".parquet");
-        // `infer_schema` over the multi-path set merges the files'
-        // schemas, so additive schema drift across files reads as the
-        // union (RFC0007.4 / RFC 0005 §3.9).
+        // The table schema must be the **union** across every scanned file
+        // (RFC0007.4 / RFC 0005 §3.9), and since RFC 0022 the union is
+        // load-bearing for predicate compilation: `attr_match` gates the
+        // promoted-column arms on the post-union schema (§3.4). A bare
+        // `ListingTableConfig::infer_schema` infers from the *first* table
+        // path only — with the per-file URLs `resolve_data_urls` produces,
+        // that is one arbitrary file, not the union — so infer per file and
+        // merge. The extra footer reads are already paid: the Parquet format
+        // fetches every listed file's footer for statistics at plan time.
+        let mut schemas = Vec::with_capacity(urls.len());
+        for url in &urls {
+            let schema = options
+                .infer_schema(&ctx.state(), url)
+                .await
+                .map_err(storage_err)?;
+            schemas.push(schema.as_ref().clone());
+        }
+        let file_schema =
+            datafusion::arrow::datatypes::Schema::try_merge(schemas).map_err(|e| {
+                QueryError::Storage {
+                    detail: format!("merging scanned file schemas: {e}"),
+                }
+            })?;
         let config = ListingTableConfig::new_with_multi_paths(urls)
             .with_listing_options(options)
-            .infer_schema(&ctx.state())
-            .await
-            .map_err(storage_err)?;
+            .with_schema(Arc::new(file_schema));
         let table = ListingTable::try_new(config).map_err(storage_err)?;
         ctx.register_table("logs", Arc::new(table))
             .map_err(storage_err)?;

crates/ourios-querier/tests/common/mod.rs

@@ -17,7 +17,7 @@ use ourios_core::otlp::any_value::Value as AvValue;
 use ourios_core::otlp::{AnyValue, KeyValue};
 use ourios_core::record::{BodyKind, MinedRecord, Param};
 use ourios_core::tenant::TenantId;
-use ourios_parquet::{PartitionKey, Writer};
+use ourios_parquet::{DEFAULT_ZSTD_LEVEL, PartitionKey, PromotedAttributes, Writer};
 
 /// 2026-04-02T10:58:00 UTC — the same base instant the execution tests
 /// use, so all fixture rows land in one `hour=` partition unless bumped.
@@ -115,6 +115,42 @@ pub fn write_all(bucket: &Path, recs: &[MinedRecord]) {
     }
 }
 
+/// [`write_all`] with an explicit RFC 0022 promoted attribute set, so a test
+/// can seed post-amendment files whose promoted columns go beyond the
+/// implicit `service.name`.
+pub fn write_all_with_promoted(bucket: &Path, recs: &[MinedRecord], promoted: &PromotedAttributes) {
+    let store = Store::local(bucket).expect("local store");
+    let mut by_part: HashMap<PartitionKey, Vec<MinedRecord>> = HashMap::new();
+    for r in recs {
+        by_part
+            .entry(PartitionKey::derive(r).expect("derive partition"))
+            .or_default()
+            .push(r.clone());
+    }
+    for (part, rs) in by_part {
+        let mut w =
+            Writer::open_in_with_promoted(&store, part, DEFAULT_ZSTD_LEVEL, promoted.clone())
+                .expect("open writer");
+        w.append_records(&rs).expect("append");
+        w.close().expect("close");
+    }
+}
+
+/// A record with explicit log `attributes` on top of [`rec_with_resource`]'s
+/// explicit resource attributes, so promoted-column tests can drive both
+/// families.
+pub fn rec_with_attrs(
+    tenant: &str,
+    ts_ns: u64,
+    resource_attributes: Vec<KeyValue>,
+    attributes: Vec<KeyValue>,
+) -> MinedRecord {
+    MinedRecord {
+        attributes,
+        ..rec_with_resource(tenant, ts_ns, resource_attributes)
+    }
+}
+
 /// A window wide enough that a query with no `range(...)` (which gets the
 /// default look-back ending at `now`) still covers all fixture rows.
 pub const DEFAULT_WINDOW_NS: u64 = 30 * 24 * HOUR_NS;