From 1f1f0993ce0b273ed3bbf2ef6d214aa617e8978a Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 18 Sep 2026 16:33:54 +0000 Subject: [PATCH] fix(promql): preserve ingestion horizons through lowering and post-ASAP IR --- crates/asap-aware-mapping/src/lib.rs | 2 + .../src/maintained_population.rs | 24 +- crates/asap-aware-mapping/src/replacement.rs | 11 +- crates/asap-aware-mapping/src/rewrite.rs | 13 +- .../src/summary_maintenance_cost/model.rs | 10 +- .../src/summary_maintenance_lifecycle.rs | 10 +- crates/asap-aware-mapping/src/test_support.rs | 37 ++++ .../devtools/examples/canonical_examples.rs | 4 +- crates/devtools/examples/topk_ir.rs | 4 +- crates/devtools/src/bin/analyze_corpora.rs | 27 ++- crates/devtools/src/bin/dag_export.rs | 27 ++- crates/devtools/src/bin/show_post_asap_ir.rs | 21 +- crates/devtools/src/bin/show_pre_asap_ir.rs | 21 +- crates/devtools/src/bin/sketch_coverage.rs | 19 +- crates/devtools/src/bin/variant_coverage.rs | 19 +- crates/devtools/src/lib.rs | 50 ++++- crates/devtools/tests/cross_language.rs | 4 +- crates/frontend-promql/src/error.rs | 10 + crates/frontend-promql/src/lib.rs | 207 ++++++++++++++---- crates/frontend-promql/src/promql.rs | 116 ++++++---- .../frontend-promql/tests/count_planning.rs | 8 +- .../tests/histogram_metadata.rs | 6 +- .../tests/maintained_population_horizon.rs | 37 ++++ .../awesome_prometheus_alerts.rs | 16 +- .../observability/metrics_observability.rs | 5 +- .../tests/observability/o11y_bench_promql.rs | 5 +- .../tests/observability/promql_corpus.rs | 5 +- .../tests/promql_binding_regressions.rs | 4 +- .../tests/promql_conformance.rs | 44 +++- .../tests/promql_equivalence.rs | 3 +- .../frontend-promql/tests/promql_lowering.rs | 117 ++++++---- crates/frontend-promql/tests/support.rs | 52 +++++ .../tests/univmon_candidates.rs | 3 +- crates/integration-tests/src/lib.rs | 41 ++++ crates/integration-tests/tests/aggregate.rs | 8 +- crates/integration-tests/tests/binary_op.rs | 11 +- crates/integration-tests/tests/cse.rs | 2 +- .../tests/exact_composition.rs | 2 +- .../tests/frontend_timestamps.rs | 2 +- crates/integration-tests/tests/nested.rs | 66 ++++-- .../tests/promql_numeric_regressions.rs | 2 +- .../tests/promql_to_post_asap.rs | 2 +- crates/integration-tests/tests/scan.rs | 32 ++- crates/integration-tests/tests/schema.rs | 2 +- .../summary_maintenance_lifecycle_e2e.rs | 20 +- crates/integration-tests/tests/time_range.rs | 2 +- .../src/post_asap/maintained_population.rs | 15 +- crates/types/src/workload.rs | 25 +++ crates/types/tests/workload_compatibility.rs | 35 +++ docs/develop_docs/library-api.md | 154 +++++++++++-- docs/user_guide_docs/run-a-query.md | 33 +-- 51 files changed, 1102 insertions(+), 293 deletions(-) create mode 100644 crates/asap-aware-mapping/src/test_support.rs create mode 100644 crates/frontend-promql/tests/maintained_population_horizon.rs create mode 100644 crates/frontend-promql/tests/support.rs create mode 100644 crates/types/tests/workload_compatibility.rs diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index b3262c28..df392483 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -206,6 +206,8 @@ pub mod storage_io; pub mod summary_maintenance_cost; pub mod summary_maintenance_dag_export; pub mod summary_maintenance_lifecycle; +#[cfg(test)] +mod test_support; pub mod topk_reuse; pub use accuracy::{ diff --git a/crates/asap-aware-mapping/src/maintained_population.rs b/crates/asap-aware-mapping/src/maintained_population.rs index ba9de4f6..81bada46 100644 --- a/crates/asap-aware-mapping/src/maintained_population.rs +++ b/crates/asap-aware-mapping/src/maintained_population.rs @@ -118,11 +118,24 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou } return Some((population, readout, Rc::clone(source))); } + // A bare PromQL selector carries the declared ingestion interval as a + // temporal input scope. Membership must expire at that horizon; retain + // the wrapper as the maintained input so validation can check agreement. + let (series_source, lookback_ms) = match source.as_ref() { + QueryExpr::TimeRange { range, child } => { + let ms = u64::try_from(range.as_millis()).ok()?; + if ms == 0 || std::time::Duration::from_millis(ms) != *range { + return None; + } + (child.as_ref(), ms) + } + other => (other, 300_000), + }; let QueryExpr::Scan { source: Source::TimeSeries { metric }, predicates, schema, - } = source.as_ref() + } = series_source else { return None; }; @@ -176,7 +189,7 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou matchers, grouping: labels, without: grouping.is_without(), - lookback_ms: 300_000, + lookback_ms, }), max_k: 0, quantiles: false, @@ -285,12 +298,11 @@ impl ReplacementStrategy for MaintainedPopulationStrategy { #[cfg(test)] mod tests { use super::*; + use crate::test_support::lower_promql; use asap_types::post_asap::{compile_executable_dag, share_common_summary_subtrees}; + fn lower(q: &str) -> Rc { - Rc::new( - asap_frontend_promql::lower_promql(q, asap_types::types::AccuracyTarget::Exact) - .unwrap(), - ) + Rc::new(lower_promql(q, asap_types::types::AccuracyTarget::Exact)) } // Instant scalar aggregations share the same retractable series population. diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 9777cf2e..e705b4fb 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -5779,6 +5779,7 @@ mod tests { use super::*; use crate::accuracy::PropagationStats; use crate::cost_model::Cost; + use crate::test_support::lower_promql; use asap_types::pre_asap::agg_intent::{ agg_is_exact, default_cardinality, default_quantile, MathFunc, TimeFunc, }; @@ -5798,10 +5799,7 @@ mod tests { // Finite samples can overflow a sum although their native average is finite. #[test] fn temporal_average_requires_finite_division_guard() { - let root = Rc::new( - asap_frontend_promql::lower_promql("avg_over_time(a[5m])", AccuracyTarget::Exact) - .unwrap(), - ); + let root = Rc::new(lower_promql("avg_over_time(a[5m])", AccuracyTarget::Exact)); let candidates = SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root)); let operator = candidates @@ -5830,8 +5828,7 @@ mod tests { "topk(5, sum_over_time(a[5m]))", "topk by(job)(5, count_over_time(a[5m]))", ] { - let root = - Rc::new(asap_frontend_promql::lower_promql(query, AccuracyTarget::Exact).unwrap()); + let root = Rc::new(lower_promql(query, AccuracyTarget::Exact)); let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel); let node = exact_topk_over_temporal_values(&root, models) .unwrap() @@ -5859,7 +5856,7 @@ mod tests { "quantile_over_time(0.5,a[5m]) / quantile_over_time(0.9,a[5m])", "avg_over_time(a[5m]) / quantile_over_time(0.5,a[5m])", ] { - let root = Rc::new(asap_frontend_promql::lower_promql(query, target.clone()).unwrap()); + let root = Rc::new(lower_promql(query, target.clone())); let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel); assert!(realize_binary(&root, models, Some(&target)) .unwrap() diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index 1c481d86..2f2e4297 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -389,8 +389,10 @@ impl ReplacementStrategy for SemanticEquivalentRewriteStrategy { #[cfg(test)] mod tests { use super::*; + use crate::test_support::lower_promql; use asap_types::pre_asap::query_expr::Source; use asap_types::pre_asap::schema::{Column, Schema}; + use asap_types::types::AccuracyTarget; use std::time::Duration; fn metric_scan(labels: &[&str]) -> QueryExpr { @@ -419,13 +421,10 @@ mod tests { // Temporal averages expose two single-measure children without closing labels. #[test] fn temporal_average_components_preserves_schema_and_exposes_sum_count() { - let root = Rc::new( - asap_frontend_promql::lower_promql( - "avg_over_time(a{job=\"api\"}[5m])", - AccuracyTarget::Exact, - ) - .unwrap(), - ); + let root = Rc::new(lower_promql( + "avg_over_time(a{job=\"api\"}[5m])", + AccuracyTarget::Exact, + )); assert!(SemanticEquivalentRewriteStrategy .replacements(&TargetSubDAG::new(&root)) .is_empty()); diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index 8a11f7b5..85f56431 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -3083,7 +3083,10 @@ mod tests { fn streaming_data_workload() -> DataWorkload { DataWorkload { arrival: DataArrival::ContinuouslyIngesting, - + data_ingestion_interval: Evidence { + value: Some(asap_types::workload::DurationMs(1_000)), + ..Default::default() + }, ingestion_rate: Evidence { value: Some(Rate(2.0)), source: EvidenceSource::Declared, @@ -3491,7 +3494,10 @@ mod tests { ComparisonScope::from_workload( &DataWorkload { arrival: DataArrival::ContinuouslyIngesting, - + data_ingestion_interval: Evidence { + value: Some(asap_types::workload::DurationMs(1_000)), + ..Default::default() + }, ingestion_rate: Evidence { value: Some(Rate(2.0)), source: EvidenceSource::Declared, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs index d56b53c3..3bfe3b34 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs @@ -1690,7 +1690,10 @@ mod tests { fn at_rest() -> DataWorkload { DataWorkload { arrival: DataArrival::AtRest, - + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, ..Default::default() } } @@ -1698,7 +1701,10 @@ mod tests { fn continuous(observed_at_ms: u64, valid_for_ms: u64) -> DataWorkload { DataWorkload { arrival: DataArrival::ContinuouslyIngesting, - + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, ingestion_rate: Evidence { value: Some(Rate(1.0)), source: EvidenceSource::Observed, diff --git a/crates/asap-aware-mapping/src/test_support.rs b/crates/asap-aware-mapping/src/test_support.rs new file mode 100644 index 00000000..612cf7c0 --- /dev/null +++ b/crates/asap-aware-mapping/src/test_support.rs @@ -0,0 +1,37 @@ +use asap_types::pre_asap::QueryExpr; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, + Predictability, Query, QueryLanguage, QueryRequirements, QueryWorkload, TimeSelection, +}; + +pub(crate) fn lower_promql(query: &str, accuracy: AccuracyTarget) -> QueryExpr { + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy), + ..Default::default() + }, + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + asap_frontend_promql::lower_promql_workload(&workload, 0) + .unwrap() + .pop() + .unwrap() +} diff --git a/crates/devtools/examples/canonical_examples.rs b/crates/devtools/examples/canonical_examples.rs index d3e4916d..c7d08924 100644 --- a/crates/devtools/examples/canonical_examples.rs +++ b/crates/devtools/examples/canonical_examples.rs @@ -3,7 +3,7 @@ // One-off: pretty-print the QueryExpr for one canonical query per variant, // plus custom Join/SetOp/Dedup/CTE probes, to eyeball the actual shape. -use asap_devtools::lower_promql; +use asap_devtools::lower_promql_with_data_ingestion_interval; use asap_frontend_sql::{lower_sql_dialect, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; @@ -70,7 +70,7 @@ async fn main() { ]; for (label, q) in promql_examples { println!("=== {label} === promql> {q}"); - match lower_promql(q, AccuracyTarget::Exact) { + match lower_promql_with_data_ingestion_interval(q, AccuracyTarget::Exact, 1_000) { Ok(qe) => println!("{qe:#?}"), Err(e) => println!("ERR: {e}"), } diff --git a/crates/devtools/examples/topk_ir.rs b/crates/devtools/examples/topk_ir.rs index a44d2501..ee467aa1 100644 --- a/crates/devtools/examples/topk_ir.rs +++ b/crates/devtools/examples/topk_ir.rs @@ -3,7 +3,7 @@ // Lowers every topk-shaped query from the design discussion and prints the // resulting pre-ASAP IR. Used for interactive exploration; not a test. -use asap_devtools::{lower_promql, lower_sql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; @@ -49,7 +49,7 @@ async fn show_sql(label: &str, query: &str) { fn show_promql(label: &str, query: &str) { println!("━━━ {label} ━━━"); println!("{query}"); - match lower_promql(query, AccuracyTarget::Exact) { + match lower_promql_with_data_ingestion_interval(query, AccuracyTarget::Exact, 1_000) { Ok(qe) => println!("{qe:#?}"), Err(e) => println!("ERR: {e}"), } diff --git a/crates/devtools/src/bin/analyze_corpora.rs b/crates/devtools/src/bin/analyze_corpora.rs index ef1df13c..40eb3038 100644 --- a/crates/devtools/src/bin/analyze_corpora.rs +++ b/crates/devtools/src/bin/analyze_corpora.rs @@ -4,7 +4,7 @@ // Corpus mode dumps all four PromQL corpora as JSONL and writes a heuristic // anomaly report. The default mode remains the ad-hoc SQL/PromQL inspector. -use asap_devtools::{lower_promql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, SqlCatalog}; use asap_frontend_sql::lower_sql_dialect; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; @@ -201,12 +201,16 @@ fn structural_shape(expression: &str) -> String { out } -fn run_corpus(name: &str, source: &str) -> CorpusResult { +fn run_corpus(name: &str, source: &str, interval_ms: u64) -> CorpusResult { let mut result = CorpusResult::default(); for (index, expression) in corpus_lines(source).into_iter().enumerate() { let normalized_expression = normalize(expression); let structural_shape = structural_shape(expression); - match lower_promql(expression, AccuracyTarget::Exact) { + match lower_promql_with_data_ingestion_interval( + expression, + AccuracyTarget::Exact, + interval_ms, + ) { Ok(ir) => result.lowered.push(DumpRecord { corpus: name.to_string(), query_number: index + 1, @@ -394,7 +398,7 @@ fn anomaly_report(all: &[DumpRecord], language: &str, manual_notes: &str) -> Str report } -fn run_corpora(out_dir: PathBuf) { +fn run_corpora(out_dir: PathBuf, interval_ms: u64) { std::fs::create_dir_all(&out_dir) .unwrap_or_else(|e| panic!("failed to create {}: {e}", out_dir.display())); let corpora = [ @@ -406,7 +410,7 @@ fn run_corpora(out_dir: PathBuf) { let mut all = Vec::new(); let mut summary = Vec::new(); for (name, source) in corpora { - let mut result = run_corpus(name, source); + let mut result = run_corpus(name, source, interval_ms); let total = result.lowered.len() + result.failed.len(); write_jsonl(&out_dir.join(format!("{name}.jsonl")), &result.lowered); write_jsonl( @@ -542,14 +546,25 @@ async fn main() { let mut args = std::env::args().skip(1); if args.next().as_deref() == Some("--corpora") { let mut out_dir = PathBuf::from("artifacts/promql_pre_asap"); + let mut interval_ms = None; while let Some(arg) = args.next() { if arg == "--out-dir" { out_dir = PathBuf::from(args.next().expect("--out-dir requires a path")); + } else if arg == "--data-ingestion-interval-ms" { + interval_ms = Some( + args.next() + .expect("--data-ingestion-interval-ms requires a value") + .parse() + .expect("--data-ingestion-interval-ms must be an unsigned integer"), + ); } else { panic!("unknown corpus-mode argument: {arg}"); } } - run_corpora(out_dir); + run_corpora( + out_dir, + interval_ms.expect("--data-ingestion-interval-ms is required for --corpora"), + ); return; } if std::env::args().nth(1).as_deref() == Some("--sql-corpora") { diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 08df63be..d49f4e06 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -677,7 +677,7 @@ fn winner_cost_annotations() -> (CostAnnotation, CostAnnotation, CostAnnotation) ) } -use asap_devtools::{lower_promql, lower_sql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; enum Lang { Sql, @@ -766,6 +766,7 @@ struct ParsedArgs { table_schemas: Vec, planner_cost: Option, topk_margin: Option, + promql_ingestion_interval_ms: Option, } #[derive(Debug, Clone, serde::Deserialize)] @@ -822,6 +823,7 @@ fn parse_args() -> ParsedArgs { let mut table_schemas = Vec::new(); let mut planner_cost_json = None; let mut topk_margin_json = None; + let mut promql_ingestion_interval_ms = None; let mut args = std::env::args().skip(1); fn flush(entries: &mut Vec<(String, Lang, String)>, pending: &mut Option<(Lang, String)>) { @@ -877,6 +879,14 @@ fn parse_args() -> ParsedArgs { .expect("--topk-margin-json requires a JSON object"), ); } + "--data-ingestion-interval-ms" => { + promql_ingestion_interval_ms = Some( + args.next() + .expect("--data-ingestion-interval-ms requires a value") + .parse() + .expect("--data-ingestion-interval-ms must be an unsigned integer"), + ); + } other => panic!("unrecognized argument: {other}"), } } @@ -899,6 +909,7 @@ fn parse_args() -> ParsedArgs { table_schemas, planner_cost, topk_margin, + promql_ingestion_interval_ms, } } @@ -1421,6 +1432,7 @@ async fn main() { table_schemas, planner_cost, topk_margin, + promql_ingestion_interval_ms, } = parse_args(); let sql_catalog = catalog(&table_schemas); let planner_started = Instant::now(); @@ -1441,7 +1453,13 @@ async fn main() { Lang::Sql => lower_sql(&query, &sql_catalog, accuracy.clone()) .await .map_err(|e| e.to_string()), - Lang::PromQl => lower_promql(&query, accuracy.clone()).map_err(|e| e.to_string()), + Lang::PromQl => lower_promql_with_data_ingestion_interval( + &query, + accuracy.clone(), + promql_ingestion_interval_ms + .expect("--data-ingestion-interval-ms is required for PromQL queries"), + ) + .map_err(|e| e.to_string()), }; match lowered { Ok(qe) => { @@ -1617,8 +1635,13 @@ mod tests { EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics, }; use asap_aware_mapping::query_physical_lowering::lower_query_physical_dag; + use asap_devtools::PromqlError; use asap_types::pre_asap::{Column, DataType, Reduction, Schema, Source}; + fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Result { + lower_promql_with_data_ingestion_interval(query, accuracy, 1_000) + } + fn non_topk_query() -> QueryExpr { QueryExpr::Aggregate { reduction: Reduction::by(vec![]), diff --git a/crates/devtools/src/bin/show_post_asap_ir.rs b/crates/devtools/src/bin/show_post_asap_ir.rs index aa53b6d8..816ad8c5 100644 --- a/crates/devtools/src/bin/show_post_asap_ir.rs +++ b/crates/devtools/src/bin/show_post_asap_ir.rs @@ -24,7 +24,7 @@ use asap_aware_mapping::replacement::keep_pre_asap; use asap_aware_mapping::{ Replacement, ReplacementStrategy, ReplacementSubDAG, SketchAlgorithmStrategy, TargetSubDAG, }; -use asap_devtools::{lower_promql, lower_sql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; use asap_types::pre_asap::query_expr::QueryExpr; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; @@ -81,7 +81,18 @@ fn catalog() -> SqlCatalog { #[tokio::main] async fn main() { - let input = match std::env::args().nth(1) { + let mut args = std::env::args().skip(1); + assert_eq!( + args.next().as_deref(), + Some("--data-ingestion-interval-ms"), + "usage: show_post_asap_ir --data-ingestion-interval-ms [queries.txt]" + ); + let interval_ms = args + .next() + .expect("missing interval") + .parse() + .expect("interval must be an unsigned integer"); + let input = match args.next() { Some(path) => { std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("failed to read {path}: {e}")) } @@ -106,7 +117,8 @@ async fn main() { .await .map_err(|e| e.to_string()) } else if let Some(q) = line.strip_prefix("promql>") { - lower_promql(q.trim(), ACCURACY.clone()).map_err(|e| e.to_string()) + lower_promql_with_data_ingestion_interval(q.trim(), ACCURACY.clone(), interval_ms) + .map_err(|e| e.to_string()) } else { println!("ERR: line must start with 'sql>' or 'promql>'"); println!(); @@ -131,9 +143,10 @@ mod tests { #[test] fn bind_all_returns_every_sketch_candidate() { - let expr = lower_promql( + let expr = lower_promql_with_data_ingestion_interval( "quantile(0.99, rate(http_requests_total[5m]))", ACCURACY.clone(), + 1_000, ) .expect("query lowers to pre-ASAP IR"); let root = Rc::new(expr.clone()); diff --git a/crates/devtools/src/bin/show_pre_asap_ir.rs b/crates/devtools/src/bin/show_pre_asap_ir.rs index c2a3c9ac..b2f60463 100644 --- a/crates/devtools/src/bin/show_pre_asap_ir.rs +++ b/crates/devtools/src/bin/show_pre_asap_ir.rs @@ -16,7 +16,7 @@ // SQL queries run against a fixed `metrics(ts, service, region, latency, // bytes)` catalog — the same table used in cross_language.rs and topk_ir.rs. -use asap_devtools::{lower_promql, lower_sql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::types::AccuracyTarget; use std::io::Read; @@ -44,7 +44,18 @@ fn catalog() -> SqlCatalog { #[tokio::main] async fn main() { - let input = match std::env::args().nth(1) { + let mut args = std::env::args().skip(1); + assert_eq!( + args.next().as_deref(), + Some("--data-ingestion-interval-ms"), + "usage: show_pre_asap_ir --data-ingestion-interval-ms [queries.txt]" + ); + let interval_ms = args + .next() + .expect("missing interval") + .parse() + .expect("interval must be an unsigned integer"); + let input = match args.next() { Some(path) => { std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("failed to read {path}: {e}")) } @@ -70,7 +81,11 @@ async fn main() { Err(e) => println!("ERR: {e}"), } } else if let Some(q) = line.strip_prefix("promql>") { - match lower_promql(q.trim(), AccuracyTarget::Exact) { + match lower_promql_with_data_ingestion_interval( + q.trim(), + AccuracyTarget::Exact, + interval_ms, + ) { Ok(qe) => println!("{qe:#?}"), Err(e) => println!("ERR: {e}"), } diff --git a/crates/devtools/src/bin/sketch_coverage.rs b/crates/devtools/src/bin/sketch_coverage.rs index a445dad3..e9f0ccb1 100644 --- a/crates/devtools/src/bin/sketch_coverage.rs +++ b/crates/devtools/src/bin/sketch_coverage.rs @@ -26,7 +26,7 @@ // dag-viewer's Union mode. use asap_aware_mapping::{explain_replacements, ExplanationKind}; -use asap_devtools::lower_promql; +use asap_devtools::lower_promql_with_data_ingestion_interval; use asap_frontend_sql::{lower_sql_dialect, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::pre_asap::QueryExpr; @@ -233,9 +233,24 @@ fn parse_epsilon() -> f64 { 0.01 } +fn parse_ingestion_interval() -> u64 { + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + if arg == "--data-ingestion-interval-ms" { + return args + .next() + .expect("--data-ingestion-interval-ms requires a value") + .parse() + .expect("--data-ingestion-interval-ms must be an unsigned integer"); + } + } + panic!("--data-ingestion-interval-ms is required") +} + #[tokio::main] async fn main() { let epsilon = parse_epsilon(); + let interval_ms = parse_ingestion_interval(); let accuracy = AccuracyTarget::Epsilon(epsilon); let mut results = Vec::new(); @@ -268,7 +283,7 @@ async fn main() { let mut roots = Vec::new(); let mut failed = 0; for (i, q) in promql_lines(corpus).enumerate() { - match lower_promql(q, accuracy.clone()) { + match lower_promql_with_data_ingestion_interval(q, accuracy.clone(), interval_ms) { Ok(qe) => roots.push((format!("q{i}"), qe)), Err(_) => failed += 1, } diff --git a/crates/devtools/src/bin/variant_coverage.rs b/crates/devtools/src/bin/variant_coverage.rs index 0cd66ee2..a96042a2 100644 --- a/crates/devtools/src/bin/variant_coverage.rs +++ b/crates/devtools/src/bin/variant_coverage.rs @@ -4,7 +4,7 @@ // resulting QueryExpr trees, and reports which enum variants show up — per // corpus, then rolled up globally. Used to find the minimal QueryExpr node set. -use asap_devtools::lower_promql; +use asap_devtools::lower_promql_with_data_ingestion_interval; use asap_frontend_sql::{lower_sql_dialect, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::pre_asap::QueryExpr; @@ -247,8 +247,23 @@ fn report(r: &CorpusResult) { println!(); } +fn parse_ingestion_interval() -> u64 { + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + if arg == "--data-ingestion-interval-ms" { + return args + .next() + .expect("--data-ingestion-interval-ms requires a value") + .parse() + .expect("--data-ingestion-interval-ms must be an unsigned integer"); + } + } + panic!("--data-ingestion-interval-ms is required") +} + #[tokio::main] async fn main() { + let interval_ms = parse_ingestion_interval(); let mut results = Vec::new(); // ── PromQL corpora ── @@ -281,7 +296,7 @@ async fn main() { let mut lowered = 0; let mut failed = 0; for q in promql_lines(corpus) { - match lower_promql(q, AccuracyTarget::Exact) { + match lower_promql_with_data_ingestion_interval(q, AccuracyTarget::Exact, interval_ms) { Ok(qe) => { walk(&qe, &mut variants); lowered += 1; diff --git a/crates/devtools/src/lib.rs b/crates/devtools/src/lib.rs index f9feca6c..5e6a0208 100644 --- a/crates/devtools/src/lib.rs +++ b/crates/devtools/src/lib.rs @@ -12,5 +12,53 @@ //! only) or [`asap_frontend_sql`] (DataFusion only) — so it never compiles the //! other's parser. -pub use asap_frontend_promql::{lower_promql, lower_promql_batch, PromqlError, PromqlLowerer}; +pub use asap_frontend_promql::{ + lower_promql_workload, lower_promql_workload_with_histograms, PromqlError, +}; pub use asap_frontend_sql::{lower_sql, lower_sql_batch, SqlCatalog, SqlError, SqlLowerer}; + +/// Lower one developer-supplied PromQL query with an explicit source cadence. +/// Tools intentionally require the cadence rather than choosing a default. +pub fn lower_promql_with_data_ingestion_interval( + query: &str, + accuracy: asap_types::types::AccuracyTarget, + interval_ms: u64, +) -> Result { + use asap_types::workload::{ + BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, Predictability, Query, + QueryRequirements, QueryWorkload, TimeSelection, + }; + + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: asap_types::workload::QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: asap_types::workload::AccuracyRequirement::Explicit(accuracy), + ..Default::default() + }, + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(interval_ms)), + ..Default::default() + }, + ..Default::default() + }), + }; + lower_promql_workload(&workload, 0)? + .into_iter() + .next() + .ok_or({ + PromqlError::InvalidWorkload( + asap_types::workload::WorkloadError::MissingPromqlDataWorkload, + ) + }) +} diff --git a/crates/devtools/tests/cross_language.rs b/crates/devtools/tests/cross_language.rs index 5426e4f2..0518f62b 100644 --- a/crates/devtools/tests/cross_language.rs +++ b/crates/devtools/tests/cross_language.rs @@ -13,7 +13,7 @@ //! match is the **shape above the leaf**: an outer `Aggregate([TopK{k}])` over an //! explicit inner `Aggregate([Count])`. -use asap_devtools::{lower_promql, lower_sql, SqlCatalog}; +use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::pre_asap::{AggIntent, GroupKeys, QueryExpr}; use asap_types::types::AccuracyTarget; @@ -46,7 +46,7 @@ async fn sql(q: &str) -> QueryExpr { } fn promql(q: &str) -> QueryExpr { - lower_promql(q, AccuracyTarget::Exact) + lower_promql_with_data_ingestion_interval(q, AccuracyTarget::Exact, 1_000) .unwrap_or_else(|e| panic!("PromQL {q:?} failed to lower: {e:?}")) } diff --git a/crates/frontend-promql/src/error.rs b/crates/frontend-promql/src/error.rs index 932f77b8..1885a5f4 100644 --- a/crates/frontend-promql/src/error.rs +++ b/crates/frontend-promql/src/error.rs @@ -1,6 +1,7 @@ use std::fmt; use asap_types::pre_asap::ResolveTreeError; +use asap_types::workload::WorkloadError; /// Errors from lowering a PromQL query (parse → the canonical, unresolved /// tree, built directly → @@ -13,6 +14,8 @@ use asap_types::pre_asap::ResolveTreeError; /// shared, so neither front end pulls the other's parser. #[derive(Debug)] pub enum PromqlError { + /// The workload omitted information required for plan-ready PromQL lowering. + InvalidWorkload(WorkloadError), /// The `promql-parser` crate rejected the query string (parse failure). Parse(String), /// A PromQL function (`rate`, `*_over_time`, …) not supported in this version. @@ -36,6 +39,7 @@ pub enum PromqlError { impl fmt::Display for PromqlError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Self::InvalidWorkload(e) => write!(f, "invalid PromQL workload: {e}"), Self::Parse(e) => write!(f, "PromQL parse error: {e}"), Self::UnsupportedFunction(n) => write!(f, "unsupported PromQL function: {n}"), Self::UnsupportedAggregateOp(n) => write!(f, "unsupported PromQL aggregate op: {n}"), @@ -56,6 +60,12 @@ impl From for PromqlError { } } +impl From for PromqlError { + fn from(e: WorkloadError) -> Self { + Self::InvalidWorkload(e) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/frontend-promql/src/lib.rs b/crates/frontend-promql/src/lib.rs index ff29379a..7a17d505 100644 --- a/crates/frontend-promql/src/lib.rs +++ b/crates/frontend-promql/src/lib.rs @@ -14,65 +14,180 @@ pub mod promql; use asap_types::pre_asap::resolve_root; use asap_types::pre_asap::QueryExpr; -use asap_types::types::AccuracyTarget; -use asap_types::workload::{QueryLanguage, QueryWorkload}; +use asap_types::workload::{DurationMs, PlanningWorkload, QueryLanguage, WorkloadError}; pub use error::PromqlError; pub use histogram::{HistogramCatalog, HistogramKind}; -pub use promql::PromqlLowerer; -/// Lower a single PromQL query string to the canonical, resolved `QueryExpr`. +/// Lower every normalized PromQL workload entry to a plan-ready `QueryExpr`. /// -/// `accuracy` is threaded onto every approximate intent (`Count`, `Quantile`, -/// `Cardinality`, `TopK`). The returned tree carries a self-contained `Schema` -/// on its `Scan`; call [`QueryExpr::output_schema`] for any node's schema. -/// -/// `histogram_quantile` discrimination uses the structural heuristic; to drive -/// it from declared sample types instead, use [`lower_promql_with_histograms`]. -pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Result { - let unresolved = PromqlLowerer::lower(query, &accuracy)?; - let resolved = resolve_root(&unresolved)?; - Ok(resolved) +/// PromQL workloads must declare a non-zero `data_ingestion_interval`; it is +/// injected around each bare instant selector. Explicit range selectors keep +/// their query-specified range. +/// `now_ms` is the planning time in Unix milliseconds; cadence evidence must +/// be valid at that time, using the same clock as downstream planning. +pub fn lower_promql_workload( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, PromqlError> { + lower_promql_workload_inner(workload, now_ms) } -/// Like [`lower_promql`], but consults `histograms` to decide whether a -/// `histogram_quantile` argument is sketch-able (generic `Quantile`) or a -/// classic-bucket interpolation (`HistogramQuantile`) — a type-driven decision -/// instead of the structural heuristic (issue #79). Metrics absent from the -/// catalog still fall back to the heuristic. -pub fn lower_promql_with_histograms( - query: &str, - accuracy: AccuracyTarget, +/// Like [`lower_promql_workload`], but uses `histograms` to distinguish classic +/// bucket interpolation from generic sketchable quantiles. +pub fn lower_promql_workload_with_histograms( + workload: &PlanningWorkload, histograms: HistogramCatalog, -) -> Result { + now_ms: u64, +) -> Result, PromqlError> { let _guard = histogram::CatalogGuard::install(histograms); - lower_promql(query, accuracy) + lower_promql_workload_inner(workload, now_ms) } -/// Lower every PromQL batch entry in `workload` to a `QueryExpr`. -/// -/// One `Result` per entry — errors are per-query, not fatal for the batch. -/// Returns an empty `Vec` if `workload.query_batch` is absent or empty, and a -/// `WrongLanguage` error for every entry if the workload language is not PromQL. -pub fn lower_promql_batch(workload: &QueryWorkload) -> Vec> { - let entries = match &workload.query_batch { - Some(e) if !e.is_empty() => e, - _ => return vec![], - }; - - if !matches!(workload.language, QueryLanguage::PromQL) { - let lang = format!("{:?}", workload.language); - return entries - .iter() - .map(|_| Err(PromqlError::WrongLanguage(lang.clone()))) - .collect(); +fn lower_promql_workload_inner( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, PromqlError> { + if !matches!(workload.query_workload.language, QueryLanguage::PromQL) { + return Err(PromqlError::WrongLanguage(format!( + "{:?}", + workload.query_workload.language + ))); } - - entries - .iter() + workload.validate()?; + let &DurationMs(interval_ms) = workload + .data_workload + .as_ref() + .expect("validated PromQL workload has data_workload") + .data_ingestion_interval + .value_at(now_ms) + .ok_or(WorkloadError::UnavailableDataIngestionInterval)?; + workload + .query_workload + .entries() .map(|entry| { - let accuracy = entry.requirements.accuracy.target(); - lower_promql(&entry.query.0, accuracy) + let unresolved = promql::PromqlLowerer::lower_with_ingestion_interval( + &entry.query.0, + &entry.requirements.accuracy.target(), + std::time::Duration::from_millis(interval_ms), + )?; + Ok(resolve_root(&unresolved)?) }) .collect() } + +#[cfg(test)] +mod tests { + // Expiring evidence without an observation timestamp is never usable. + #[test] + fn rejects_unusable_ingestion_evidence() { + let mut input = workload("sum(data)"); + input + .data_workload + .as_mut() + .unwrap() + .data_ingestion_interval + .valid_for_ms = Some(100); + assert!(lower_promql_workload(&input, 0).is_err()); + } + + // Cadence expiry is inclusive; future and expired evidence cannot set a horizon. + #[test] + fn ingestion_evidence_respects_planning_time_with_and_without_histograms() { + let mut input = workload("sum(data)"); + let evidence = &mut input + .data_workload + .as_mut() + .unwrap() + .data_ingestion_interval; + evidence.observed_at_ms = Some(1_000); + evidence.valid_for_ms = Some(100); + for (now_ms, usable) in [(999, false), (1_000, true), (1_100, true), (1_101, false)] { + assert_eq!(lower_promql_workload(&input, now_ms).is_ok(), usable); + assert_eq!( + lower_promql_workload_with_histograms(&input, HistogramCatalog::default(), now_ms) + .is_ok(), + usable + ); + } + input + .data_workload + .as_mut() + .unwrap() + .data_ingestion_interval + .observed_at_ms = None; + assert!( + lower_promql_workload_with_histograms(&input, HistogramCatalog::default(), 1_000) + .is_err() + ); + } + use std::time::Duration; + + use asap_types::pre_asap::QueryExpr; + use asap_types::workload::{ + BatchEntry, DataWorkload, Evidence, PlanningWorkload, Query, QueryRequirements, + QueryWorkload, TimeSelection, + }; + + use super::*; + + fn workload(query: &str) -> PlanningWorkload { + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements::default(), + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + } + } + + #[test] + fn instant_selector_uses_declared_ingestion_interval() { + let query = lower_promql_workload(&workload("sum by (job) (data)"), 0).unwrap(); + let QueryExpr::Aggregate { child, .. } = &query[0] else { + panic!("expected aggregate") + }; + assert!( + matches!(child.as_ref(), QueryExpr::TimeRange { range, child } + if *range == Duration::from_secs(1) && matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); + } + + #[test] + fn explicit_range_selector_keeps_its_query_range() { + let query = lower_promql_workload(&workload("sum_over_time(data[5m])"), 0).unwrap(); + let QueryExpr::Aggregate { child, .. } = &query[0] else { + panic!("expected aggregate") + }; + assert!( + matches!(child.as_ref(), QueryExpr::TimeRange { range, child } + if *range == Duration::from_secs(300) && matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); + } + + #[test] + fn workload_without_interval_fails_loudly() { + let mut workload = workload("sum(data)"); + workload.data_workload = Some(DataWorkload::default()); + assert!(matches!( + lower_promql_workload(&workload, 0), + Err(PromqlError::InvalidWorkload( + asap_types::workload::WorkloadError::MissingDataIngestionInterval + )) + )); + } +} diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 424666aa..de380c92 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -80,7 +80,7 @@ use crate::error::PromqlError as LoweringError; type Result = std::result::Result; /// Parses and lowers (→ the canonical, unresolved tree) a PromQL query string. -pub struct PromqlLowerer; +pub(crate) struct PromqlLowerer; #[derive(Debug, Clone)] enum Outer { @@ -172,22 +172,14 @@ struct Inner { const MAX_DEPTH: usize = 256; impl PromqlLowerer { - /// Lower `query` to the canonical Unresolved tree, threading `accuracy` onto every - /// approximate intent (`Count`, `Quantile`, `Cardinality`, `TopK`) as it is - /// built — this front end constructs the canonical shape directly (issue - /// #179), so accuracy is baked in here rather than threaded through a - /// later, separate converter pass. `accuracy` rides the same ambient, - /// thread-local mechanism as `histogram::CatalogGuard` - /// — synchronous, one-query-at-a-time lowering, injected into the deep - /// `walk` recursion without a parameter on every one of its ~30 mutually - /// recursive signatures; consulted only at the handful of sites that build - /// an accuracy-bearing `AggIntent`. - pub fn lower(query: &str, accuracy: &AccuracyTarget) -> Result { + pub(crate) fn lower_with_ingestion_interval( + query: &str, + accuracy: &AccuracyTarget, + interval: Duration, + ) -> Result { let _guard = AccuracyGuard::install(accuracy.clone()); + let _interval = IngestionIntervalGuard::install(interval); let ast = parser::parse(query).map_err(LoweringError::Parse)?; - // Reject over-deep nesting up front, so the (mutually-recursive) walk - // below cannot blow the stack. The check itself recurses at most - // `MAX_DEPTH` frames before erroring, so it is bounded too. check_depth(&ast, MAX_DEPTH)?; walk(&ast) } @@ -196,6 +188,7 @@ impl PromqlLowerer { std::thread_local! { static ACCURACY: std::cell::RefCell = const { std::cell::RefCell::new(AccuracyTarget::Exact) }; + static INGESTION_INTERVAL: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; } /// RAII guard installing `accuracy` as the ambient accuracy target for the @@ -221,6 +214,28 @@ fn current_accuracy() -> AccuracyTarget { ACCURACY.with(|a| a.borrow().clone()) } +struct IngestionIntervalGuard(Option); + +impl IngestionIntervalGuard { + fn install(interval: Duration) -> Self { + Self(INGESTION_INTERVAL.with(|current| current.replace(Some(interval)))) + } +} + +impl Drop for IngestionIntervalGuard { + fn drop(&mut self) { + INGESTION_INTERVAL.with(|current| *current.borrow_mut() = self.0.take()); + } +} + +fn current_ingestion_interval() -> Duration { + INGESTION_INTERVAL.with(|current| { + current + .borrow() + .expect("ingestion interval is installed for workload lowering") + }) +} + /// Bounded depth check over the parser AST: errors once nesting would exceed /// `budget` frames, descending into every child expression. fn check_depth(expr: &Expr, budget: usize) -> Result<()> { @@ -299,7 +314,7 @@ fn walk(expr: &Expr) -> Result { }), Expr::VectorSelector(vs) => { let (metric, matchers, shift) = vs_parts(vs)?; - Ok(filtered_source(metric, matchers, shift)) + Ok(instant_source(metric, matchers, shift)) } Expr::MatrixSelector(ms) => { let (metric, matchers, shift) = vs_parts(&ms.vs)?; @@ -361,7 +376,7 @@ fn range_fn_over_subquery(call: &Call) -> Result> { "increase" => InnerFunc::Increase, _ => unreachable!(), }; - return Ok(Some(outer_aggregate( + return Ok(Some(per_series_aggregate( vec![], inner_intent(&inner), walk(arg_expr)?, @@ -409,7 +424,7 @@ fn range_fn_over_subquery(call: &Call) -> Result> { if !is_subquery(arg_expr) { return Ok(None); } - Ok(Some(outer_aggregate( + Ok(Some(per_series_aggregate( vec![], inner_intent(&inner), walk(arg_expr)?, @@ -742,7 +757,7 @@ fn walk_histogram_quantiles(call: &Call) -> Result { AggIntent::HistogramQuantile { q: phi } }; let child = walk(vec_expr)?; - let reduction = reduction_for(&[], &intent, &child); + let reduction = reduction_for(&[], intent.is_per_series()); let quantile = Unresolved::Aggregate { reduction, measures: vec![intent], @@ -1421,7 +1436,7 @@ fn lower_inner_call(call: &Call) -> Result { fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { match outer { Outer::None => match &inner.func { - None => Ok(filtered_source(inner.metric, inner.matchers, inner.shift)), + None => Ok(instant_source(inner.metric, inner.matchers, inner.shift)), Some(f) => { let intent = inner_intent(f); Ok(windowed_aggregate(inner, keys, intent)) @@ -1463,7 +1478,7 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result // wrapped in a reducing aggregate (issue #86). let base = match inner.func.as_ref().map(inner_intent) { Some(intent) => windowed_aggregate(inner, vec![], intent), - None => filtered_source(inner.metric, inner.matchers, inner.shift), + None => instant_source(inner.metric, inner.matchers, inner.shift), }; Ok(Unresolved::PromqlSeriesSample { by: keys.into(), @@ -1526,7 +1541,7 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result // `Sort.partition_by` can rank within each group (issue #12). let base = match inner.func.as_ref().map(inner_intent) { Some(intent) => windowed_aggregate(inner, vec![], intent), - None => filtered_source(inner.metric, inner.matchers, inner.shift), + None => instant_source(inner.metric, inner.matchers, inner.shift), }; let sorted = Unresolved::Sort { keys: vec![SortKey { @@ -1548,22 +1563,13 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result } /// Decide `PerEntity` vs `Reduce(by)` for a canonical `Aggregate`, entirely -/// from local, already-in-scope information (issue #179's "local, -/// context-free structural rewriting"): the keys list, the intent's own -/// `is_per_series` flag, and whether the child being wrapped is already a -/// range/subquery marker — no schema needed. `without()` is applied +/// from local PromQL semantics: the keys and whether this operation preserves +/// each input series. It never infers entity reduction from the child tree's +/// temporal shape. `without()` is applied /// separately, post-hoc, by `mark_without` — see its doc for why that's still /// correct here. -fn reduction_for( - keys: &[ColumnRef], - intent: &AggIntent, - child: &Unresolved, -) -> Reduction { - let is_range_child = matches!( - child, - Unresolved::TimeRange { .. } | Unresolved::PromqlSubquery { .. } - ); - if keys.is_empty() && (intent.is_per_series() || is_range_child) { +fn reduction_for(keys: &[ColumnRef], per_entity: bool) -> Reduction { + if keys.is_empty() && per_entity { Reduction::PerEntity } else { Reduction::Reduce(GroupKeys::by(keys.to_vec())) @@ -1590,7 +1596,15 @@ fn windowed_aggregate( }, None => base, }; - let reduction = reduction_for(&keys, &intent, &child); + let reduction = reduction_for(&keys, inner.window.is_some() || intent.is_per_series()); + let child = if inner.window.is_none() { + Unresolved::TimeRange { + range: current_ingestion_interval(), + child: Rc::new(child), + } + } else { + child + }; Unresolved::Aggregate { reduction, measures: vec![intent], @@ -1611,7 +1625,26 @@ fn outer_aggregate( intent: AggIntent, child: Unresolved, ) -> Unresolved { - let reduction = reduction_for(&keys, &intent, &child); + let reduction = reduction_for(&keys, intent.is_per_series()); + Unresolved::Aggregate { + reduction, + measures: vec![intent], + output_names: vec![String::new()], + having: None, + child: Rc::new(child), + } +} + +/// A temporal range function over a subquery consumes each series' subquery +/// samples independently. Unlike an ordinary outer aggregate, this cannot be +/// inferred from the intent: `max` is cross-series in `max(v)`, but per-series +/// in `max_over_time(v[...])`. +fn per_series_aggregate( + keys: Vec, + intent: AggIntent, + child: Unresolved, +) -> Unresolved { + let reduction = reduction_for(&keys, true); Unresolved::Aggregate { reduction, measures: vec![intent], @@ -1641,6 +1674,13 @@ fn filtered_source(metric: String, matchers: Vec, shift: TimeShift) } } +fn instant_source(metric: String, matchers: Vec, shift: TimeShift) -> Unresolved { + Unresolved::TimeRange { + range: current_ingestion_interval(), + child: Rc::new(filtered_source(metric, matchers, shift)), + } +} + /// Count vector elements regardless of their sample values. fn count() -> AggIntent { AggIntent::Count { diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs index 232f1edd..b255aac3 100644 --- a/crates/frontend-promql/tests/count_planning.rs +++ b/crates/frontend-promql/tests/count_planning.rs @@ -2,12 +2,13 @@ use std::rc::Rc; use asap_aware_mapping::{Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG}; -use asap_frontend_promql::lower_promql; +mod support; use asap_types::post_asap::{ compile_executable_dag, ExactKind, ExecutableOperatorPayload, NonNegativeWeightProof, SketchAlgorithm, SummaryExpr, SummaryFamilyType, SummaryInputExpr, WeightDomain, }; use asap_types::types::AccuracyTarget; +use support::lower_promql; // Exact series and temporal counts must select a count accumulator, not distinct or sum. #[test] @@ -111,7 +112,10 @@ fn aggregate_fixture(query: &str, series: &[Vec]) -> Vec { match child.as_ref() { QueryExpr::Scan { .. } => assert!(series.iter().all(|samples| samples.len() == 1)), QueryExpr::TimeRange { range, child } => { - assert_eq!(range.as_secs(), 300); + assert!(matches!(range.as_secs(), 1 | 300)); + if range.as_secs() == 1 { + assert!(series.iter().all(|samples| samples.len() == 1)); + } assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); } other => panic!("unsupported fixture input: {other:?}"), diff --git a/crates/frontend-promql/tests/histogram_metadata.rs b/crates/frontend-promql/tests/histogram_metadata.rs index 38de7fac..60ff5a60 100644 --- a/crates/frontend-promql/tests/histogram_metadata.rs +++ b/crates/frontend-promql/tests/histogram_metadata.rs @@ -5,11 +5,11 @@ //! heuristic's false-positive and false-negative cases. Undeclared metrics still //! fall back to the heuristic. -use asap_frontend_promql::{ - lower_promql, lower_promql_with_histograms, HistogramCatalog, HistogramKind, -}; +use asap_frontend_promql::{HistogramCatalog, HistogramKind}; +mod support; use asap_types::pre_asap::{AggIntent, QueryExpr}; use asap_types::types::AccuracyTarget; +use support::{lower_promql, lower_promql_with_histograms}; /// The histogram/quantile intent kind in the lowered tree: `"HQ"` for the /// classic-bucket `HistogramQuantile`, `"Q"` for the sketch-able `Quantile`. diff --git a/crates/frontend-promql/tests/maintained_population_horizon.rs b/crates/frontend-promql/tests/maintained_population_horizon.rs new file mode 100644 index 00000000..f0d47ada --- /dev/null +++ b/crates/frontend-promql/tests/maintained_population_horizon.rs @@ -0,0 +1,37 @@ +mod support; +use asap_aware_mapping::maintained_population::MaintainedPopulationStrategy; +use asap_types::post_asap::maintained_population::PopulationInput; +use asap_types::post_asap::{SummaryExpr, ValueOperation}; +use asap_types::types::AccuracyTarget; +use std::rc::Rc; + +// A population for a one-second selector must expire members after one second. +#[test] +fn population_preserves_selector_horizon() { + let root = Rc::new(support::lower_promql("sum(a)", AccuracyTarget::Exact).unwrap()); + let candidate = MaintainedPopulationStrategy::new(std::slice::from_ref(&root)) + .candidate(&root) + .unwrap(); + let SummaryExpr::ValueOperation { child, .. } = &candidate.expr else { + panic!() + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &child.expr + else { + panic!() + }; + let PopulationInput::CurrentSeries(spec) = &population.input else { + panic!() + }; + assert_eq!(spec.lookback_ms, 1_000); + asap_types::post_asap::compile_executable_dag(&candidate).unwrap(); + let asap_types::pre_asap::QueryExpr::Aggregate { child: source, .. } = root.as_ref() else { + panic!() + }; + assert!(spec.matches_input(source)); + let mut wrong = spec.clone(); + wrong.lookback_ms = 300_000; + assert!(!wrong.matches_input(source)); +} diff --git a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs index 254297de..b2592925 100644 --- a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs +++ b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs @@ -26,9 +26,12 @@ // `__GAP`-suffixed names intentionally SHOUT the documented divergences. #![allow(non_snake_case)] -use asap_frontend_promql::{lower_promql, PromqlError as LoweringError}; +use asap_frontend_promql::PromqlError as LoweringError; +#[path = "../support.rs"] +mod support; use asap_types::pre_asap::{AggIntent, BinaryOpKind, CompareOpKind, QueryExpr, Reduction}; use asap_types::types::AccuracyTarget; +use support::lower_promql; const CORPUS: &str = include_str!("data/awesome_prometheus_alerts.txt"); @@ -174,14 +177,19 @@ fn corpus_lowering_is_total_and_fully_parseable() { #[test] fn vector_vs_vector_comparison_lowers_to_binaryop() { // node-exporter: `node_hwmon_temp_celsius > node_hwmon_temp_max_celsius`. - // Both operands are instant vectors → a `BinaryOp{Compare}` of two scans. + // Both operands are instant vectors → a `BinaryOp{Compare}` of two + // ingestion-interval-bounded scans. let qe = ok("node_hwmon_temp_celsius > node_hwmon_temp_max_celsius"); let QueryExpr::BinaryOp { op, lhs, rhs, .. } = &qe else { panic!("expected BinaryOp, got {qe:?}"); }; assert_eq!(*op, BinaryOpKind::Compare(CompareOpKind::Gt)); - assert!(matches!(lhs.as_ref(), QueryExpr::Scan { .. })); - assert!(matches!(rhs.as_ref(), QueryExpr::Scan { .. })); + assert!( + matches!(lhs.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); + assert!( + matches!(rhs.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); } #[test] diff --git a/crates/frontend-promql/tests/observability/metrics_observability.rs b/crates/frontend-promql/tests/observability/metrics_observability.rs index bd3433e4..a8daba3c 100644 --- a/crates/frontend-promql/tests/observability/metrics_observability.rs +++ b/crates/frontend-promql/tests/observability/metrics_observability.rs @@ -10,10 +10,13 @@ use asap_aware_mapping::replacement::{keep_pre_asap, ImplementError}; use asap_aware_mapping::{ Replacement, ReplacementStrategy, ReplacementSubDAG, SketchAlgorithmStrategy, TargetSubDAG, }; -use asap_frontend_promql::{lower_promql, PromqlError}; +use asap_frontend_promql::PromqlError; +#[path = "../support.rs"] +mod support; use asap_types::post_asap::{SummaryExpr, SummaryNode}; use asap_types::pre_asap::query_expr::QueryExpr; use asap_types::types::AccuracyTarget; +use support::lower_promql; const CORPORA: &[(&str, &str)] = &[ ( diff --git a/crates/frontend-promql/tests/observability/o11y_bench_promql.rs b/crates/frontend-promql/tests/observability/o11y_bench_promql.rs index 9895d0c4..e178d898 100644 --- a/crates/frontend-promql/tests/observability/o11y_bench_promql.rs +++ b/crates/frontend-promql/tests/observability/o11y_bench_promql.rs @@ -18,8 +18,11 @@ //! assert full lowering coverage — a regression here means a real pattern //! broke, not statistical noise. -use asap_frontend_promql::{lower_promql, PromqlError as LoweringError}; +use asap_frontend_promql::PromqlError as LoweringError; +#[path = "../support.rs"] +mod support; use asap_types::types::AccuracyTarget; +use support::lower_promql; const CORPUS: &str = include_str!("data/o11y_bench_promql.txt"); diff --git a/crates/frontend-promql/tests/observability/promql_corpus.rs b/crates/frontend-promql/tests/observability/promql_corpus.rs index 99599952..75bc76cd 100644 --- a/crates/frontend-promql/tests/observability/promql_corpus.rs +++ b/crates/frontend-promql/tests/observability/promql_corpus.rs @@ -19,10 +19,13 @@ use asap_aware_mapping::replacement::{keep_pre_asap, ImplementError}; use asap_aware_mapping::{ Replacement, ReplacementStrategy, ReplacementSubDAG, SketchAlgorithmStrategy, TargetSubDAG, }; -use asap_frontend_promql::{lower_promql, PromqlError as LoweringError}; +use asap_frontend_promql::PromqlError as LoweringError; +#[path = "../support.rs"] +mod support; use asap_types::post_asap::{SummaryExpr, SummaryNode}; use asap_types::pre_asap::query_expr::QueryExpr; use asap_types::types::AccuracyTarget; +use support::lower_promql; /// This crate has no "bind me one tree" public API any more — /// `SketchAlgorithmStrategy::replacements` always returns every candidate, and diff --git a/crates/frontend-promql/tests/promql_binding_regressions.rs b/crates/frontend-promql/tests/promql_binding_regressions.rs index 97d177f9..b5edf604 100644 --- a/crates/frontend-promql/tests/promql_binding_regressions.rs +++ b/crates/frontend-promql/tests/promql_binding_regressions.rs @@ -1,6 +1,8 @@ -use asap_frontend_promql::lower_promql; use asap_types::types::AccuracyTarget; +mod support; +use support::lower_promql; + /// Prometheus treats these quantile parameters as valid queries returning special values. #[test] fn quantile_parameters_retain_prometheus_special_value_semantics() { diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs index b724cf3c..27af2c48 100644 --- a/crates/frontend-promql/tests/promql_conformance.rs +++ b/crates/frontend-promql/tests/promql_conformance.rs @@ -33,13 +33,15 @@ use std::time::Duration; -use asap_frontend_promql::{lower_promql, PromqlError as LoweringError}; +use asap_frontend_promql::PromqlError as LoweringError; +mod support; use asap_types::pre_asap::schema::DataType; use asap_types::pre_asap::{ AggIntent, ArithmeticOpKind, AtModifier, BinaryOpKind, CompareOpKind, MathFunc, PromQLVectorSetOpKind, QueryExpr, Reduction, SampleKind, Source, TimeFunc, }; use asap_types::types::AccuracyTarget; +use support::lower_promql; // ── harness helpers ───────────────────────────────────────────────────────────── @@ -191,8 +193,11 @@ fn promql_scan_schema_is_open() { // runtime-only, so the binding schema lists only the (ts, value) floor + // referenced labels and may be a subset of the runtime row. let qe = ok("node_cpu_seconds_total"); - let QueryExpr::Scan { schema, .. } = &qe else { - panic!("expected a Scan for a bare selector, got {qe:?}"); + let QueryExpr::TimeRange { child, .. } = &qe else { + panic!("expected a TimeRange for a bare selector, got {qe:?}"); + }; + let QueryExpr::Scan { schema, .. } = child.as_ref() else { + panic!("expected a Scan inside the TimeRange, got {qe:?}"); }; assert!( !schema.closed, @@ -324,7 +329,9 @@ fn sum_by_groups_via_positional_aggregate() { "group keys resolve to positional ColumnIds" ); assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }])); - assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); + assert!( + matches!(child.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); } #[test] @@ -647,7 +654,7 @@ fn unary_negation_lowers_as_multiply_by_minus_one() { }; assert_eq!(*op, BinaryOpKind::Arithmetic(ArithmeticOpKind::Mul)); assert!( - matches!(lhs.as_ref(), QueryExpr::Scan { .. }), + matches!(lhs.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })), "vector on the left" ); assert!( @@ -756,7 +763,7 @@ fn scalar_literal_operand_lowers_as_binaryop_scalar() { }; assert_eq!(*op, BinaryOpKind::Compare(CompareOpKind::Gt)); assert!( - matches!(lhs.as_ref(), QueryExpr::Scan { .. }), + matches!(lhs.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })), "vector on the left" ); assert!( @@ -1198,7 +1205,10 @@ fn offset_modifier_lowers_to_a_time_shift() { // past — a `TimeShift` wrapper over the selector (signed ms; a negative // offset shifts forward). Schema is unchanged (the shift only moves *when*). let qe = ok("http_requests_total offset 5m"); - let QueryExpr::TimeShift { shift, child } = &qe else { + let QueryExpr::TimeRange { child, .. } = &qe else { + panic!("expected an ingestion TimeRange, got {qe:?}"); + }; + let QueryExpr::TimeShift { shift, child } = child.as_ref() else { panic!("expected a TimeShift, got {qe:?}"); }; assert_eq!(shift.offset_ms, 300_000); @@ -1206,7 +1216,10 @@ fn offset_modifier_lowers_to_a_time_shift() { assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); // `offset -5m` shifts forward → negative ms. - let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total offset -5m") else { + let QueryExpr::TimeRange { child, .. } = &ok("http_requests_total offset -5m") else { + panic!("expected an ingestion TimeRange"); + }; + let QueryExpr::TimeShift { shift, .. } = child.as_ref() else { panic!("expected a TimeShift"); }; assert_eq!(shift.offset_ms, -300_000); @@ -1217,20 +1230,29 @@ fn at_modifier_lowers_to_a_time_shift() { // SEMANTICS (PromQL, issue #40): `@ ` pins the evaluation to an absolute // instant (PromQL seconds → IR milliseconds); `@ start()` / `@ end()` anchor // to the query range bounds. - let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total @ 1609746000") else { + let QueryExpr::TimeRange { child, .. } = &ok("http_requests_total @ 1609746000") else { + panic!("expected an ingestion TimeRange"); + }; + let QueryExpr::TimeShift { shift, .. } = child.as_ref() else { panic!("expected a TimeShift for `@ `"); }; assert_eq!(shift.at, Some(AtModifier::Timestamp(1_609_746_000_000))); assert_eq!(shift.offset_ms, 0); - let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total @ start()") else { + let QueryExpr::TimeRange { child, .. } = &ok("http_requests_total @ start()") else { + panic!("expected an ingestion TimeRange"); + }; + let QueryExpr::TimeShift { shift, .. } = child.as_ref() else { panic!("expected a TimeShift for `@ start()`"); }; assert_eq!(shift.at, Some(AtModifier::Start)); // Offset and `@` compose: `@ end() offset 5m` carries both. let qe = ok("http_requests_total @ end() offset 5m"); - let QueryExpr::TimeShift { shift, .. } = &qe else { + let QueryExpr::TimeRange { child, .. } = &qe else { + panic!("expected an ingestion TimeRange, got {qe:?}"); + }; + let QueryExpr::TimeShift { shift, .. } = child.as_ref() else { panic!("expected a TimeShift, got {qe:?}"); }; assert_eq!(shift.at, Some(AtModifier::End)); diff --git a/crates/frontend-promql/tests/promql_equivalence.rs b/crates/frontend-promql/tests/promql_equivalence.rs index 01046322..d1177cc5 100644 --- a/crates/frontend-promql/tests/promql_equivalence.rs +++ b/crates/frontend-promql/tests/promql_equivalence.rs @@ -17,9 +17,10 @@ #![allow(non_snake_case)] -use asap_frontend_promql::lower_promql; +mod support; use asap_types::pre_asap::QueryExpr; use asap_types::types::AccuracyTarget; +use support::lower_promql; fn lo(q: &str) -> QueryExpr { lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("{q:?} should lower: {e}")) diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index 9bc29927..207db3a9 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -8,11 +8,13 @@ use asap_types::pre_asap::{ }; use asap_types::types::AccuracyTarget; use asap_types::workload::{ - AccuracyRequirement, BatchEntry, Predictability, Query, QueryLanguage, QueryRequirements, - QueryWorkload, TimeSelection, + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, + Predictability, Query, QueryLanguage, QueryRequirements, QueryWorkload, TimeSelection, }; -use asap_frontend_promql::{lower_promql, lower_promql_batch, PromqlError as LoweringError}; +use asap_frontend_promql::{lower_promql_workload, PromqlError as LoweringError}; +mod support; +use support::lower_promql; fn lower(q: &str) -> QueryExpr { lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}")) @@ -75,9 +77,12 @@ fn distinct_over_time_preserves_cardinality_accuracy_and_nested_windows() { #[test] fn bare_selector_is_scan_with_predicates() { let qe = lower(r#"http_requests_total{env="prod",status!="500"}"#); + let QueryExpr::TimeRange { child, .. } = &qe else { + panic!("expected TimeRange, got {qe:?}"); + }; let QueryExpr::Scan { source, predicates, .. - } = &qe + } = child.as_ref() else { panic!("expected Scan, got {qe:?}"); }; @@ -93,9 +98,12 @@ fn bare_selector_is_scan_with_predicates() { #[test] fn regex_matcher_lowers_to_regex_compareop() { let qe = lower(r#"http_requests_total{path=~"/api/.*"}"#); + let QueryExpr::TimeRange { child, .. } = &qe else { + panic!("expected TimeRange, got {qe:?}"); + }; let QueryExpr::Scan { predicates, schema, .. - } = &qe + } = child.as_ref() else { panic!("expected Scan, got {qe:?}"); }; @@ -905,55 +913,66 @@ fn scan_schema_carries_ts_value_and_group_keys() { #[test] fn batch_lowers_each_entry_and_reads_per_query_accuracy() { - let workload = QueryWorkload { - language: QueryLanguage::PromQL, - query_batch: Some(vec![ - BatchEntry { - query: Query("rate(a[5m])".into()), - requirements: QueryRequirements::default(), - predictability: Predictability::Unknown, - invocations: 1, - execute_at: None, - time_selection: TimeSelection::default(), - }, - BatchEntry { - query: Query("quantile_over_time(0.9, b[5m])".into()), - requirements: QueryRequirements { - accuracy: AccuracyRequirement::Explicit(AccuracyTarget::Epsilon(0.02)), - ..Default::default() + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![ + BatchEntry { + query: Query("rate(a[5m])".into()), + requirements: QueryRequirements::default(), + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), }, - predictability: Predictability::Unknown, - invocations: 1, - execute_at: None, - time_selection: TimeSelection::default(), + BatchEntry { + query: Query("quantile_over_time(0.9, b[5m])".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(AccuracyTarget::Epsilon(0.02)), + ..Default::default() + }, + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }, + ]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() }, - ]), - repeating_queries: None, + ..Default::default() + }), }; - let results = lower_promql_batch(&workload); + let results = lower_promql_workload(&workload, 0).expect("valid workload"); assert_eq!(results.len(), 2); - assert!(results[0].is_ok()); - assert!(results[1].is_ok()); } #[test] fn batch_rejects_non_promql_language() { use asap_types::workload::SqlDialect; - let workload = QueryWorkload { - language: QueryLanguage::SQL(SqlDialect::DataFusionSQL), - query_batch: Some(vec![BatchEntry { - query: Query("SELECT 1".into()), - requirements: QueryRequirements::default(), - predictability: Predictability::Unknown, - invocations: 1, - execute_at: None, - time_selection: TimeSelection::default(), - }]), - repeating_queries: None, + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::SQL(SqlDialect::DataFusionSQL), + query_batch: Some(vec![BatchEntry { + query: Query("SELECT 1".into()), + requirements: QueryRequirements::default(), + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: None, }; - let results = lower_promql_batch(&workload); - assert_eq!(results.len(), 1); - assert!(matches!(results[0], Err(LoweringError::WrongLanguage(_)))); + assert!(matches!( + lower_promql_workload(&workload, 0), + Err(LoweringError::WrongLanguage(_)) + )); } // ── #12: one home per grouping concept (the canonical `Partition` node is removed) ── @@ -1031,10 +1050,10 @@ fn topk_over_bare_selector_by_label_ranks_per_group() { assert!(!keys[0].ascending, "topk ranks descending"); assert_eq!(partition_by, &vec![2], "job is col 2 in [ts, value, job]"); // No implicit reducing aggregate — the selector is label-preserving, so the - // sort is directly over the Scan (the `job` label survives to partition by). + // sort is directly over the selector horizon (the `job` label survives to partition by). assert!( - matches!(child.as_ref(), QueryExpr::Scan { .. }), - "ranking is over the bare Scan, not a reducing Aggregate, got {child:?}" + matches!(child.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })), + "ranking is over the bare selector horizon, not a reducing Aggregate, got {child:?}" ); assert!( !has_intent(&q, |i| matches!(i, AggIntent::Sum { .. })), @@ -1059,7 +1078,9 @@ fn topk_over_bare_selector_ranks_raw_samples() { panic!("expected Sort, got {child:?}"); }; assert!(partition_by.is_empty(), "no `by` → global ranking"); - assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); + assert!( + matches!(child.as_ref(), QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), QueryExpr::Scan { .. })) + ); assert!(!has_intent(&q, |i| matches!(i, AggIntent::Sum { .. }))); } diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs new file mode 100644 index 00000000..1f15b1ca --- /dev/null +++ b/crates/frontend-promql/tests/support.rs @@ -0,0 +1,52 @@ +use asap_frontend_promql::{ + lower_promql_workload, lower_promql_workload_with_histograms, HistogramCatalog, PromqlError, +}; +use asap_types::pre_asap::QueryExpr; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, + Predictability, Query, QueryLanguage, QueryRequirements, QueryWorkload, TimeSelection, +}; + +fn workload(query: &str, accuracy: AccuracyTarget) -> PlanningWorkload { + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy), + ..Default::default() + }, + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + } +} + +pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Result { + let mut lowered = lower_promql_workload(&workload(query, accuracy), 0)?; + Ok(lowered.remove(0)) +} + +#[allow(dead_code)] +pub fn lower_promql_with_histograms( + query: &str, + accuracy: AccuracyTarget, + histograms: HistogramCatalog, +) -> Result { + let mut lowered = + lower_promql_workload_with_histograms(&workload(query, accuracy), histograms, 0)?; + Ok(lowered.remove(0)) +} diff --git a/crates/frontend-promql/tests/univmon_candidates.rs b/crates/frontend-promql/tests/univmon_candidates.rs index 4afb5539..568e8df2 100644 --- a/crates/frontend-promql/tests/univmon_candidates.rs +++ b/crates/frontend-promql/tests/univmon_candidates.rs @@ -5,13 +5,14 @@ use asap_aware_mapping::accuracy::{ }; use asap_aware_mapping::cost_model::DefaultCostModel; use asap_aware_mapping::{Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG}; -use asap_frontend_promql::lower_promql; +mod support; use asap_types::post_asap::{ compile_executable_dag, cse::share_common_summary_subtrees, AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, ProbabilityExpr, ResultGuarantee, SketchAlgorithm, SketchQuery, SummaryExpr, SummaryFamilyType, SummaryInputExpr, SummaryNode, }; use asap_types::types::AccuracyTarget; +use support::lower_promql; // Synthetic evidence exercises structural sharing, never runtime accuracy. struct TestEvidence; diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs index 5a96aa07..fa867ba7 100644 --- a/crates/integration-tests/src/lib.rs +++ b/crates/integration-tests/src/lib.rs @@ -12,7 +12,48 @@ //! here derives or computes expected outputs. pub mod fixtures { + use asap_frontend_promql::lower_promql_workload; use asap_types::pre_asap::schema::{Column, DataType, Schema}; + use asap_types::pre_asap::QueryExpr; + use asap_types::types::AccuracyTarget; + use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, + Predictability, Query, QueryLanguage, QueryRequirements, QueryWorkload, TimeSelection, + }; + + /// Lower one query through the plan-ready workload API using the test + /// suite's declared one-second source cadence. + pub fn lower_promql( + query: &str, + accuracy: AccuracyTarget, + ) -> Result { + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy), + ..Default::default() + }, + predictability: Predictability::Unknown, + invocations: 1, + execute_at: None, + time_selection: TimeSelection::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + let mut lowered = lower_promql_workload(&workload, 0)?; + Ok(lowered.remove(0)) + } pub fn ts_col() -> Column { Column::new("ts", DataType::Timestamp, false) diff --git a/crates/integration-tests/tests/aggregate.rs b/crates/integration-tests/tests/aggregate.rs index 5395eb71..dd0bef4e 100644 --- a/crates/integration-tests/tests/aggregate.rs +++ b/crates/integration-tests/tests/aggregate.rs @@ -9,8 +9,9 @@ //! and `having: None`. use std::rc::Rc; +use std::time::Duration; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_integration_tests::fixtures::metric_schema; use asap_types::pre_asap::{AggIntent, QueryExpr, Reduction, Source}; use asap_types::types::AccuracyTarget; @@ -35,7 +36,10 @@ fn agg(by: Vec, intent: AggIntent, child: QueryExpr) -> QueryExpr { measures: vec![intent], output_names: vec!["".into()], having: None, - child: Rc::new(child), + child: Rc::new(QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(child), + }), } } diff --git a/crates/integration-tests/tests/binary_op.rs b/crates/integration-tests/tests/binary_op.rs index 406023de..2793808e 100644 --- a/crates/integration-tests/tests/binary_op.rs +++ b/crates/integration-tests/tests/binary_op.rs @@ -8,7 +8,7 @@ use std::rc::Rc; use std::time::Duration; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_integration_tests::fixtures::metric_schema; use asap_types::pre_asap::{ AggIntent, ArithmeticOpKind, BinaryOpKind, CompareOpKind, GroupSide, QueryExpr, Reduction, @@ -21,6 +21,13 @@ fn lower(q: &str) -> QueryExpr { } fn scan(metric: &str, labels: &[&str]) -> QueryExpr { + QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(source_scan(metric, labels)), + } +} + +fn source_scan(metric: &str, labels: &[&str]) -> QueryExpr { QueryExpr::Scan { source: Source::TimeSeries { metric: metric.into(), @@ -38,7 +45,7 @@ fn rate_agg(metric: &str) -> QueryExpr { having: None, child: Rc::new(QueryExpr::TimeRange { range: Duration::from_secs(300), - child: Rc::new(scan(metric, &[])), + child: Rc::new(source_scan(metric, &[])), }), } } diff --git a/crates/integration-tests/tests/cse.rs b/crates/integration-tests/tests/cse.rs index c42226f8..99d9e353 100644 --- a/crates/integration-tests/tests/cse.rs +++ b/crates/integration-tests/tests/cse.rs @@ -25,7 +25,7 @@ use std::rc::Rc; use asap_aware_mapping::{search_workload, Replacement}; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_types::pre_asap::query_expr::QueryExpr; use asap_types::types::AccuracyTarget; diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs index de138d48..79dacb19 100644 --- a/crates/integration-tests/tests/exact_composition.rs +++ b/crates/integration-tests/tests/exact_composition.rs @@ -23,7 +23,7 @@ use asap_aware_mapping::replacement::{ use asap_aware_mapping::{ CostModel, DefaultCostModel, EvaluationRate, ExplanationKind, OperationPlacement, }; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_types::dag_export; use asap_types::post_asap::{ validate_execution_data_states, ExactKind, ExecutionDataState, ExecutionDataStateError, diff --git a/crates/integration-tests/tests/frontend_timestamps.rs b/crates/integration-tests/tests/frontend_timestamps.rs index 9160887b..67c5418c 100644 --- a/crates/integration-tests/tests/frontend_timestamps.rs +++ b/crates/integration-tests/tests/frontend_timestamps.rs @@ -1,7 +1,7 @@ //! Cross-frontend evaluation-time semantics (issues #46 and #184). -use asap_frontend_promql::lower_promql; use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_integration_tests::fixtures::lower_promql; use asap_types::pre_asap::schema::{Column, DataType, Schema}; use asap_types::pre_asap::QueryExpr; use asap_types::types::AccuracyTarget; diff --git a/crates/integration-tests/tests/nested.rs b/crates/integration-tests/tests/nested.rs index 2df9f6e0..2d937082 100644 --- a/crates/integration-tests/tests/nested.rs +++ b/crates/integration-tests/tests/nested.rs @@ -11,7 +11,7 @@ use std::rc::Rc; use std::time::Duration; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_integration_tests::fixtures::metric_schema; use asap_types::pre_asap::{ AggIntent, ArithmeticOpKind, AtModifier, BinaryOpKind, CompareOpKind, GroupKeys, Predicate, @@ -85,7 +85,14 @@ fn q23_sum_by_job_over_filtered_scan() { }))], schema: metric_schema(&["job", "status"]), }; - let expected = agg(vec![2], AggIntent::Sum { col: None }, scan); + let expected = agg( + vec![2], + AggIntent::Sum { col: None }, + QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(scan), + }, + ); assert_eq!( lower(r#"sum by (job) (http_requests_total{status="200"})"#), expected @@ -203,7 +210,14 @@ fn q53_outer_group_key_absent_from_nested_aggregate() { }))], schema: metric_schema(&["group", "job"]), }; - let inner = agg(vec![2], AggIntent::Sum { col: None }, scan); + let inner = agg( + vec![2], + AggIntent::Sum { col: None }, + QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(scan), + }, + ); let expected = agg(vec![], AggIntent::Sum { col: None }, inner); assert_eq!( lower(r#"sum(sum by (group)(http_requests{job="api-server"})) by (job)"#), @@ -220,16 +234,19 @@ fn q53_outer_group_key_absent_from_nested_aggregate() { // parser's default `ignoring([])` match modifier. #[test] fn q52_outer_name_label_over_binary_op() { - let side = |metric: &str, env: &str| QueryExpr::Scan { - source: Source::TimeSeries { - metric: metric.into(), - }, - predicates: vec![Predicate(Rc::new(QueryExpr::Compare { - left: Rc::new(QueryExpr::Column(2)), // env - op: CompareOpKind::Eq, - right: Rc::new(QueryExpr::Literal(ScalarValue::Utf8(env.into()))), - }))], - schema: metric_schema(&["env", "__name__"]), + let side = |metric: &str, env: &str| QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(QueryExpr::Scan { + source: Source::TimeSeries { + metric: metric.into(), + }, + predicates: vec![Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(2)), // env + op: CompareOpKind::Eq, + right: Rc::new(QueryExpr::Literal(ScalarValue::Utf8(env.into()))), + }))], + schema: metric_schema(&["env", "__name__"]), + }), }; let expected = agg( vec![3], // __name__ @@ -326,17 +343,20 @@ fn q40_week_over_week_offset() { // (seconds → ms); a bare selector wrapped in a `TimeShift` carrying the anchor. #[test] fn q40_at_modifier_absolute() { - let expected = QueryExpr::TimeShift { - shift: TimeShift { - offset_ms: 0, - at: Some(AtModifier::Timestamp(1_609_746_000_000)), - }, - child: Rc::new(QueryExpr::Scan { - source: Source::TimeSeries { - metric: "up".into(), + let expected = QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(QueryExpr::TimeShift { + shift: TimeShift { + offset_ms: 0, + at: Some(AtModifier::Timestamp(1_609_746_000_000)), }, - predicates: vec![], - schema: metric_schema(&[]), + child: Rc::new(QueryExpr::Scan { + source: Source::TimeSeries { + metric: "up".into(), + }, + predicates: vec![], + schema: metric_schema(&[]), + }), }), }; assert_eq!(lower("up @ 1609746000"), expected); diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs index 41ea2396..97f847d2 100644 --- a/crates/integration-tests/tests/promql_numeric_regressions.rs +++ b/crates/integration-tests/tests/promql_numeric_regressions.rs @@ -1,7 +1,7 @@ //! Numeric regression fixtures: actual PromQL lowering plus numeric update/readout checks. //! The count/sum interpreter below verifies planner update semantics, not a deployed backend. use asap_aware_mapping::{Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG}; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_types::post_asap::{ compile_executable_dag, ExactKind, SummaryExpr, SummaryFamilyType, SummaryInputExpr, SummaryNode, SummaryUpdate, diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 30b85e6e..3a8afa29 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -18,7 +18,7 @@ use asap_aware_mapping::{ search_workload, search_workload_with_targets, AccuracyModel, Replacement, ReplacementStrategy, ReplacementSubDAG, SketchAlgorithmStrategy, TargetSubDAG, }; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_types::post_asap::{ compile_executable_dag, CandidateCompleteness, CompositionOperator, EdgeRole, EntityIdentity, ExactKind, ExactParams, GroupingStrategy, NonNegativeWeightProof, SketchAlgorithm, SketchKind, diff --git a/crates/integration-tests/tests/scan.rs b/crates/integration-tests/tests/scan.rs index 6a1d0de2..e78fccd9 100644 --- a/crates/integration-tests/tests/scan.rs +++ b/crates/integration-tests/tests/scan.rs @@ -7,8 +7,9 @@ //! Predicates are canonicalized alphabetically by label name at lowering time. use std::rc::Rc; +use std::time::Duration; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_integration_tests::fixtures::metric_schema; use asap_types::pre_asap::{CompareOpKind, Predicate, QueryExpr, ScalarValue, Source}; use asap_types::types::AccuracyTarget; @@ -27,6 +28,13 @@ fn bare_scan(metric: &str, labels: &[&str]) -> QueryExpr { } } +fn instant(child: QueryExpr) -> QueryExpr { + QueryExpr::TimeRange { + range: Duration::from_secs(1), + child: Rc::new(child), + } +} + fn eq_pred(col_id: usize, value: &str) -> Predicate { Predicate(Rc::new(QueryExpr::Compare { left: Rc::new(QueryExpr::Column(col_id)), @@ -64,7 +72,7 @@ fn notregex_pred(col_id: usize, pattern: &str) -> Predicate { fn q01_bare_scan() { assert_eq!( lower("http_requests_total"), - bare_scan("http_requests_total", &[]) + instant(bare_scan("http_requests_total", &[])) ); } @@ -72,13 +80,13 @@ fn q01_bare_scan() { // schema: [ts(0), value(1), job(2)] #[test] fn q02_equality_predicate() { - let expected = QueryExpr::Scan { + let expected = instant(QueryExpr::Scan { source: Source::TimeSeries { metric: "http_requests_total".into(), }, predicates: vec![eq_pred(2, "api-server")], schema: metric_schema(&["job"]), - }; + }); assert_eq!(lower(r#"http_requests_total{job="api-server"}"#), expected); } @@ -86,13 +94,13 @@ fn q02_equality_predicate() { // schema: [ts(0), value(1), status(2)] #[test] fn q03_inequality_predicate() { - let expected = QueryExpr::Scan { + let expected = instant(QueryExpr::Scan { source: Source::TimeSeries { metric: "http_requests_total".into(), }, predicates: vec![ne_pred(2, "500")], schema: metric_schema(&["status"]), - }; + }); assert_eq!(lower(r#"http_requests_total{status!="500"}"#), expected); } @@ -100,13 +108,13 @@ fn q03_inequality_predicate() { // schema: [ts(0), value(1), job(2)] #[test] fn q04_regex_predicate() { - let expected = QueryExpr::Scan { + let expected = instant(QueryExpr::Scan { source: Source::TimeSeries { metric: "http_requests_total".into(), }, predicates: vec![regex_pred(2, "api.*")], schema: metric_schema(&["job"]), - }; + }); assert_eq!(lower(r#"http_requests_total{job=~"api.*"}"#), expected); } @@ -114,13 +122,13 @@ fn q04_regex_predicate() { // schema: [ts(0), value(1), job(2)] #[test] fn q_notregex_predicate() { - let expected = QueryExpr::Scan { + let expected = instant(QueryExpr::Scan { source: Source::TimeSeries { metric: "http_requests_total".into(), }, predicates: vec![notregex_pred(2, "internal.*")], schema: metric_schema(&["job"]), - }; + }); assert_eq!(lower(r#"http_requests_total{job!~"internal.*"}"#), expected); } @@ -129,13 +137,13 @@ fn q_notregex_predicate() { // predicates in same alphabetical order: job first, then status #[test] fn q_multi_two_predicates() { - let expected = QueryExpr::Scan { + let expected = instant(QueryExpr::Scan { source: Source::TimeSeries { metric: "http_requests_total".into(), }, predicates: vec![eq_pred(2, "api-server"), ne_pred(3, "500")], schema: metric_schema(&["job", "status"]), - }; + }); assert_eq!( lower(r#"http_requests_total{job="api-server",status!="500"}"#), expected, diff --git a/crates/integration-tests/tests/schema.rs b/crates/integration-tests/tests/schema.rs index 4db05eba..6024619c 100644 --- a/crates/integration-tests/tests/schema.rs +++ b/crates/integration-tests/tests/schema.rs @@ -9,7 +9,7 @@ //! `Aggregate` or a `Project`. Per-series reductions (`rate`, `*_over_time`) //! are label-preserving and keep the schema open. -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_types::types::AccuracyTarget; fn lower(q: &str) -> asap_types::pre_asap::QueryExpr { diff --git a/crates/integration-tests/tests/summary_maintenance_lifecycle_e2e.rs b/crates/integration-tests/tests/summary_maintenance_lifecycle_e2e.rs index a1b31d33..6ab6bcae 100644 --- a/crates/integration-tests/tests/summary_maintenance_lifecycle_e2e.rs +++ b/crates/integration-tests/tests/summary_maintenance_lifecycle_e2e.rs @@ -12,16 +12,17 @@ use asap_aware_mapping::{ SummaryMaintenanceCapabilities, SummaryMaintenanceLifecycleCapabilities, SummaryMaintenanceLifecycleCostInputs, SummaryMaintenanceLifecycleRejection, WorkloadDemand, }; -use asap_frontend_promql::lower_promql_batch; +use asap_frontend_promql::lower_promql_workload; use asap_types::post_asap::{ EvaluationSchedule, SummaryMaintenanceLifecycle, SummaryMaintenanceMode, SummaryNode, }; use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::types::AccuracyTarget; use asap_types::workload::{ - AccuracyRequirement, BatchEntry, DataArrival, DataWorkload, Evidence, EvidenceSource, - PlanningWorkload, Predictability, Query, QueryLanguage, QueryRequirements, QueryTimeScope, - QueryWorkload, Rate, RepeatedDemand, RepeatingEntry, RepetitionInterval, TimeSelection, + AccuracyRequirement, BatchEntry, DataArrival, DataWorkload, DurationMs, Evidence, + EvidenceSource, PlanningWorkload, Predictability, Query, QueryLanguage, QueryRequirements, + QueryTimeScope, QueryWorkload, Rate, RepeatedDemand, RepeatingEntry, RepetitionInterval, + TimeSelection, }; const NOW_MS: u64 = 1_000_000; @@ -106,7 +107,10 @@ fn dashboard_workload() -> PlanningWorkload { observed_at_ms: Some(NOW_MS), valid_for_ms: Some(60_000), }, - + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, ..DataWorkload::default() }), } @@ -117,11 +121,11 @@ fn promql_dashboard_materializes_continuous_summary_with_explained_rejections() let workload = dashboard_workload(); workload.validate().unwrap(); - let lowered = lower_promql_batch(&workload.query_workload) + let lowered = lower_promql_workload(&workload, 0) + .expect("valid PromQL workload") .into_iter() .next() - .expect("one normalized workload entry") - .expect("valid PromQL"); + .expect("one normalized workload entry"); let root = Rc::new(lowered); let strategies = asap_aware_mapping::default_strategies_with(&FullyCostedRuntime); let space = search_workload_with(vec![("dashboard", Rc::clone(&root))], &strategies); diff --git a/crates/integration-tests/tests/time_range.rs b/crates/integration-tests/tests/time_range.rs index 34fdd78e..940facc3 100644 --- a/crates/integration-tests/tests/time_range.rs +++ b/crates/integration-tests/tests/time_range.rs @@ -10,7 +10,7 @@ use std::rc::Rc; use std::time::Duration; -use asap_frontend_promql::lower_promql; +use asap_integration_tests::fixtures::lower_promql; use asap_integration_tests::fixtures::metric_schema; use asap_types::pre_asap::{AggIntent, QueryExpr, Reduction, Source}; use asap_types::types::AccuracyTarget; diff --git a/crates/types/src/post_asap/maintained_population.rs b/crates/types/src/post_asap/maintained_population.rs index fc953d53..30dacf82 100644 --- a/crates/types/src/post_asap/maintained_population.rs +++ b/crates/types/src/post_asap/maintained_population.rs @@ -39,6 +39,20 @@ impl CurrentSeriesInput { /// Verify the named contract against the canonical maintenance input. pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool { use crate::pre_asap::{CompareOpKind, DataType, QueryExpr, ScalarValue, Source}; + // PromQL instant selectors carry an ingestion-interval `TimeRange` as + // their input scope. The population must use the same expiry horizon; + // shifted and otherwise transformed inputs still fail below. + let input = match input { + QueryExpr::TimeRange { range, child } + if self.lookback_ms > 0 + && *range == std::time::Duration::from_millis(self.lookback_ms) => + { + child.as_ref() + } + QueryExpr::TimeRange { .. } => return false, + other if self.lookback_ms == 300_000 => other, + _ => return false, + }; let QueryExpr::Scan { source: Source::TimeSeries { metric }, predicates, @@ -51,7 +65,6 @@ impl CurrentSeriesInput { || *metric != self.metric || schema.closed || schema.time_index.is_none() - || self.lookback_ms != 300_000 { return false; } diff --git a/crates/types/src/workload.rs b/crates/types/src/workload.rs index df576eab..5e1a5d48 100644 --- a/crates/types/src/workload.rs +++ b/crates/types/src/workload.rs @@ -530,6 +530,10 @@ pub struct Rate(pub f64); #[serde(deny_unknown_fields)] pub struct DataWorkload { pub arrival: DataArrival, + /// Cadence at which each PromQL source supplies a sample. PromQL instant + /// selectors use this as their explicit selection horizon. + #[serde(default)] + pub data_ingestion_interval: Evidence, pub ingestion_volume: Evidence, pub ingestion_rate: Evidence, pub input_cardinality: Evidence, @@ -604,6 +608,19 @@ impl PlanningWorkload { if let Some(data) = &self.data_workload { data.validate()?; } + if matches!(self.query_workload.language, QueryLanguage::PromQL) { + let data = self + .data_workload + .as_ref() + .ok_or(WorkloadError::MissingPromqlDataWorkload)?; + let interval = data + .data_ingestion_interval + .value + .ok_or(WorkloadError::MissingDataIngestionInterval)?; + if interval.0 == 0 { + return Err(WorkloadError::ZeroDataIngestionInterval); + } + } Ok(()) } } @@ -656,6 +673,14 @@ fn validate_rate(rate: Rate) -> Result { #[derive(Debug, Clone, Copy, PartialEq, thiserror::Error)] pub enum WorkloadError { + #[error("data_ingestion_interval evidence is unavailable at planning time")] + UnavailableDataIngestionInterval, + #[error("a PromQL workload requires data_workload")] + MissingPromqlDataWorkload, + #[error("a PromQL workload requires data_ingestion_interval")] + MissingDataIngestionInterval, + #[error("data_ingestion_interval must be greater than zero")] + ZeroDataIngestionInterval, #[error("a one-time query must have at least one invocation")] ZeroInvocations, #[error("a fixed repetition interval must be greater than zero")] diff --git a/crates/types/tests/workload_compatibility.rs b/crates/types/tests/workload_compatibility.rs new file mode 100644 index 00000000..e5c31a5b --- /dev/null +++ b/crates/types/tests/workload_compatibility.rs @@ -0,0 +1,35 @@ +use asap_types::workload::{ + DataWorkload, PlanningWorkload, QueryLanguage, QueryWorkload, SqlDialect, +}; + +// Non-PromQL wire workloads should not require a PromQL-only cadence fact. +#[test] +fn non_promql_payload_accepts_absent_ingestion_interval() { + for language in [ + QueryLanguage::DataFusion, + QueryLanguage::SQL(SqlDialect::DataFusionSQL), + ] { + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language, + query_batch: None, + repeating_queries: None, + }, + data_workload: Some(DataWorkload::default()), + }; + workload.validate().unwrap(); + let mut payload = serde_json::to_value(&workload).unwrap(); + payload["data_workload"] + .as_object_mut() + .unwrap() + .remove("data_ingestion_interval"); + let decoded = serde_json::from_value::(payload.clone()).unwrap(); + decoded.validate().unwrap(); + payload["query_workload"]["language"] = serde_json::json!("promql"); + let promql = serde_json::from_value::(payload).unwrap(); + assert_eq!( + promql.validate(), + Err(asap_types::workload::WorkloadError::MissingDataIngestionInterval) + ); + } +} diff --git a/docs/develop_docs/library-api.md b/docs/develop_docs/library-api.md index 8b899137..9a3d155f 100644 --- a/docs/develop_docs/library-api.md +++ b/docs/develop_docs/library-api.md @@ -38,29 +38,39 @@ asap-types = { git = "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/ProjectASAP/ASAPPlanner", rev = "e7fdb2 | Public function | Required input | Output | | --- | --- | --- | -| `asap_frontend_promql::lower_promql` | Query string, `AccuracyTarget` | `Result` | +| `asap_frontend_promql::lower_promql_workload` | PromQL `PlanningWorkload` with a nonzero `data_ingestion_interval` | All-or-nothing `Result, PromqlError>` for normalized batch and repeating entries | | `asap_frontend_metricsql::lower_metricsql` | Query string, `AccuracyTarget` | `Result` | | `asap_frontend_sql::lower_sql` | Query string, `SqlCatalog`, accuracy | Async `Result`; default SQL dialect is DataFusionSQL | | `asap_frontend_sql::lower_sql_dialect` | Same inputs plus `SqlDialect` | Async resolved Pre-ASAP query or error | -| `lower_promql_batch` / `lower_sql_batch` in their frontend crates | `QueryWorkload`; SQL additionally needs catalog | Per-query results for `query_batch`; these helpers do not iterate `repeating_queries` | +| `asap_frontend_sql::lower_sql_batch` | `QueryWorkload` and catalog | Per-query results for `query_batch`; does not iterate `repeating_queries` | Lowering resolves the supported source language into the canonical query representation. It does not enumerate Post-ASAP alternatives. A frontend may reject unsupported syntax or semantics; a declared language/dialect enum does -not imply complete support. For mixed one-time/repeating workloads, use normalized -`QueryWorkload::entries()` and the appropriate single-query frontend, preserving -entry-to-root associations for later workload-aware operations. +not imply complete support. PromQL workload lowering uses normalized +`PlanningWorkload::query_workload.entries()` order, preserving entry-to-root associations for later +workload-aware operations. For SQL mixed one-time/repeating workloads, iterate +those entries with the single-query frontend. ### Definition and example PromQL's public signature (types are imported from their respective crates): ```text -lower_promql(query: &str, accuracy: AccuracyTarget) - -> Result +lower_promql_workload(workload: &PlanningWorkload, now_ms: u64) + -> Result, PromqlError> ``` -`query` and `accuracy` are required. These are the accuracy argument's choices: +`DataWorkload.data_ingestion_interval` must contain a nonzero `Evidence`. +Pass the actual planning time as `now_ms` (Unix milliseconds), consistently with +downstream lifecycle planning. Expired or future cadence evidence is rejected, +as is expiring evidence without an observation timestamp. The histogram variant +takes the same timestamp after its histogram catalog argument. The examples use +`0` only because their explicitly supplied cadence is timeless. +Bare instant selectors receive this selection horizon; explicit range selectors +retain their query-specified range. Use `lower_promql_workload_with_histograms` +to supply a histogram catalog for the whole workload. Each entry carries its +own accuracy requirement, with these explicit target choices: | Value | Meaning | Example | | --- | --- | --- | @@ -73,11 +83,39 @@ candidate's guarantee and its error metric; a target is a requirement, not proof that a supported candidate exists. ```rust -use asap_frontend_promql::lower_promql; +use asap_frontend_promql::lower_promql_workload; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, Query, + PlanningWorkload, QueryLanguage, QueryRequirements, QueryWorkload, +}; use asap_types::types::AccuracyTarget; fn main() -> Result<(), Box> { - let pre_asap = lower_promql("sum(latency)", AccuracyTarget::Exact)?; + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query("sum(latency)".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(AccuracyTarget::Exact), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + let pre_asap = lower_promql_workload(&workload, 0)?; println!("{pre_asap:#?}"); Ok(()) } @@ -149,7 +187,11 @@ The default cost model is suitable for inspection, not deployment calibration. ```rust use std::rc::Rc; -use asap_frontend_promql::lower_promql; +use asap_frontend_promql::lower_promql_workload; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, Query, + PlanningWorkload, QueryLanguage, QueryRequirements, QueryWorkload, +}; use asap_aware_mapping::{ default_strategies_with, search_workload_with_targets, DefaultAccuracyModel, DefaultCostModel, @@ -158,7 +200,31 @@ use asap_types::types::AccuracyTarget; fn main() -> Result<(), Box> { let accuracy = AccuracyTarget::Epsilon(0.01); - let root = Rc::new(lower_promql("quantile(0.99, latency)", accuracy.clone())?); + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query("quantile(0.99, latency)".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy.clone()), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + let root = Rc::new(lower_promql_workload(&workload, 0)?.remove(0)); let cost_model = DefaultCostModel; let strategies = default_strategies_with(&cost_model); let space = search_workload_with_targets( @@ -247,7 +313,11 @@ replacement::default_strategies_with_evidence<'a>( ```rust use std::rc::Rc; -use asap_frontend_promql::lower_promql; +use asap_frontend_promql::lower_promql_workload; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, Query, + PlanningWorkload, QueryLanguage, QueryRequirements, QueryWorkload, +}; use asap_aware_mapping::{ search_workload_with_targets, DefaultAccuracyModel, DefaultCostModel, ReplacementStrategy, SketchAlgorithmStrategy, SharedSubtreeStrategy, @@ -256,7 +326,31 @@ use asap_types::types::AccuracyTarget; fn main() -> Result<(), Box> { let accuracy = AccuracyTarget::Epsilon(0.01); - let root = Rc::new(lower_promql("quantile(0.99, latency)", accuracy.clone())?); + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query("quantile(0.99, latency)".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy.clone()), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + let root = Rc::new(lower_promql_workload(&workload, 0)?.remove(0)); let model = DefaultCostModel; let strategies: Vec> = vec![ Box::new(SketchAlgorithmStrategy::new(&model)), @@ -575,12 +669,40 @@ Use lifecycle-aware selection above when the comparison needs those decisions. ```rust use std::rc::Rc; -use asap_frontend_promql::lower_promql; +use asap_frontend_promql::lower_promql_workload; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, Query, + PlanningWorkload, QueryLanguage, QueryRequirements, QueryWorkload, +}; use asap_aware_mapping::{search_workload, DefaultCostModel}; use asap_types::types::AccuracyTarget; fn main() -> Result<(), Box> { - let root = Rc::new(lower_promql("sum(latency)", AccuracyTarget::Exact)?); + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query("sum(latency)".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(AccuracyTarget::Exact), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1_000)), + ..Default::default() + }, + ..Default::default() + }), + }; + let root = Rc::new(lower_promql_workload(&workload, 0)?.remove(0)); let space = search_workload(vec![("q1", root)]); let selection = space.global_selection(&DefaultCostModel); // Search may canonicalize roots; use the root returned by PlanSpace. diff --git a/docs/user_guide_docs/run-a-query.md b/docs/user_guide_docs/run-a-query.md index b5eddddb..31fe139e 100644 --- a/docs/user_guide_docs/run-a-query.md +++ b/docs/user_guide_docs/run-a-query.md @@ -15,14 +15,14 @@ tool on first use. | Command (`cargo run -p asap-devtools --bin … -- …`) | Input / options | Result | | --- | --- | --- | -| `show_pre_asap_ir queries.txt` | File path, or stdin when omitted | Prints canonical Pre-ASAP IR | -| `show_post_asap_ir queries.txt` | Same query file format | Prints all sketch-strategy Post-ASAP candidates using a fixed approximate target, in cost-model order | -| `dag_export --promql ""` | One PromQL expression | Exports a query graph for inspection | +| `show_pre_asap_ir --data-ingestion-interval-ms 1000 queries.txt` | File path, or stdin when omitted | Prints canonical Pre-ASAP IR | +| `show_post_asap_ir --data-ingestion-interval-ms 1000 queries.txt` | Same query file format | Prints all sketch-strategy Post-ASAP candidates using a fixed approximate target, in cost-model order | +| `dag_export --data-ingestion-interval-ms 1000 --promql ""` | One PromQL expression | Exports a query graph for inspection | | `dag_export --sql ""` | One SQL expression using the tool's catalog | Exports a query graph for inspection | -| `analyze_corpora --corpora --out-dir ` | Repository PromQL corpora, output directory | Writes successful/error IR dumps and summary reports | +| `analyze_corpora --corpora --data-ingestion-interval-ms 1000 --out-dir ` | Repository PromQL corpora, output directory | Writes successful/error IR dumps and summary reports | | `analyze_corpora --sql-corpora --out-dir ` | Repository SQL corpora, output directory | Writes SQL corpus reports | -| `variant_coverage` | Repository corpora | Reports Pre-ASAP IR variant coverage | -| `sketch_coverage --epsilon 0.01` | Repository corpora; epsilon defaults to `0.01` | Reports sketch/reuse opportunities among successfully lowered queries | +| `variant_coverage --data-ingestion-interval-ms 1000` | Repository corpora | Reports Pre-ASAP IR variant coverage | +| `sketch_coverage --data-ingestion-interval-ms 1000 --epsilon 0.01` | Repository corpora; epsilon defaults to `0.01` | Reports sketch/reuse opportunities among successfully lowered queries | ## Inspect a query from the command line @@ -51,19 +51,19 @@ For your own schema, provide a `SqlCatalog` through the library API. Run: ```sh -cargo run -p asap-devtools --bin show_pre_asap_ir -- queries.txt +cargo run -p asap-devtools --bin show_pre_asap_ir -- --data-ingestion-interval-ms 1000 queries.txt ``` You can also provide the queries through stdin: ```sh -cargo run -p asap-devtools --bin show_pre_asap_ir < queries.txt +cargo run -p asap-devtools --bin show_pre_asap_ir -- --data-ingestion-interval-ms 1000 < queries.txt ``` To dump and compare every PromQL corpus, run: ```sh -cargo run -p asap-devtools --bin analyze_corpora -- --corpora --out-dir artifacts/promql_pre_asap +cargo run -p asap-devtools --bin analyze_corpora -- --corpora --data-ingestion-interval-ms 1000 --out-dir artifacts/promql_pre_asap ``` This writes one successful-lowering JSONL dump and one error JSONL dump per @@ -83,13 +83,13 @@ cargo run -p asap-devtools --bin analyze_corpora -- --sql-corpora --out-dir arti Run: ```sh -cargo run -p asap-devtools --bin show_post_asap_ir -- queries.txt +cargo run -p asap-devtools --bin show_post_asap_ir -- --data-ingestion-interval-ms 1000 queries.txt ``` Or through stdin: ```sh -cargo run -p asap-devtools --bin show_post_asap_ir < queries.txt +cargo run -p asap-devtools --bin show_post_asap_ir -- --data-ingestion-interval-ms 1000 < queries.txt ``` `show_post_asap_ir` uses an approximation target of ε = 0.01 and prints every @@ -119,7 +119,7 @@ cargo run -p asap-devtools --bin dag_export -- --sql "" or: ```sh -cargo run -p asap-devtools --bin dag_export -- --promql "" +cargo run -p asap-devtools --bin dag_export -- --data-ingestion-interval-ms 1000 --promql "" ``` See [`tools/dag-viewer/RUNNING.md`](../../tools/dag-viewer/RUNNING.md) for instructions on running the DAG viewer. @@ -129,7 +129,7 @@ See [`tools/dag-viewer/RUNNING.md`](../../tools/dag-viewer/RUNNING.md) for instr Parse the query corpora in the repository and report which pre-ASAP IR variants are exercised: ```sh -cargo run -p asap-devtools --bin variant_coverage +cargo run -p asap-devtools --bin variant_coverage -- --data-ingestion-interval-ms 1000 ``` ### Check sketch-replacement coverage @@ -137,7 +137,7 @@ cargo run -p asap-devtools --bin variant_coverage Parse the same query corpora, lower them with an approximate `AccuracyTarget`, and report what fraction of each corpus's successfully-lowered queries got a genuine sketch alternative (`SketchApproximation`, e.g. KLL vs. DDSketch) and/or a cross-query common-subexpression-reuse candidate (`CommonSubexpressionReuse`): ```sh -cargo run -p asap-devtools --bin sketch_coverage -- --epsilon 0.01 +cargo run -p asap-devtools --bin sketch_coverage -- --data-ingestion-interval-ms 1000 --epsilon 0.01 ``` `--epsilon` is optional (defaults to `0.01`). See the binary's own doc comment for exactly how "coverage" is defined and attributed back to each query. @@ -163,3 +163,8 @@ cargo run -p asap-devtools --example canonical_examples - [Design overview](../design_docs/README.md) - [Pre-ASAP IR reference](../develop_docs/pre-asap-ir.md) - [Post-ASAP IR reference](../design_docs/concepts/post-asap-ir.md) + +PromQL commands require `--data-ingestion-interval-ms` with the nonzero source +sample cadence in milliseconds. The examples use a one-second cadence; supply +the interval for your data. The mixed-input `show_pre_asap_ir` and +`show_post_asap_ir` tools require this option even for SQL-only input files.