feat(querier): group count by a promoted attribute column (RFC 0037 §3.3)

crates/ourios-querier/src/compile.rs

@@ -434,6 +434,33 @@ pub(crate) fn group_exprs(by: &[GroupTerm], df: &DataFrame) -> Result<Vec<Expr>,
         .collect()
 }
 
+/// Lower a group-by on a promoted attribute column (RFC 0037 §3.3). Groups on
+/// the promoted `resource.<key>` / `attr.<key>` column when it is present in
+/// the scanned union schema; otherwise rejects with a hint pointing at
+/// promotion, so grouping never silently degrades to a single NULL bucket or
+/// an unpruned JSON scan.
+fn group_by_promoted(column: &str, key: &str, df: &DataFrame) -> Result<Expr, QueryError> {
+    let name = promoted_column_name(column, key);
+    if has_column(df, &name) {
+        Ok(Expr::Column(Column::new_unqualified(name)))
+    } else {
+        // Name the raw config key (no `attr.`/`resource.` prefix) and the
+        // sublist it belongs under, so the hint points at the exact string to
+        // add rather than the derived column name.
+        let sublist = if column == columns::RESOURCE_ATTRIBUTES {
+            "resource"
+        } else {
+            "log"
+        };
+        Err(QueryError::InvalidQuery {
+            detail: format!(
+                "grouping by '{name}' requires the attribute to be promoted to a column present \
+                 in the queried range; add '{key}' to storage.promoted_attributes.{sublist}"
+            ),
+        })
+    }
+}
+
 fn field_group_expr(field: &Field, df: &DataFrame) -> Result<Expr, QueryError> {
     match field {
         Field::Service => {
@@ -445,13 +472,16 @@ fn field_group_expr(field: &Field, df: &DataFrame) -> Result<Expr, QueryError> {
                 Ok(lit(ScalarValue::Utf8(None)))
             }
         }
-        // §7 confines `by`-list fields to the bare set, so these are only
-        // reachable from a hand-built IR — reject rather than group over the
-        // JSON-encoded attribute columns.
-        Field::Resource(_) | Field::Attr(_) => Err(QueryError::InvalidQuery {
-            detail: "grouping by resource/attr attributes is not supported in this query surface"
-                .to_string(),
-        }),
+        // RFC 0037 §3.3: group by a *promoted* attribute column. The column
+        // is present in the scanned union schema exactly when ≥ 1 scanned
+        // file promoted the key (DataFusion supplies per-file NULLs for any
+        // pre-promotion partitions within a mixed scan — the typed-NULL
+        // fallback happens for free). Absent from every scanned file, the key
+        // is not a usable group key here: reject with a promotion hint rather
+        // than collapse every row into one NULL bucket or group over an
+        // unpruned JSON scan (hazard #6).
+        Field::Resource(key) => group_by_promoted(columns::RESOURCE_ATTRIBUTES, key, df),
+        Field::Attr(key) => group_by_promoted(columns::ATTRIBUTES, key, df),
         _ => {
             let (column, optional) = column_of(field);
             if optional && !has_column(df, column) {

crates/ourios-querier/src/dsl/parse.rs

@@ -1044,7 +1044,10 @@ impl<'a> Parser<'a> {
                 self.expect(&Tok::RParen, "')' to close bucket(...)")?;
                 Ok(GroupTerm::Bucket(width))
             }
-            _ => Ok(GroupTerm::Field(self.parse_field()?)),
+            // A bare field, or (RFC 0037 §3.3) a `resource.`/`attr.` path —
+            // `parse_path` accepts both; the compiler gates the path forms on
+            // the key being a promoted column.
+            _ => Ok(GroupTerm::Field(self.parse_path()?)),
         }
     }
 

crates/ourios-querier/src/dsl/structured.rs

@@ -467,19 +467,15 @@ impl RawGroupTerm {
             Self::Bucket(RawBucketTerm { bucket }) => {
                 Ok(GroupTerm::Bucket(parse_duration_lexeme_pub(&bucket)?))
             }
-            // The string DSL's `group_term = field` production (§7 v1.1)
-            // only accepts a bare top-level field — `resource.`/`attr.`
-            // paths are rejected there (see `parse_field`'s error message).
-            // A `{resource|attr}` object here would let the structured
-            // surface express a `by`-list the string grammar cannot, so it
-            // is rejected at the same boundary rather than reaching the
-            // planner (RFC0002.2).
-            Self::Field(RawField::Object(_)) => Err(DslError::new(
-                "a by-list field must be a bare field name (resource./attr. paths \
-                 are not allowed here)"
-                    .to_string(),
-            )),
-            Self::Field(field @ RawField::Name(_)) => Ok(GroupTerm::Field(field.into_ir()?)),
+            // RFC 0037 §3.3: a `by`-list field is a bare top-level field or a
+            // `resource.`/`attr.` path (the `{resource|attr}` object form).
+            // Both surfaces admit the same set — the string grammar's
+            // `parse_group_term` now routes through `parse_path` — and the
+            // compiler gates the path forms on the key being a promoted
+            // column.
+            Self::Field(field @ (RawField::Object(_) | RawField::Name(_))) => {
+                Ok(GroupTerm::Field(field.into_ir()?))
+            }
         }
     }
 }
@@ -639,7 +635,7 @@ fn parse_time(s: &str) -> Result<Time, DslError> {
 #[cfg(test)]
 mod tests {
     use super::parse_structured;
-    use crate::dsl::ir::{Call, CmpOp, Field, OrdOp, Predicate, Stage, Value};
+    use crate::dsl::ir::{Call, CmpOp, Field, GroupTerm, OrdOp, Predicate, Stage, Value};
 
     #[test]
     fn parses_comparison_with_attr_object() {
@@ -826,25 +822,41 @@ mod tests {
     }
 
     #[test]
-    fn rejects_resource_and_attr_group_terms() {
-        // The string DSL's `group_term = field` production (§7 v1.1) is
-        // bare-field-only; a structured `{resource|attr}` by-element would
-        // let this surface express a query the string grammar cannot
-        // (RFC0002.2), so it is rejected here rather than reaching the
-        // planner.
-        for req in [
-            r#"{"predicate":{"const":true},"stages":[{"count":{"by":[{"resource":"k8s.pod.name"}]}}]}"#,
-            r#"{"predicate":{"const":true},"stages":[{"count":{"by":[{"attr":"http.status_code"}]}}]}"#,
-            r#"{"predicate":{"const":true},"stages":[{"avg":"confidence","by":[{"attr":"k"}]}]}"#,
-        ] {
-            let err = parse_structured(req).unwrap_err();
-            assert!(
-                err.message().contains("bare field"),
-                "{}: {}",
-                req,
-                err.message()
-            );
+    fn accepts_resource_and_attr_group_terms() {
+        // RFC 0037 §3.3 extends the by-list to `resource.`/`attr.` paths on
+        // both surfaces — the string grammar's `parse_group_term` now routes
+        // through `parse_path`, so the two surfaces still admit the same set.
+        // The compiler gates the path forms on the key being a promoted
+        // column; parsing accepts them. (Previously this surface rejected them
+        // to stay within the older bare-field-only string grammar.)
+        let count_cases: &[(&str, GroupTerm)] = &[
+            (
+                r#"{"predicate":{"const":true},"stages":[{"count":{"by":[{"resource":"k8s.pod.name"}]}}]}"#,
+                GroupTerm::Field(Field::Resource("k8s.pod.name".into())),
+            ),
+            (
+                r#"{"predicate":{"const":true},"stages":[{"count":{"by":[{"attr":"http.status_code"}]}}]}"#,
+                GroupTerm::Field(Field::Attr("http.status_code".into())),
+            ),
+        ];
+        for (req, expected) in count_cases {
+            let query = parse_structured(req).expect("resource/attr count group terms now parse");
+            let Some(Stage::Count { by }) = query.stages.first() else {
+                panic!("{req}: expected a count stage");
+            };
+            assert_eq!(by.len(), 1, "{req}: one group term");
+            assert_eq!(&by[0], expected, "{req}");
         }
+
+        // The relaxation applies to any grouped aggregate (they share
+        // `group_terms_to_ir`), e.g. `avg … by attr.k`.
+        assert!(
+            parse_structured(
+                r#"{"predicate":{"const":true},"stages":[{"avg":"confidence","by":[{"attr":"k"}]}]}"#
+            )
+            .is_ok(),
+            "avg grouped by a resource/attr path also parses now"
+        );
     }
 
     #[test]

crates/ourios-querier/tests/it/rfc0002_dsl.rs

@@ -1775,6 +1775,108 @@ async fn rfc0002_13_count_by_param_bucket_grouped_map() {
     );
 }
 
+/// Scenario RFC0037.4 — `count … by` a *promoted* attribute column
+/// (`attr.gen_ai.request.model`) matches a brute-force `(bucket, model) →
+/// count` oracle; the same query against a *non-promoted* key is rejected
+/// with a promotion hint. See `docs/rfcs/0037-genai-structured-log-events.md`
+/// §3.3.
+#[tokio::test]
+async fn rfc0037_4_count_by_promoted_attribute() {
+    use std::collections::BTreeMap;
+
+    use crate::common::{
+        DEFAULT_WINDOW_NS, NOW, TS0, kv, no_aliases, rec_with_attrs, write_all,
+        write_all_with_promoted,
+    };
+    use ourios_core::tenant::TenantId;
+    use ourios_parquet::PromotedAttributes;
+
+    const SECOND_NS: u64 = 1_000_000_000;
+    const WIDTH_NS: u64 = 300 * SECOND_NS; // 5m
+
+    // Rows carrying a `gen_ai.request.model` log attribute across two models
+    // and two 5-minute windows.
+    let rows: &[(&str, u64)] = &[
+        ("gpt-4", TS0),
+        ("gpt-4", TS0 + 100 * SECOND_NS),
+        ("claude", TS0 + 150 * SECOND_NS),
+        ("gpt-4", TS0 + 400 * SECOND_NS),
+    ];
+    let recs: Vec<_> = rows
+        .iter()
+        .map(|(model, ts)| {
+            rec_with_attrs(
+                "a",
+                *ts,
+                vec![kv("service.name", "checkout")],
+                vec![kv("gen_ai.request.model", model)],
+            )
+        })
+        .collect();
+
+    // Promote the log key so it gets a dedicated, groupable column.
+    let promoted = PromotedAttributes::new(
+        Vec::<String>::new(),
+        vec!["gen_ai.request.model".to_string()],
+    );
+
+    let bucket = tempfile::TempDir::new().expect("temp");
+    write_all_with_promoted(bucket.path(), &recs, &promoted);
+
+    // Brute-force oracle: (model, bucket_key) → count.
+    let bucket_key = |ts: u64| {
+        let start = ts / WIDTH_NS * WIDTH_NS;
+        chrono::DateTime::from_timestamp_nanos(i64::try_from(start).expect("fixture ns"))
+            .to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true)
+    };
+    let mut oracle: BTreeMap<Vec<String>, u64> = BTreeMap::new();
+    for (model, ts) in rows {
+        *oracle
+            .entry(vec![(*model).to_string(), bucket_key(*ts)])
+            .or_default() += 1;
+    }
+
+    let result = run_dsl(
+        bucket.path(),
+        "template_id == 1 | range(2026-04-02T10:00:00Z, 2026-04-02T12:00:00Z) \
+         | count by attr.gen_ai.request.model, bucket(5m)",
+    )
+    .await;
+    assert_eq!(
+        group_map(&result),
+        oracle,
+        "grouped count over the promoted attribute matches the brute-force map"
+    );
+
+    // Rejection: the identical grouping over a store that did NOT promote the
+    // key is rejected with a promotion hint — never a silent unpruned scan.
+    let unpromoted = tempfile::TempDir::new().expect("temp");
+    write_all(unpromoted.path(), &recs);
+    let query = ourios_querier::dsl::parse("template_id == 1 | count by attr.gen_ai.request.model")
+        .expect("parse");
+    let err = ourios_querier::Querier::new(unpromoted.path())
+        .run_query(
+            &query,
+            &TenantId::new("a"),
+            NOW,
+            DEFAULT_WINDOW_NS,
+            Some(&no_aliases()),
+        )
+        .await
+        .expect_err("grouping by a non-promoted attribute must be rejected");
+    let msg = err.to_string();
+    // The hint names the raw config key (no `attr.` prefix) and the sublist to
+    // add it under, not just the derived column name.
+    assert!(
+        msg.contains("gen_ai.request.model"),
+        "the rejection names the raw config key; got: {msg}"
+    );
+    assert!(
+        msg.contains("storage.promoted_attributes.log"),
+        "the rejection names the config sublist; got: {msg}"
+    );
+}
+
 /// Scenario RFC0002.14 — `param(n)` misuse is a specific compile-time
 /// error. See `docs/rfcs/0002-query-dsl.md` §5 (amendment 2026-07-15).
 #[tokio::test]