diff --git a/crates/asap-aware-mapping/src/analytical_cost.rs b/crates/asap-aware-mapping/src/analytical_cost.rs index ed2a316d..bdbfb795 100644 --- a/crates/asap-aware-mapping/src/analytical_cost.rs +++ b/crates/asap-aware-mapping/src/analytical_cost.rs @@ -2193,6 +2193,8 @@ pub enum AnalyticalCostError { UnsupportedCandidate, #[error("query operator has no physical implementation in the analytical model")] UnsupportedQueryOperator, + #[error("multi-measure per-entity aggregates have no physical implementation; lower each measure separately")] + UnsupportedMultiMeasurePerEntity, #[error("inconsistent operator statistics: {0}")] InconsistentOperatorStatistics(&'static str), #[error("summary operation {0} has no lifecycle-aware cost formula")] diff --git a/crates/asap-aware-mapping/src/query_physical_lowering.rs b/crates/asap-aware-mapping/src/query_physical_lowering.rs index 39d04ced..7ba41831 100644 --- a/crates/asap-aware-mapping/src/query_physical_lowering.rs +++ b/crates/asap-aware-mapping/src/query_physical_lowering.rs @@ -292,7 +292,7 @@ pub fn lower_query_physical_dag( } if matches!(reduction, asap_types::pre_asap::Reduction::PerEntity) { if measures.len() != 1 { - return Err(AnalyticalCostError::UnsupportedQueryOperator); + return Err(AnalyticalCostError::UnsupportedMultiMeasurePerEntity); } let accumulator_count = u64::try_from(measures.len()) .map_err(|_| AnalyticalCostError::Overflow)?; @@ -2486,6 +2486,42 @@ mod tests { )); } + /// Multi-measure schemas must not silently lower to a single accumulator. + #[test] + fn multi_measure_per_entity_lowering_is_explicitly_unsupported() { + use asap_types::pre_asap::{ + AggIntent, Column, DataType, QueryExpr, Reduction, Schema, Source, + }; + let source = Source::TimeSeries { + metric: "requests".into(), + }; + let root = Rc::new(QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![ + AggIntent::Sum { col: None }, + AggIntent::Count { + accuracy: asap_types::types::AccuracyTarget::Exact, + }, + ], + output_names: vec![], + having: None, + child: Rc::new(QueryExpr::Scan { + source: source.clone(), + predicates: vec![], + schema: Schema::new(vec![Column::new("value", DataType::Float64, false)]), + }), + }); + let provided = HashMap::new(); + assert!(matches!( + lower_query_physical_dag( + &root, + &scope(vec![coverage(source, vec![])]), + &scripted(&provided) + ), + Err(AnalyticalCostError::UnsupportedMultiMeasurePerEntity) + )); + } + #[test] fn promql_relabel_sample_and_per_series_lower_as_a_complete_chain() { use asap_types::pre_asap::{ diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index 459f2e8a..7309ec46 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -26,33 +26,15 @@ //! that reshaping — see "Non-goals" below for why it does not also decide //! whether the reshaping is worth it. //! -//! ## Scope: `by(...)` grouping only (issue #253's own scope note) +//! ## Scope //! -//! [`AvgToSumOverCountStrategy::matches`] additionally requires -//! `Reduction::Reduce(by)` with `by` an ordinary (non-`without`) grouping — -//! narrower than [`SketchAlgorithmStrategy`]'s `bindable_intent`, which is -//! `Reduction`-agnostic. Two concrete reasons, not stylistic ones: -//! -//! - **`Reduction::PerEntity`** (`rate`/`increase`/`*_over_time`) is -//! single-measure by construction — -//! [`aggregate_output_schema`](asap_types::pre_asap::query_expr::aggregate_output_schema) -//! `debug_assert!`s exactly one measure for it. This rewrite's entire -//! point is introducing a *second* measure (`Count` alongside `Sum`) -//! under the same node, which would violate that invariant outright, not -//! just drift a schema detail. -//! - **`without(...)` grouping** leaves an `Aggregate`'s own output schema -//! *open* (`closed: false`, see `without_output_schema`), while the -//! `Project` this strategy always wraps the rewrite in forces -//! `closed: true` (see `QueryExpr::output_schema`'s `Project` arm). Under -//! `without(...)` the rewritten form's `closed` flag would silently flip -//! relative to the original — exactly the kind of schema drift this -//! module exists to avoid. -//! -//! Both are follow-ups (issue #253 itself scopes to "the concrete case in -//! Peilin's comment"), not correctness bugs in what ships here — a node -//! outside this scope simply doesn't `match`, the same "safe but -//! uninformative" fallback [`SketchAlgorithmStrategy`]/[`SharedSubtreeStrategy`] -//! already use for shapes they don't have an opinion on. +//! Ordinary `by(...)` averages with non-null inputs are eligible. +//! Per-entity range averages are excluded: even finite Float64 samples can +//! overflow SUM while AVG remains finite. A future decomposition needs a +//! proven arithmetic domain or an overflow-safe execution path; schema +//! compatibility alone cannot establish semantic equivalence. +//! `without(...)` remains excluded because the grouped cast projection closes +//! its input schema. Nullable inputs are excluded because Count means COUNT(*). //! //! ## Non-goals (mirrors [`replacement`]'s own discipline) //! @@ -76,9 +58,8 @@ use asap_types::types::AccuracyTarget; use crate::replacement::{Replacement, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG}; /// The shape [`AvgToSumOverCountStrategy`] rewrites: a single `Avg{col}` -/// measure, no `HAVING`, grouped with an ordinary `by(...)` reduction (see -/// the module docs' "Scope" for why `without(...)`/`PerEntity` are -/// excluded). Returns the grouping key count and the summed column so +/// measure, no `HAVING`, with ordinary grouping. +/// Returns the grouping key count and summed column so /// [`build_rewrite`] doesn't have to re-match. fn avg_rewrite_target(node: &QueryExpr) -> Option<(usize, Option)> { let QueryExpr::Aggregate { @@ -91,12 +72,10 @@ fn avg_rewrite_target(node: &QueryExpr) -> Option<(usize, Option)> { else { return None; }; - let Reduction::Reduce(by) = reduction else { - return None; + let by = match reduction { + Reduction::Reduce(by) if !by.is_without() => by, + _ => return None, }; - if by.is_without() { - return None; - } let [AggIntent::Avg { col }] = measures.as_slice() else { return None; }; @@ -114,12 +93,12 @@ fn avg_rewrite_target(node: &QueryExpr) -> Option<(usize, Option)> { Some((by.keys().len(), *col)) } -/// Build the rewritten `Project{ cast(sum) } / Aggregate{ Count }` tree for +/// Build `Sum / Count` (with a cast projection for grouped aggregates) for /// `root`, or `None` if `root` isn't [`avg_rewrite_target`]'s shape. `Sum` and /// `Count` deliberately live in separate, single-measure aggregates so the /// replacement fixpoint discovers each as an independently bindable target. /// -/// The `Project`'s leading `by.len()` items are bare `Column(i)` +/// For grouped aggregates, the `Project`'s leading `by.len()` items are bare `Column(i)` /// pass-throughs of the grouping keys — identical in name/type to the /// original `Avg` aggregate's own leading columns, since both aggregates /// share the same `reduction`/`child` and only differ in `measures` @@ -463,18 +442,42 @@ mod tests { assert!(AvgToSumOverCountStrategy.replacements(&target).is_empty()); } + /// Matching schemas do not make range SUM/COUNT safe for unbounded samples. #[test] - fn does_not_match_a_per_entity_avg_aggregate() { + fn per_entity_avg_rewrite_is_rejected_without_arithmetic_proof() { let q = Rc::new(QueryExpr::Aggregate { reduction: Reduction::PerEntity, measures: vec![AggIntent::Avg { col: None }], output_names: vec![], having: None, - child: Rc::new(metric_scan(&[])), + child: Rc::new(QueryExpr::TimeRange { + range: Duration::from_secs(300), + child: Rc::new(metric_scan(&["job", "instance"])), + }), }); let target = TargetSubDAG::new(&q); assert!(!AvgToSumOverCountStrategy.matches(&target)); assert!(AvgToSumOverCountStrategy.replacements(&target).is_empty()); + assert!(build_rewrite(&q).is_none()); + } + + /// COUNT(*) cannot replace the denominator of a nullable sample average. + #[test] + fn per_entity_nullable_or_non_sample_average_is_not_rewritten() { + for (nullable, column) in [(true, None), (false, Some(2)), (false, Some(99))] { + let mut scan = metric_scan(&["job"]); + if let QueryExpr::Scan { schema, .. } = &mut scan { + schema.columns[1].nullable = nullable; + } + let root = Rc::new(QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![AggIntent::Avg { col: column }], + output_names: vec![], + having: None, + child: Rc::new(scan), + }); + assert!(build_rewrite(&root).is_none()); + } } #[test] diff --git a/crates/integration-tests/tests/avg_over_time_rewrite.rs b/crates/integration-tests/tests/avg_over_time_rewrite.rs new file mode 100644 index 00000000..713fb769 --- /dev/null +++ b/crates/integration-tests/tests/avg_over_time_rewrite.rs @@ -0,0 +1,28 @@ +use std::rc::Rc; + +use asap_aware_mapping::replacement::{ReplacementStrategy, TargetSubDAG}; +use asap_aware_mapping::rewrite::AvgToSumOverCountStrategy; +use asap_frontend_promql::lower_promql; +use asap_types::types::AccuracyTarget; + +/// Unbounded Float64 ranges must retain AVG: finite samples can overflow SUM. +/// The independent Prometheus oracle is fixtures/avg_over_time_overflow.test.yml. +#[test] +fn range_average_does_not_offer_unconditional_sum_count() { + for query in [ + "avg_over_time(latency[5m])", + "avg_over_time(latency{job=\"api\"}[5m])", + "avg_over_time(latency[5m:1m])", + ] { + let root = Rc::new(lower_promql(query, AccuracyTarget::Exact).unwrap()); + let target = TargetSubDAG::new(&root); + assert!( + !AvgToSumOverCountStrategy.matches(&target), + "unbounded average must not match: {query}" + ); + assert!( + AvgToSumOverCountStrategy.replacements(&target).is_empty(), + "unbounded average must not produce a sum/count rewrite: {query}" + ); + } +} diff --git a/crates/integration-tests/tests/fixtures/avg_over_time_overflow.test.yml b/crates/integration-tests/tests/fixtures/avg_over_time_overflow.test.yml new file mode 100644 index 00000000..48918d74 --- /dev/null +++ b/crates/integration-tests/tests/fixtures/avg_over_time_overflow.test.yml @@ -0,0 +1,34 @@ +# Official Prometheus 3.5.0 oracle: AVG remains finite when SUM overflows. +# Run: promtool test rules crates/integration-tests/tests/fixtures/avg_over_time_overflow.test.yml +evaluation_interval: 1m +tests: +- interval: 1m + input_series: + - series: 'latency{job="api"}' + values: '1e308 1e308' + promql_expr_test: + - expr: 'avg_over_time(latency[5m])' + eval_time: 1m + exp_samples: + - labels: '{job="api"}' + value: 1e308 + - expr: 'sum_over_time(latency[5m]) / count_over_time(latency[5m])' + eval_time: 1m + exp_samples: + - labels: '{job="api"}' + value: .inf +- interval: 1m + input_series: + - series: 'latency{job="api"}' + values: '-1e308 -1e308' + promql_expr_test: + - expr: 'avg_over_time(latency[5m])' + eval_time: 1m + exp_samples: + - labels: '{job="api"}' + value: -1e308 + - expr: 'sum_over_time(latency[5m]) / count_over_time(latency[5m])' + eval_time: 1m + exp_samples: + - labels: '{job="api"}' + value: -.inf diff --git a/crates/types/src/pre_asap/query_expr.rs b/crates/types/src/pre_asap/query_expr.rs index dbc14b48..22eaaaf3 100644 --- a/crates/types/src/pre_asap/query_expr.rs +++ b/crates/types/src/pre_asap/query_expr.rs @@ -52,6 +52,8 @@ impl ColState for ColumnRef { /// Errors from schema derivation over a canonical tree. #[derive(Debug, Error)] pub enum QueryExprError { + #[error("invalid per-entity aggregate: {0}")] + InvalidPerEntityAggregate(String), #[error("invalid scalar function signature: {0}")] InvalidScalarSignature(String), #[error("by-column id {0} out of range (input has {1} columns)")] @@ -1484,8 +1486,14 @@ fn per_series_reduction_schema(input: &Schema, agg: &AggIntent) -> Schema { /// /// `Reduction::PerEntity` selects the label-preserving /// [`per_series_reduction_schema`] (`rate`/`increase`/`*_over_time`) instead -/// of the cross-series `by ++ measures` shape. Which one applies is read directly -/// off `reduction` — decided once, at construction, by whoever built the +/// of the cross-series `by ++ measures` shape. Multiple measures replace the +/// value slot with the first measure and append the rest, preserving label and +/// timestamp positions. Names come from output overrides or the intents and +/// must be distinct from one another and retained columns; all samples are Float64. +/// Frontends resolving named multi-measure outputs should supply a scan schema: +/// usage-derived binding may otherwise infer those names as input labels, which +/// this validation rejects as collisions. +/// Which shape applies is read directly off `reduction` — decided once, at construction, by whoever built the /// `Aggregate` node (issue #165) — not re-derived here from `by`/child shape. pub fn aggregate_output_schema( in_schema: &Schema, @@ -1495,12 +1503,57 @@ pub fn aggregate_output_schema( ) -> Result { let by = match reduction { Reduction::PerEntity => { - debug_assert_eq!( - measures.len(), - 1, - "a per-entity reduction is single-aggregate" - ); - return Ok(per_series_reduction_schema(in_schema, &measures[0])); + if measures.is_empty() { + return Err(QueryExprError::InvalidPerEntityAggregate( + "at least one measure is required".into(), + )); + } + if measures.len() == 1 { + return Ok(per_series_reduction_schema(in_schema, &measures[0])); + } + let invalid = |message: &str| QueryExprError::InvalidPerEntityAggregate(message.into()); + let vi = in_schema + .column_id("value") + .ok_or_else(|| invalid("multiple measures require an input value column"))?; + if in_schema.time_index == Some(vi) { + return Err(invalid("the value column cannot be the timestamp")); + } + if output_names.len() > measures.len() { + return Err(invalid("more output names than measures")); + } + let mut output = in_schema.clone(); + let mut names: std::collections::HashSet = in_schema + .columns + .iter() + .enumerate() + .filter(|(i, _)| *i != vi) + .map(|(_, c)| c.name.clone()) + .collect(); + for (i, measure) in measures.iter().enumerate() { + if matches!(measure, AggIntent::CountValues { .. }) { + return Err(invalid("count_values changes series identity and cannot be combined with other measures")); + } + let input = in_schema + .columns + .get(measure.input_col().unwrap_or(vi)) + .ok_or_else(|| invalid("measure input column is out of range"))?; + let mut column = measure.output_column(input); + if let Some(name) = output_names.get(i).filter(|n| !n.is_empty()) { + column.name = name.clone(); + } + if !names.insert(column.name.clone()) { + return Err(invalid("measure names must be unique and must not collide with labels or timestamp")); + } + column.dtype = DataType::Float64; + if i == 0 { + output.columns[vi] = column; + } else { + output.columns.push(column); + } + } + // A computed sample cannot retain a uniqueness proof about raw values. + output.unique_keys.retain(|key| !key.contains(&vi)); + return Ok(output); } Reduction::Reduce(by) => by, }; @@ -2196,6 +2249,80 @@ mod tests { assert_eq!(back, s); } + /// Multiple measures retain series metadata and expose separate named float samples. + #[test] + fn per_entity_multiple_measures_preserve_schema() { + let input = Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("job", DataType::Utf8, true), + ], + 0, + vec![vec![0, 2]], + ); + let measures = vec![ + AggIntent::Sum { col: None }, + AggIntent::Count { + accuracy: crate::types::AccuracyTarget::Exact, + }, + ]; + let output = + aggregate_output_schema(&input, &Reduction::PerEntity, &measures, &[]).unwrap(); + assert_eq!( + output + .columns + .iter() + .map(|c| c.name.as_str()) + .collect::>(), + vec!["ts", "sum", "job", "count"] + ); + assert_eq!(output.time_index, input.time_index); + assert_eq!(output.unique_keys, input.unique_keys); + assert_eq!(output.closed, input.closed); + assert_eq!(output.columns[1].dtype, DataType::Float64); + assert_eq!(output.columns[3].dtype, DataType::Float64); + } + + /// Invalid or ambiguous multi-measure shapes fail instead of losing outputs. + #[test] + fn per_entity_measure_names_are_validated() { + let input = Schema::new(vec![ + col("value", DataType::Float64, false), + col("job", DataType::Utf8, true), + ]); + let measures = vec![AggIntent::Sum { col: None }, AggIntent::Sum { col: None }]; + for names in [ + vec![], + vec!["a".into(), "a".into()], + vec!["job".into(), "b".into()], + ] { + assert!( + aggregate_output_schema(&input, &Reduction::PerEntity, &measures, &names).is_err() + ); + } + let output = aggregate_output_schema( + &input, + &Reduction::PerEntity, + &measures, + &["first".into(), "second".into()], + ) + .unwrap(); + assert_eq!(output.column_id("first"), Some(0)); + assert_eq!(output.column_id("second"), Some(2)); + assert!(aggregate_output_schema(&input, &Reduction::PerEntity, &[], &[]).is_err()); + assert!(aggregate_output_schema( + &input, + &Reduction::PerEntity, + &[ + AggIntent::Sum { col: Some(99) }, + AggIntent::Sum { col: None } + ], + &[] + ) + .is_err()); + } + #[test] fn per_series_rate_preserves_labels() { // A per-series range reduction (`rate`) is label-preserving: it produces diff --git a/crates/types/src/pre_asap/resolve.rs b/crates/types/src/pre_asap/resolve.rs index bfd0f1bf..3a5c31d8 100644 --- a/crates/types/src/pre_asap/resolve.rs +++ b/crates/types/src/pre_asap/resolve.rs @@ -609,6 +609,65 @@ mod tests { BinaryOpKind, QueryExpr, Source, VectorMatch, VectorMatchKind, }; + /// Parent expressions can resolve every named per-entity measure after binding. + #[test] + fn resolves_multiple_per_entity_measure_outputs() { + use crate::pre_asap::query_expr::ProjectItem; + let aggregate = QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![ + AggIntent::Sum { + col: Some(ColumnRef::SampleValue), + }, + AggIntent::Count { + accuracy: crate::types::AccuracyTarget::Exact, + }, + ], + output_names: vec!["total".into(), "samples".into()], + having: None, + child: Rc::new(QueryExpr::Scan { + source: Source::TimeSeries { + metric: "latency".into(), + }, + predicates: vec![], + schema: Some(crate::pre_asap::schema::Schema::new(vec![ + crate::pre_asap::schema::Column::new( + "value", + crate::pre_asap::schema::DataType::Float64, + false, + ), + ])), + }), + }; + let query = QueryExpr::Project { + cols: vec![ + ProjectItem { + alias: None, + expr: QueryExpr::Column(ColumnRef::Named("total".into())), + }, + ProjectItem { + alias: None, + expr: QueryExpr::Column(ColumnRef::Named("samples".into())), + }, + ], + qualifier: None, + child: Rc::new(aggregate), + }; + let output = resolve_root(&query).unwrap().output_schema().unwrap(); + assert_eq!( + output + .columns + .iter() + .map(|c| c.name.as_str()) + .collect::>(), + vec!["total", "samples"] + ); + assert!(output + .columns + .iter() + .all(|c| c.dtype == crate::pre_asap::schema::DataType::Float64)); + } + /// `resolve_root` over a `BinaryOp { , PromqlScalarBridge, vector_match }` /// (issue #220): the bridged scalar operand resolves through the same /// generic walk as every other node (its `Literal` child has no