From fa57681b3403e801046f91e66c61969aae8240ed Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 18:29:00 -0400 Subject: [PATCH 1/4] feat(asap-common): RQE expression tree type and validation Add `asap_types::rqe_expr`: a serde type for RQE expression trees (rqe leaves, agg, binary, scalar nodes), a `validate` check run at load, and `leaves` / `output_labels` helpers. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011gKTCAEz3sJ1Hzde16xxU1 --- .../dependencies/rs/asap_types/src/lib.rs | 2 + .../rs/asap_types/src/rqe_expr.rs | 573 ++++++++++++++++++ 2 files changed, 575 insertions(+) create mode 100644 asap-common/dependencies/rs/asap_types/src/rqe_expr.rs diff --git a/asap-common/dependencies/rs/asap_types/src/lib.rs b/asap-common/dependencies/rs/asap_types/src/lib.rs index 167c753a..cdba5a38 100644 --- a/asap-common/dependencies/rs/asap_types/src/lib.rs +++ b/asap-common/dependencies/rs/asap_types/src/lib.rs @@ -7,6 +7,7 @@ pub mod inference_config; pub mod promql_schema; pub mod query_config; pub mod query_requirements; +pub mod rqe_expr; pub mod streaming_config; pub mod traits; pub mod utils; @@ -19,4 +20,5 @@ pub use inference_config::*; pub use promql_schema::*; pub use query_config::*; pub use query_requirements::*; +pub use rqe_expr::*; pub use streaming_config::*; diff --git a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs new file mode 100644 index 00000000..05e6000d --- /dev/null +++ b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs @@ -0,0 +1,573 @@ +//! RQE expression tree: self-contained RQE leaves combined by aggregations and +//! binary ops. Each JSON node is an object identified by its key: +//! +//! ```json +//! {"agg": "avg", "by": ["job"], "of": +//! {"op": "/", "lhs": {"rqe": {...}}, "rhs": {"scalar": 2.0}}} +//! ``` + +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RqeCapability { + Sum, + Quantile, + Cardinality, + TopkByValue, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RqeLeaf { + pub id: String, + /// Metric name (PromQL) or table name (SQL). + pub metric: String, + pub capability: RqeCapability, + pub lookback_ms: u64, + pub grouping: Vec, + /// Quantile rank; set iff `capability` is `quantile`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub phi: Option, + /// Top-k size; set iff `capability` is `topk_by_value`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub k: Option, + /// SQL value column. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RqeAggOp { + Avg, + Min, + Max, + Sum, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum RqeBinaryOp { + #[serde(rename = "+")] + Add, + #[serde(rename = "-")] + Sub, + #[serde(rename = "*")] + Mul, + #[serde(rename = "/")] + Div, +} + +// Serialized untagged (each variant's field names are the JSON keys); +// deserialized through `RawNode` so shape errors name the offending key. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(untagged, try_from = "RawNode")] +pub enum RqeExpr { + Rqe { + rqe: RqeLeaf, + }, + Agg { + agg: RqeAggOp, + by: Vec, + of: Box, + }, + Binary { + op: RqeBinaryOp, + lhs: Box, + rhs: Box, + }, + Scalar { + scalar: f64, + }, +} + +/// Union of all node keys; `TryFrom` picks the kind and rejects mixed keys. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawNode { + rqe: Option, + agg: Option, + by: Option>, + of: Option>, + op: Option, + lhs: Option>, + rhs: Option>, + scalar: Option, +} + +impl TryFrom for RqeExpr { + type Error = String; + + fn try_from(raw: RawNode) -> Result { + match raw { + RawNode { + rqe: Some(rqe), + agg: None, + by: None, + of: None, + op: None, + lhs: None, + rhs: None, + scalar: None, + } => Ok(RqeExpr::Rqe { rqe }), + RawNode { + rqe: None, + agg: Some(agg), + by: Some(by), + of: Some(of), + op: None, + lhs: None, + rhs: None, + scalar: None, + } => Ok(RqeExpr::Agg { agg, by, of }), + RawNode { + rqe: None, + agg: None, + by: None, + of: None, + op: Some(op), + lhs: Some(lhs), + rhs: Some(rhs), + scalar: None, + } => Ok(RqeExpr::Binary { op, lhs, rhs }), + RawNode { + rqe: None, + agg: None, + by: None, + of: None, + op: None, + lhs: None, + rhs: None, + scalar: Some(scalar), + } => Ok(RqeExpr::Scalar { scalar }), + _ => Err("RQE expression node must be exactly one of \ + {\"rqe\"}, {\"agg\", \"by\", \"of\"}, {\"op\", \"lhs\", \"rhs\"}, {\"scalar\"}" + .to_string()), + } + } +} + +#[derive(Debug, PartialEq, thiserror::Error)] +pub enum RqeExprError { + #[error("RQE {id}: quantile requires phi in [0, 1], got {phi:?}")] + InvalidPhi { id: String, phi: Option }, + #[error("RQE {id}: phi is only allowed for quantile")] + UnexpectedPhi { id: String }, + #[error("RQE {id}: topk_by_value requires k > 0, got {k:?}")] + InvalidK { id: String, k: Option }, + #[error("RQE {id}: k is only allowed for topk_by_value")] + UnexpectedK { id: String }, + #[error("RQE {id}: topk_by_value is only allowed at the root")] + TopkNotAtRoot { id: String }, + #[error("agg by {by:?} is not a subset of child labels {child:?}")] + AggByNotSubset { by: Vec, child: Vec }, + #[error("binary op needs at least one non-scalar side")] + BinaryBothScalar, + #[error("binary op sides have different labels: {lhs:?} vs {rhs:?}")] + BinaryLabelMismatch { lhs: Vec, rhs: Vec }, +} + +impl RqeExpr { + /// Checks the whole tree; call once after loading. + pub fn validate(&self) -> Result<(), RqeExprError> { + self.validate_node(true) + } + + fn validate_node(&self, is_root: bool) -> Result<(), RqeExprError> { + match self { + RqeExpr::Rqe { rqe } => rqe.validate(is_root), + RqeExpr::Agg { by, of, .. } => { + of.validate_node(false)?; + let child = label_set(of.output_labels()); + if !label_set(by).is_subset(&child) { + return Err(RqeExprError::AggByNotSubset { + by: by.clone(), + child: of.output_labels().to_vec(), + }); + } + Ok(()) + } + RqeExpr::Binary { lhs, rhs, .. } => { + lhs.validate_node(false)?; + rhs.validate_node(false)?; + match (lhs.is_scalar(), rhs.is_scalar()) { + (true, true) => Err(RqeExprError::BinaryBothScalar), + (false, false) + if label_set(lhs.output_labels()) != label_set(rhs.output_labels()) => + { + Err(RqeExprError::BinaryLabelMismatch { + lhs: lhs.output_labels().to_vec(), + rhs: rhs.output_labels().to_vec(), + }) + } + _ => Ok(()), + } + } + RqeExpr::Scalar { .. } => Ok(()), + } + } + + /// Labels of this node's result: leaf grouping, agg `by`, the non-scalar + /// side of a binary op (lhs if both), none for a scalar. + pub fn output_labels(&self) -> &[String] { + match self { + RqeExpr::Rqe { rqe } => &rqe.grouping, + RqeExpr::Agg { by, .. } => by, + RqeExpr::Binary { lhs, rhs, .. } => { + if lhs.is_scalar() { + rhs.output_labels() + } else { + lhs.output_labels() + } + } + RqeExpr::Scalar { .. } => &[], + } + } + + /// All RQE leaves, left to right. + pub fn leaves(&self) -> Vec<&RqeLeaf> { + let mut out = Vec::new(); + self.collect_leaves(&mut out); + out + } + + fn collect_leaves<'a>(&'a self, out: &mut Vec<&'a RqeLeaf>) { + match self { + RqeExpr::Rqe { rqe } => out.push(rqe), + RqeExpr::Agg { of, .. } => of.collect_leaves(out), + RqeExpr::Binary { lhs, rhs, .. } => { + lhs.collect_leaves(out); + rhs.collect_leaves(out); + } + RqeExpr::Scalar { .. } => {} + } + } + + fn is_scalar(&self) -> bool { + matches!(self, RqeExpr::Scalar { .. }) + } +} + +impl RqeLeaf { + fn validate(&self, is_root: bool) -> Result<(), RqeExprError> { + let id = || self.id.clone(); + match (self.capability, self.phi) { + (RqeCapability::Quantile, Some(phi)) if (0.0..=1.0).contains(&phi) => {} + (RqeCapability::Quantile, phi) => { + return Err(RqeExprError::InvalidPhi { id: id(), phi }) + } + (_, Some(_)) => return Err(RqeExprError::UnexpectedPhi { id: id() }), + (_, None) => {} + } + match (self.capability, self.k) { + (RqeCapability::TopkByValue, Some(k)) if k > 0 => {} + (RqeCapability::TopkByValue, k) => return Err(RqeExprError::InvalidK { id: id(), k }), + (_, Some(_)) => return Err(RqeExprError::UnexpectedK { id: id() }), + (_, None) => {} + } + if self.capability == RqeCapability::TopkByValue && !is_root { + return Err(RqeExprError::TopkNotAtRoot { id: id() }); + } + Ok(()) + } +} + +fn label_set(labels: &[String]) -> BTreeSet<&str> { + labels.iter().map(String::as_str).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::{json, Value}; + + fn leaf_json(id: &str, capability: &str, grouping: &[&str]) -> Value { + json!({"rqe": { + "id": id, "metric": "http_requests", "capability": capability, + "lookback_ms": 60000, "grouping": grouping, + }}) + } + + fn sum_leaf(id: &str, grouping: &[&str]) -> Value { + leaf_json(id, "sum", grouping) + } + + fn parse(v: Value) -> RqeExpr { + serde_json::from_value(v).expect("valid JSON shape") + } + + fn parse_err(v: Value) -> String { + serde_json::from_value::(v) + .expect_err("should not deserialize") + .to_string() + } + + fn assert_round_trip(v: Value) { + let expr = parse(v.clone()); + assert_eq!(serde_json::to_value(&expr).unwrap(), v); + } + + // --- JSON round trip --- + + #[test] + fn round_trip_leaf_minimal() { + assert_round_trip(sum_leaf("q0", &["job"])); + } + + #[test] + fn round_trip_leaf_all_optional_fields() { + assert_round_trip(json!({"rqe": { + "id": "q1", "metric": "events", "capability": "quantile", + "lookback_ms": 300000, "grouping": [], "phi": 0.99, "value": "latency", + }})); + assert_round_trip(json!({"rqe": { + "id": "q2", "metric": "events", "capability": "topk_by_value", + "lookback_ms": 1000, "grouping": ["user_id"], "k": 10, + }})); + assert_round_trip(leaf_json("q3", "cardinality", &["job"])); + } + + #[test] + fn round_trip_agg() { + for op in ["avg", "min", "max", "sum"] { + assert_round_trip( + json!({"agg": op, "by": ["job"], "of": sum_leaf("q", &["job", "instance"])}), + ); + } + } + + #[test] + fn round_trip_binary() { + for op in ["+", "-", "*", "/"] { + assert_round_trip( + json!({"op": op, "lhs": sum_leaf("a", &["job"]), "rhs": sum_leaf("b", &["job"])}), + ); + } + } + + #[test] + fn round_trip_scalar() { + assert_round_trip(json!({"scalar": 2.5})); + } + + #[test] + fn round_trip_nested() { + assert_round_trip(json!({"agg": "avg", "by": ["job"], "of": { + "op": "/", "lhs": sum_leaf("a", &["job", "instance"]), "rhs": sum_leaf("b", &["instance", "job"]), + }})); + } + + // --- deserialization errors --- + + #[test] + fn rejects_unknown_capability() { + for cap in ["rate", "topk_by_count"] { + let err = parse_err(leaf_json("q", cap, &[])); + assert!(err.contains(&format!("unknown variant `{cap}`")), "{err}"); + } + } + + #[test] + fn rejects_unknown_ops() { + let err = parse_err(json!({"agg": "count", "by": [], "of": sum_leaf("q", &[])})); + assert!(err.contains("unknown variant `count`"), "{err}"); + let err = parse_err(json!({"op": "%", "lhs": sum_leaf("a", &[]), "rhs": {"scalar": 1.0}})); + assert!(err.contains("unknown variant `%`"), "{err}"); + } + + #[test] + fn rejects_mixed_or_incomplete_node() { + let mixed = json!({"scalar": 1.0, "op": "+"}); + let incomplete = json!({"agg": "sum", "of": sum_leaf("q", &[])}); + for v in [mixed, incomplete, json!({})] { + assert!(parse_err(v).contains("must be exactly one of")); + } + } + + #[test] + fn rejects_unknown_keys() { + assert!(parse_err(json!({"scalar": 1.0, "extra": 1})).contains("unknown field `extra`")); + let mut leaf = sum_leaf("q", &[]); + leaf["rqe"]["filter"] = json!("x"); + assert!(parse_err(leaf).contains("unknown field `filter`")); + } + + // --- validation: phi / k --- + + #[test] + fn quantile_phi_rules() { + let with_phi = |phi: Value| { + let mut v = leaf_json("q", "quantile", &[]); + v["rqe"]["phi"] = phi; + parse(v).validate() + }; + assert_eq!(with_phi(json!(0.5)), Ok(())); + assert_eq!(with_phi(json!(0.0)), Ok(())); + assert_eq!(with_phi(json!(1.0)), Ok(())); + assert!(matches!( + with_phi(json!(1.5)), + Err(RqeExprError::InvalidPhi { .. }) + )); + assert!(matches!( + with_phi(json!(-0.1)), + Err(RqeExprError::InvalidPhi { .. }) + )); + assert!(matches!( + parse(leaf_json("q", "quantile", &[])).validate(), + Err(RqeExprError::InvalidPhi { phi: None, .. }) + )); + } + + #[test] + fn rejects_phi_on_non_quantile() { + let mut v = sum_leaf("q", &[]); + v["rqe"]["phi"] = json!(0.5); + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::UnexpectedPhi { .. }) + )); + } + + #[test] + fn topk_k_rules() { + let with_k = |k: Value| { + let mut v = leaf_json("q", "topk_by_value", &[]); + v["rqe"]["k"] = k; + parse(v).validate() + }; + assert_eq!(with_k(json!(5)), Ok(())); + assert!(matches!( + with_k(json!(0)), + Err(RqeExprError::InvalidK { k: Some(0), .. }) + )); + assert!(matches!( + parse(leaf_json("q", "topk_by_value", &[])).validate(), + Err(RqeExprError::InvalidK { k: None, .. }) + )); + } + + #[test] + fn rejects_k_on_non_topk() { + let mut v = leaf_json("q", "cardinality", &[]); + v["rqe"]["k"] = json!(3); + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::UnexpectedK { .. }) + )); + } + + // --- validation: top-k placement --- + + #[test] + fn rejects_topk_below_root() { + let mut topk = leaf_json("t", "topk_by_value", &["job"]); + topk["rqe"]["k"] = json!(3); + let agg = json!({"agg": "sum", "by": ["job"], "of": topk.clone()}); + let binary = json!({"op": "*", "lhs": topk, "rhs": {"scalar": 2.0}}); + for v in [agg, binary] { + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::TopkNotAtRoot { .. }) + )); + } + } + + // --- validation: agg --- + + #[test] + fn agg_by_subset_rules() { + let ok = json!({"agg": "max", "by": ["job"], "of": sum_leaf("q", &["instance", "job"])}); + assert_eq!(parse(ok).validate(), Ok(())); + let empty_by = json!({"agg": "sum", "by": [], "of": sum_leaf("q", &["job"])}); + assert_eq!(parse(empty_by).validate(), Ok(())); + let bad = json!({"agg": "max", "by": ["pod"], "of": sum_leaf("q", &["job"])}); + assert!(matches!( + parse(bad).validate(), + Err(RqeExprError::AggByNotSubset { .. }) + )); + } + + #[test] + fn agg_validates_child() { + let v = json!({"agg": "sum", "by": [], "of": leaf_json("q", "quantile", &[])}); + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::InvalidPhi { .. }) + )); + } + + // --- validation: binary --- + + #[test] + fn binary_label_set_rules() { + let same_set = json!({"op": "-", "lhs": sum_leaf("a", &["job", "instance"]), "rhs": sum_leaf("b", &["instance", "job"])}); + assert_eq!(parse(same_set).validate(), Ok(())); + let differ = + json!({"op": "-", "lhs": sum_leaf("a", &["job"]), "rhs": sum_leaf("b", &["instance"])}); + assert!(matches!( + parse(differ).validate(), + Err(RqeExprError::BinaryLabelMismatch { .. }) + )); + } + + #[test] + fn binary_scalar_operand_rules() { + let scalar_rhs = + json!({"op": "/", "lhs": sum_leaf("a", &["job"]), "rhs": {"scalar": 1000.0}}); + assert_eq!(parse(scalar_rhs).validate(), Ok(())); + let scalar_lhs = json!({"op": "-", "lhs": {"scalar": 1.0}, "rhs": sum_leaf("a", &["job"])}); + assert_eq!(parse(scalar_lhs).validate(), Ok(())); + let both = json!({"op": "+", "lhs": {"scalar": 1.0}, "rhs": {"scalar": 2.0}}); + assert_eq!(parse(both).validate(), Err(RqeExprError::BinaryBothScalar)); + } + + #[test] + fn binary_validates_children() { + let bad_child = json!({"agg": "sum", "by": ["pod"], "of": sum_leaf("q", &["job"])}); + let v = json!({"op": "+", "lhs": sum_leaf("a", &["pod"]), "rhs": bad_child}); + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::AggByNotSubset { .. }) + )); + } + + // --- nested example and helpers --- + + fn nested_example() -> RqeExpr { + parse(json!({"agg": "avg", "by": ["job"], "of": { + "op": "/", "lhs": sum_leaf("a", &["job", "instance"]), "rhs": sum_leaf("b", &["instance", "job"]), + }})) + } + + #[test] + fn nested_example_validates() { + assert_eq!(nested_example().validate(), Ok(())); + } + + #[test] + fn leaves_in_order() { + let ids: Vec<_> = nested_example() + .leaves() + .iter() + .map(|l| l.id.clone()) + .collect(); + assert_eq!(ids, ["a", "b"]); + assert!(parse(json!({"scalar": 1.0})).leaves().is_empty()); + } + + #[test] + fn output_labels_per_kind() { + let expr = nested_example(); + assert_eq!(expr.output_labels(), ["job"]); + let RqeExpr::Agg { of, .. } = &expr else { + unreachable!() + }; + assert_eq!(of.output_labels(), ["job", "instance"]); + let scalar_lhs = + parse(json!({"op": "*", "lhs": {"scalar": 2.0}, "rhs": sum_leaf("a", &["pod"])})); + assert_eq!(scalar_lhs.output_labels(), ["pod"]); + assert!(parse(json!({"scalar": 2.0})).output_labels().is_empty()); + } +} From 7c77405e4267b2a837a554b043843ca266a398eb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 20:16:43 -0400 Subject: [PATCH 2/4] feat(asap-common): per-bucket top-k (`topk_by`) on RQE leaves An RQE leaf may name `topk_by` labels: like PromQL `topk by (...)`, a top-k leaf then ranks its input independently within each value combination of those labels, while `grouping` stays its output labels. Absent, the whole input is one bucket. Validation allows `topk_by` only on `topk_by_value` leaves, non-empty and within `grouping`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011gKTCAEz3sJ1Hzde16xxU1 --- .../rs/asap_types/src/rqe_expr.rs | 98 +++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs index 05e6000d..d320909e 100644 --- a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs +++ b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs @@ -27,7 +27,13 @@ pub struct RqeLeaf { pub metric: String, pub capability: RqeCapability, pub lookback_ms: u64, + /// Output labels: one series per value combination. pub grouping: Vec, + /// Top-k buckets, as in PromQL `topk by (topk_by) (k, ...)`: the input is + /// ranked independently within each value combination of these labels, a + /// subset of `grouping`. Absent ranks the whole input as one bucket. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub topk_by: Option>, /// Quantile rank; set iff `capability` is `quantile`. #[serde(default, skip_serializing_if = "Option::is_none")] pub phi: Option, @@ -159,6 +165,16 @@ pub enum RqeExprError { InvalidK { id: String, k: Option }, #[error("RQE {id}: k is only allowed for topk_by_value")] UnexpectedK { id: String }, + #[error("RQE {id}: topk_by is only allowed for topk_by_value")] + UnexpectedTopkBy { id: String }, + #[error("RQE {id}: topk_by must name at least one label; omit it to rank as one bucket")] + EmptyTopkBy { id: String }, + #[error("RQE {id}: topk_by {topk_by:?} is not a subset of grouping {grouping:?}")] + TopkByNotSubset { + id: String, + topk_by: Vec, + grouping: Vec, + }, #[error("RQE {id}: topk_by_value is only allowed at the root")] TopkNotAtRoot { id: String }, #[error("agg by {by:?} is not a subset of child labels {child:?}")] @@ -267,6 +283,22 @@ impl RqeLeaf { (_, Some(_)) => return Err(RqeExprError::UnexpectedK { id: id() }), (_, None) => {} } + match (self.capability, &self.topk_by) { + (RqeCapability::TopkByValue, Some(topk_by)) if topk_by.is_empty() => { + return Err(RqeExprError::EmptyTopkBy { id: id() }) + } + (RqeCapability::TopkByValue, Some(topk_by)) + if !label_set(topk_by).is_subset(&label_set(&self.grouping)) => + { + return Err(RqeExprError::TopkByNotSubset { + id: id(), + topk_by: topk_by.clone(), + grouping: self.grouping.clone(), + }) + } + (RqeCapability::TopkByValue, _) | (_, None) => {} + (_, Some(_)) => return Err(RqeExprError::UnexpectedTopkBy { id: id() }), + } if self.capability == RqeCapability::TopkByValue && !is_root { return Err(RqeExprError::TopkNotAtRoot { id: id() }); } @@ -326,6 +358,11 @@ mod tests { "id": "q2", "metric": "events", "capability": "topk_by_value", "lookback_ms": 1000, "grouping": ["user_id"], "k": 10, }})); + assert_round_trip(json!({"rqe": { + "id": "q4", "metric": "data", "capability": "topk_by_value", + "lookback_ms": 60000, "grouping": ["region", "job"], "k": 4, + "topk_by": ["region"], + }})); assert_round_trip(leaf_json("q3", "cardinality", &["job"])); } @@ -458,6 +495,67 @@ mod tests { )); } + // --- validation: topk_by --- + + fn topk_by_leaf(grouping: &[&str], topk_by: Value) -> Value { + let mut v = leaf_json("t", "topk_by_value", grouping); + v["rqe"]["k"] = json!(4); + v["rqe"]["topk_by"] = topk_by; + v + } + + #[test] + fn topk_by_rules() { + let leaf = parse(topk_by_leaf(&["region", "job"], json!(["region"]))); + assert_eq!(leaf.validate(), Ok(())); + let RqeExpr::Rqe { rqe } = &leaf else { + unreachable!() + }; + assert_eq!(rqe.topk_by.as_deref(), Some(&["region".to_string()][..])); + // The output labels stay the leaf's grouping, not its buckets. + assert_eq!(leaf.output_labels(), ["region", "job"]); + // topk_by equal to grouping keeps one series per bucket. + let all = topk_by_leaf(&["region", "job"], json!(["job", "region"])); + assert_eq!(parse(all).validate(), Ok(())); + } + + #[test] + fn rejects_topk_by_outside_grouping() { + let v = topk_by_leaf(&["region", "job"], json!(["region", "pod"])); + assert!(matches!( + parse(v).validate(), + Err(RqeExprError::TopkByNotSubset { .. }) + )); + } + + #[test] + fn rejects_empty_topk_by() { + let v = topk_by_leaf(&["region"], json!([])); + assert_eq!( + parse(v).validate(), + Err(RqeExprError::EmptyTopkBy { id: "t".into() }) + ); + } + + #[test] + fn rejects_topk_by_on_non_topk() { + for capability in ["sum", "cardinality"] { + let mut v = leaf_json("q", capability, &["job"]); + v["rqe"]["topk_by"] = json!(["job"]); + assert_eq!( + parse(v).validate(), + Err(RqeExprError::UnexpectedTopkBy { id: "q".into() }) + ); + } + let mut quantile = leaf_json("q", "quantile", &["job"]); + quantile["rqe"]["phi"] = json!(0.5); + quantile["rqe"]["topk_by"] = json!(["job"]); + assert!(matches!( + parse(quantile).validate(), + Err(RqeExprError::UnexpectedTopkBy { .. }) + )); + } + // --- validation: top-k placement --- #[test] From 1b07c8364a6d0c8d7a90ca5f95c70f1d03e5a2e9 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 22:32:57 -0400 Subject: [PATCH 3/4] fix(asap-common): tighten RQE tree validation Require every subtree that feeds the result (root, agg operand, one side of each binary op) to read at least one RQE leaf; allow a repeated leaf id only for identical leaves; reject duplicate labels in grouping, by and topk_by; reject an empty id or metric and a zero lookback. Document that label lists are sets matched by name. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011gKTCAEz3sJ1Hzde16xxU1 --- .../rs/asap_types/src/rqe_expr.rs | 168 +++++++++++++++++- 1 file changed, 160 insertions(+), 8 deletions(-) diff --git a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs index d320909e..53907354 100644 --- a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs +++ b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs @@ -5,8 +5,12 @@ //! {"agg": "avg", "by": ["job"], "of": //! {"op": "/", "lhs": {"rqe": {...}}, "rhs": {"scalar": 2.0}}} //! ``` +//! +//! Label lists (`grouping`, `by`, `topk_by`) are sets: their order carries no +//! meaning and series are matched by label name, so an engine must sort or +//! match by name rather than pair values by position. -use std::collections::BTreeSet; +use std::collections::{BTreeSet, HashMap}; use serde::{Deserialize, Serialize}; @@ -179,8 +183,18 @@ pub enum RqeExprError { TopkNotAtRoot { id: String }, #[error("agg by {by:?} is not a subset of child labels {child:?}")] AggByNotSubset { by: Vec, child: Vec }, - #[error("binary op needs at least one non-scalar side")] - BinaryBothScalar, + #[error("RQE id must not be empty")] + EmptyId, + #[error("RQE {id}: metric must not be empty")] + EmptyMetric { id: String }, + #[error("RQE {id}: lookback_ms must be > 0")] + ZeroLookback { id: String }, + #[error("RQE {id} appears more than once with different leaves")] + ConflictingLeafId { id: String }, + #[error("{field} lists label {label:?} more than once")] + DuplicateLabel { field: &'static str, label: String }, + #[error("{node} reads no RQE leaf")] + NoLeaf { node: &'static str }, #[error("binary op sides have different labels: {lhs:?} vs {rhs:?}")] BinaryLabelMismatch { lhs: Vec, rhs: Vec }, } @@ -188,7 +202,20 @@ pub enum RqeExprError { impl RqeExpr { /// Checks the whole tree; call once after loading. pub fn validate(&self) -> Result<(), RqeExprError> { - self.validate_node(true) + self.validate_node(true)?; + if !self.has_leaf() { + return Err(RqeExprError::NoLeaf { node: "root" }); + } + // A leaf may be repeated (e.g. `x - x`) but an id names one leaf. + let mut by_id: HashMap<&str, &RqeLeaf> = HashMap::new(); + for leaf in self.leaves() { + if *by_id.entry(&leaf.id).or_insert(leaf) != leaf { + return Err(RqeExprError::ConflictingLeafId { + id: leaf.id.clone(), + }); + } + } + Ok(()) } fn validate_node(&self, is_root: bool) -> Result<(), RqeExprError> { @@ -196,6 +223,10 @@ impl RqeExpr { RqeExpr::Rqe { rqe } => rqe.validate(is_root), RqeExpr::Agg { by, of, .. } => { of.validate_node(false)?; + if !of.has_leaf() { + return Err(RqeExprError::NoLeaf { node: "agg" }); + } + check_unique_labels("agg by", by)?; let child = label_set(of.output_labels()); if !label_set(by).is_subset(&child) { return Err(RqeExprError::AggByNotSubset { @@ -208,9 +239,9 @@ impl RqeExpr { RqeExpr::Binary { lhs, rhs, .. } => { lhs.validate_node(false)?; rhs.validate_node(false)?; - match (lhs.is_scalar(), rhs.is_scalar()) { - (true, true) => Err(RqeExprError::BinaryBothScalar), - (false, false) + match (lhs.has_leaf(), rhs.has_leaf()) { + (false, false) => Err(RqeExprError::NoLeaf { node: "binary op" }), + (true, true) if label_set(lhs.output_labels()) != label_set(rhs.output_labels()) => { Err(RqeExprError::BinaryLabelMismatch { @@ -264,11 +295,33 @@ impl RqeExpr { fn is_scalar(&self) -> bool { matches!(self, RqeExpr::Scalar { .. }) } + + fn has_leaf(&self) -> bool { + match self { + RqeExpr::Rqe { .. } => true, + RqeExpr::Agg { of, .. } => of.has_leaf(), + RqeExpr::Binary { lhs, rhs, .. } => lhs.has_leaf() || rhs.has_leaf(), + RqeExpr::Scalar { .. } => false, + } + } } impl RqeLeaf { fn validate(&self, is_root: bool) -> Result<(), RqeExprError> { let id = || self.id.clone(); + if self.id.is_empty() { + return Err(RqeExprError::EmptyId); + } + if self.metric.is_empty() { + return Err(RqeExprError::EmptyMetric { id: id() }); + } + if self.lookback_ms == 0 { + return Err(RqeExprError::ZeroLookback { id: id() }); + } + check_unique_labels("grouping", &self.grouping)?; + if let Some(topk_by) = &self.topk_by { + check_unique_labels("topk_by", topk_by)?; + } match (self.capability, self.phi) { (RqeCapability::Quantile, Some(phi)) if (0.0..=1.0).contains(&phi) => {} (RqeCapability::Quantile, phi) => { @@ -310,6 +363,17 @@ fn label_set(labels: &[String]) -> BTreeSet<&str> { labels.iter().map(String::as_str).collect() } +fn check_unique_labels(field: &'static str, labels: &[String]) -> Result<(), RqeExprError> { + let mut seen = BTreeSet::new(); + match labels.iter().find(|l| !seen.insert(l.as_str())) { + Some(label) => Err(RqeExprError::DuplicateLabel { + field, + label: label.clone(), + }), + None => Ok(()), + } +} + #[cfg(test)] mod tests { use super::*; @@ -618,7 +682,95 @@ mod tests { let scalar_lhs = json!({"op": "-", "lhs": {"scalar": 1.0}, "rhs": sum_leaf("a", &["job"])}); assert_eq!(parse(scalar_lhs).validate(), Ok(())); let both = json!({"op": "+", "lhs": {"scalar": 1.0}, "rhs": {"scalar": 2.0}}); - assert_eq!(parse(both).validate(), Err(RqeExprError::BinaryBothScalar)); + let node = "binary op"; + assert_eq!(parse(both).validate(), Err(RqeExprError::NoLeaf { node })); + } + + // --- validation: every subtree reads a leaf --- + + #[test] + fn rejects_subtrees_without_leaves() { + let scalar_agg = |x: f64| json!({"agg": "sum", "by": [], "of": {"scalar": x}}); + // Neither side is a literal scalar, yet neither reads a leaf. + let binary = json!({"op": "+", "lhs": scalar_agg(1.0), "rhs": scalar_agg(2.0)}); + let nested_binary = json!({"op": "*", + "lhs": {"op": "+", "lhs": {"scalar": 1.0}, "rhs": {"scalar": 2.0}}, + "rhs": sum_leaf("a", &[])}); + for (v, node) in [ + (json!({"scalar": 1.0}), "root"), + (scalar_agg(1.0), "agg"), + (binary, "agg"), + (nested_binary, "binary op"), + ] { + assert_eq!(parse(v).validate(), Err(RqeExprError::NoLeaf { node })); + } + // A side that reads a leaf through an agg is not a scalar side. + let agg_side = json!({"op": "+", + "lhs": {"agg": "sum", "by": [], "of": sum_leaf("a", &["job"])}, + "rhs": {"scalar": 1.0}}); + assert_eq!(parse(agg_side).validate(), Ok(())); + } + + // --- validation: leaf ids --- + + #[test] + fn repeated_identical_leaf_is_allowed() { + let v = json!({"op": "-", "lhs": sum_leaf("x", &["job"]), "rhs": sum_leaf("x", &["job"])}); + let expr = parse(v); + assert_eq!(expr.validate(), Ok(())); + assert_eq!(expr.leaves().len(), 2); + } + + #[test] + fn rejects_one_id_for_different_leaves() { + let mut other = sum_leaf("x", &["job"]); + other["rqe"]["metric"] = json!("other_metric"); + let v = json!({"op": "/", "lhs": sum_leaf("x", &["job"]), "rhs": other}); + assert_eq!( + parse(v).validate(), + Err(RqeExprError::ConflictingLeafId { id: "x".into() }) + ); + } + + // --- validation: leaf fields --- + + #[test] + fn rejects_empty_or_zero_leaf_fields() { + let with = |key: &str, value: Value| { + let mut v = sum_leaf("q", &[]); + v["rqe"][key] = value; + parse(v).validate() + }; + assert_eq!(with("id", json!("")), Err(RqeExprError::EmptyId)); + assert_eq!( + with("metric", json!("")), + Err(RqeExprError::EmptyMetric { id: "q".into() }) + ); + assert_eq!( + with("lookback_ms", json!(0)), + Err(RqeExprError::ZeroLookback { id: "q".into() }) + ); + } + + // --- validation: duplicate labels --- + + #[test] + fn rejects_duplicate_labels() { + let dup = |field: &'static str| { + Err(RqeExprError::DuplicateLabel { + field, + label: "job".into(), + }) + }; + let grouping = sum_leaf("q", &["job", "instance", "job"]); + assert_eq!(parse(grouping).validate(), dup("grouping")); + let by = json!({"agg": "sum", "by": ["job", "job"], "of": sum_leaf("q", &["job"])}); + assert_eq!(parse(by).validate(), dup("agg by")); + let topk_by = topk_by_leaf(&["job", "region"], json!(["job", "job"])); + assert_eq!(parse(topk_by).validate(), dup("topk_by")); + // A duplicated grouping no longer slips past the binary label-set check. + let binary = json!({"op": "+", "lhs": sum_leaf("a", &["job", "job"]), "rhs": sum_leaf("b", &["job"])}); + assert_eq!(parse(binary).validate(), dup("grouping")); } #[test] From 60a78443d69dcd3f08ee5a114057babcf875ffe7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:23:37 -0400 Subject: [PATCH 4/4] style(asap-common): state behaviour in duplicate-label test comment Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011gKTCAEz3sJ1Hzde16xxU1 --- asap-common/dependencies/rs/asap_types/src/rqe_expr.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs index 53907354..22d91b52 100644 --- a/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs +++ b/asap-common/dependencies/rs/asap_types/src/rqe_expr.rs @@ -768,7 +768,7 @@ mod tests { assert_eq!(parse(by).validate(), dup("agg by")); let topk_by = topk_by_leaf(&["job", "region"], json!(["job", "job"])); assert_eq!(parse(topk_by).validate(), dup("topk_by")); - // A duplicated grouping no longer slips past the binary label-set check. + // A duplicated grouping fails the binary label-set check. let binary = json!({"op": "+", "lhs": sum_leaf("a", &["job", "job"]), "rhs": sum_leaf("b", &["job"])}); assert_eq!(parse(binary).validate(), dup("grouping")); }