diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 782689870..72e24d63d 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -163,40 +163,34 @@ fn display_cap_is_recorded_and_keeps_the_selection() { ); } -/// A choice composition cannot substitute is listed with its error, and the -/// other candidates are still written. +/// A top-k partitioned by every label of its input has no item, so a heap +/// over it could not be composed; such a heap reads one row per item and +/// is pruned in Pass 1 (`TopKHeapOverOneRowPerItem`), listed under +/// `pass1_pruned`, and every candidate left composes. #[test] -fn uncomposable_candidate_is_recorded_with_its_error() { - let out = - std::env::temp_dir().join(format!("stage_pipeline_{}_error.json", std::process::id())); - let status = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) - .args([ - "--promql", - // Top-K partitioned by every label of its input has no item. - "topk by (job) (3, sum by (job) (rate(m[5m])))", - "--max-error", - "0.01", - "--failure-probability", - "0.001", - "--out", - ]) - .arg(&out) - .status() - .unwrap(); - assert!(status.success()); - let document: Value = serde_json::from_str(&std::fs::read_to_string(&out).unwrap()).unwrap(); - std::fs::remove_file(out).unwrap(); - let candidates = document["stage1_logical_asap_optimization"]["candidates"] - .as_array() - .unwrap(); - let (composed, failed): (Vec<_>, Vec<_>) = candidates - .iter() - .partition(|candidate| candidate.get("dag").is_some()); - assert!(!composed.is_empty() && !failed.is_empty()); - for candidate in failed { - assert!(candidate["error"].as_str().unwrap().contains("Top-K")); - } - for candidate in composed { +fn a_topk_heap_over_one_row_per_item_is_pruned_with_its_reason() { + let document = document_of(&[ + "--promql", + "topk by (job) (3, sum by (job) (rate(m[5m])))", + "--max-error", + "0.01", + "--failure-probability", + "0.001", + ]); + let stage1 = &document["stage1_logical_asap_optimization"]; + assert_eq!( + stage1["pass1_pruned"], + serde_json::json!([{ + "target": 0, + "query": "Q1", + "pruned": [ + { "candidate": "CMS+heap", "reason": "TopKHeapOverOneRowPerItem" }, + { "candidate": "CountSketch+heap", "reason": "TopKHeapOverOneRowPerItem" }, + ], + "fused_pruned": [], + }]) + ); + for candidate in stage1["candidates"].as_array().unwrap() { validated_dag(&candidate["dag"], 1); } } @@ -405,12 +399,13 @@ fn workload_above_the_candidate_limit_is_rejected() { ); } -/// The default limit is 10^4: nine approximate Top-K queries (3^9 = 19683 -/// Pass 1 combinations) are rejected once the count passes it. +/// The default limit is 10^4: nine approximate quantile queries, three +/// candidates each (3^9 = 19683 Pass 1 combinations), are rejected once the +/// count passes it. #[test] fn default_candidate_limit_is_ten_thousand() { let queries: Vec = (1..=9) - .map(|i| format!("topk(5, rate(m{i}[5m]))")) + .map(|i| format!("quantile_over_time(0.99, m{i}[5m])")) .collect(); let mut args = Vec::new(); for query in &queries { diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index dfdd42b3a..da8df64f0 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -21,7 +21,10 @@ //! grouped `sum` over a `sum` (`ExactAccumulator([Sum])`). //! - Summary candidates are unsized; they are sized after Pass 2 sharing. //! - Candidates are pruned only when provably unable to meet the accuracy -//! requirement, and each pruned candidate keeps its [`PruneReason`]. +//! requirement or to run, or when the exact candidate provably dominates +//! them (a PromQL top-k's heap over its instant vector, one row per item: +//! [`PruneReason::TopKHeapOverOneRowPerItem`]); each pruned candidate +//! keeps its [`PruneReason`]. //! - A measure filter or `HAVING` prunes nothing. The target records the //! measure filter (applied as `SummaryAgg.filter`) and one [`HavingEntry`] //! per consumer `Filter` directly above the aggregate: a key-only predicate @@ -178,6 +181,14 @@ pub enum PruneReason { /// +Inf: no sketch estimates that, and the executor's sketches refuse /// such a φ, so only the exact answer remains. QuantileOutsideUnitInterval, + /// A PromQL top-k's heap (Count-Min or CountSketch with a heap) over its + /// own input, an instant vector with one row per ranked series or + /// group: the heap reads every row and adds + /// nothing to it, so an exact top-k over the same rows, which keeps only + /// `k` of them, is exact and no costlier. Only the whole-expression heap + /// over the inner aggregate's input, where an item's rows repeat, is + /// offered ([`FusedComputation`]). + TopKHeapOverOneRowPerItem, } /// Unranked candidates of one measure and the candidates pruned for it. @@ -671,6 +682,7 @@ pub fn generate_per_computation_candidates( } } add_whole_expression_targets(&mut targets, &consumers); + prune_topk_heaps_over_one_row_per_item(&mut targets); add_nested_sum_targets(&mut targets, &consumers); Ok(PerComputationCandidates { roots, targets }) } @@ -855,6 +867,46 @@ fn add_whole_expression_targets( } } +/// Prunes the own heap candidates of each PromQL top-k +/// ([`PruneReason::TopKHeapOverOneRowPerItem`]); a fused whole-expression +/// heap ([`add_whole_expression_targets`]) has already moved out of them. +fn prune_topk_heaps_over_one_row_per_item(targets: &mut [ComputationTarget]) { + for target in targets.iter_mut() { + let Some(NonASAPOp::Aggregate { + measures, child, .. + }) = target.aggregate.non_asap() + else { + continue; + }; + if !matches!(measures.get(target.measure), Some(AggIntent::TopK { .. })) { + continue; + } + // A PromQL top-k ranks an instant vector: one row per series or + // group. (A SQL top-k keeps its heaps.) + if !reads_only_time_series(child) { + continue; + } + target + .choices + .prune(PruneReason::TopKHeapOverOneRowPerItem, |candidate| { + matches!( + candidate, + ComputationCandidate::Summary( + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ) + ) + }); + } +} + +/// Whether every scan below `node` reads a time series (PromQL rows). +fn reads_only_time_series(node: &OperatorNode) -> bool { + match node.non_asap() { + Some(NonASAPOp::Scan { source, .. }) => matches!(source, Source::TimeSeries { .. }), + _ => node.children().iter().all(|c| reads_only_time_series(c)), + } +} + /// The input, update and partitions of a whole-expression top-k over /// `target`'s inner aggregate, by the keyed-additive rule, or `None` when it /// does not apply. Rows that carry the full series identity rank it as a diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 988dca581..8939da9fd 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -75,7 +75,8 @@ //! columns, so an exact root has the same column names as a summary //! candidate's and as the frontend declares. Over PromQL rows it also keeps //! the series identity, after them: `topk` answers whole input series, so -//! each answer carries the chosen series' full label set. +//! each answer carries the chosen series' full label set. PromQL rows without +//! one (e.g. of `sum by (job)`) keep their other string label columns instead. //! `topk_k` carries the ranked value's type and nullability (the frontend //! declares a non-null `Utf8` placeholder); casting the value to `Utf8` would //! not make it the same answer. @@ -96,7 +97,7 @@ use asap_types::ir::physical_export::{ use asap_types::ir::properties::{ExecutionDataStateError, ExecutionTiming}; use asap_types::ir::scalar::column_resolution::resolve_column_ref; use asap_types::ir::scalar::expr_ir::ColumnRef; -use asap_types::ir::schema::{ExactKind, FieldDataType}; +use asap_types::ir::schema::{DataType, ExactKind, FieldDataType}; use asap_types::ir::{ apply_materialization_timings, is_filtered_build, split_shared_by_timing, ASAPOp, MaterializationAssignment, NonASAPOp, Operator, OperatorNode, ProjectItem, QueryRoot, @@ -779,6 +780,20 @@ fn implement_topk( alias: Some(asap_types::ir::schema::PROMQL_SERIES_IDENTITY.into()), expr: ScalarExpr::Column(identity), }); + } else if reads_only_time_series(&child) { + // PromQL rows without it (e.g. those of `sum by (job)`) carry their + // labels as plain string columns: each answer keeps the ones the + // logical columns do not, so `topk(k, sum by (job) (...))` answers + // each job. A SQL result keeps only its logical columns. + for (index, field) in child.schema.fields.iter().enumerate() { + let is_label = field.dtype == FieldDataType::Plain(DataType::Utf8); + if is_label && index != value && !key_fields.iter().any(|k| k.name == field.name) { + cols.push(ProjectItem { + alias: Some(field.name.clone()), + expr: ScalarExpr::Column(index), + }); + } + } } Ok(Rc::new( OperatorNode::new(Operator::NonASAP(NonASAPOp::Project { @@ -888,7 +903,8 @@ mod tests { let root = lower_promql("topk(5, sum by (job) (m))", AccuracyTarget::exact()); let candidate = first_candidate(&[root.into()]); let physical = operator_root(&candidate, 0); - assert_eq!(names(&physical), ["topk_5"]); + // The answers keep their `job` label. + assert_eq!(names(&physical), ["topk_5", "job"]); let limit = topk_limit(&physical); assert_eq!(names(&limit), ["job", "sum"]); assert_eq!(topk_sort_key(&limit), (ScalarExpr::Column(1), false)); diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 04847750a..d87152045 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -269,9 +269,9 @@ pub mod pane_sharing; /// The Stage 1 to Stage 3 driver and the optimization-pass interface. pub mod pass; pub use pass::{ - optimize, run_stage1_to_stage3, run_stage1_to_stage3_with, OptimizationInput, - OptimizationInputError, OptimizationPass, OptimizeError, PassNameConflict, PassRegistry, - PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, + optimize, run_stage1_to_stage3, run_stage1_to_stage3_with, ExecutorCompileCheck, + OptimizationInput, OptimizationInputError, OptimizationPass, OptimizeError, PassNameConflict, + PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, Stage1ToStage3Search, }; @@ -749,6 +749,11 @@ pub enum RejectionCause { /// executor cannot compile it /// ([`asap_types::ir::operator::agg_has_exact_per_entity_kernel`]). PerEntityAggregateNotExecutable, + /// The deployment's executor cannot compile its plan + /// ([`PlanningModels::executor_compile_check`]), e.g. a summary readout + /// whose output shape the executor does not build. The reason is the + /// executor's error. + ExecutorCannotCompile, /// A query's output has no accuracy guarantee (Stage 1 has no /// analytical rule for it and the deployment's evidence supplies none), /// and the deployment's accuracy model did not accept it directly @@ -791,6 +796,7 @@ impl RejectionCause { | Self::AccuracyTargetMissed | Self::CountMinWeightsNotProvenNonNegative | Self::PerEntityAggregateNotExecutable + | Self::ExecutorCannotCompile | Self::NoAccuracyGuarantee | Self::RawDataNotRetained | Self::LatencyBoundExceeded @@ -1208,6 +1214,17 @@ impl<'a> PlanSelector<'a> { if !selectable { return Ok(()); } + } + // Only a candidate that may be selected is compiled: the check runs + // the executor's own compiler, the costliest check. + if let Some(check) = self.models.executor_compile_check { + if let Err(reason) = check.compiles(&sized.roots) { + self.reject(&candidate.id, RejectionCause::ExecutorCannotCompile, reason); + return Ok(()); + } + } + if !self.keep_every_candidate { + let kept_cost = |kept: &KeptCandidate| &self.costs[&kept.built.id]; let dominated: Vec = self .kept .iter() @@ -5381,6 +5398,56 @@ mod tests { } } + /// An executor that compiles every plan but the one it is first asked + /// about. + struct RefusesTheFirstPlan(std::cell::Cell); + + impl crate::ExecutorCompileCheck for RefusesTheFirstPlan { + fn compiles(&self, _roots: &[QueryRoot]) -> Result<(), String> { + if self.0.replace(true) { + Ok(()) + } else { + Err("invalid keyed evaluation shape or capacity".into()) + } + } + } + + /// A candidate the deployment's executor does not compile is rejected + /// as `ExecutorCannotCompile` with the executor's reason, and another + /// is selected; without the check the same candidate is valid. + #[test] + fn a_plan_the_executor_cannot_compile_is_rejected() { + let target = AccuracyTarget::max_error(0.01) + .failure_probability(0.001) + .accuracy_metric(AccuracyMetric::ABSOLUTE); + let entries = [entry_with(target, repeating(10_000))]; + // More than one valid candidate: every output accepted. + let settings = PlanSelectionSettings { + ignore_accuracy_guarantees: true, + ..PlanSelectionSettings::default() + }; + let models = PlanningModels::builtin().with_plan_selection(&settings); + let unchecked = select_plan(&candidates(), &entries, &data(), models).unwrap(); + let check = RefusesTheFirstPlan(std::cell::Cell::new(false)); + let checked = select_plan( + &candidates(), + &entries, + &data(), + models.with_executor_compile_check(&check), + ) + .unwrap(); + let refused = checked + .rejected + .iter() + .find(|r| r.cause == RejectionCause::ExecutorCannotCompile) + .expect("the first plan is refused"); + assert_eq!(refused.reason, "invalid keyed evaluation shape or capacity"); + assert!(refused.cause.failed_a_check()); + assert!(unchecked.costs.contains_key(&refused.id), "{}", refused.id); + assert!(!checked.costs.contains_key(&refused.id)); + assert_ne!(checked.selected, refused.id); + } + /// `ignore_accuracy_guarantees` (issue #774) does not override the /// executor's requirement on Count-Min update weights either: over /// samples of a gauge, P2's CmsWithHeap is still rejected as diff --git a/crates/plan-selection/src/pass/mod.rs b/crates/plan-selection/src/pass/mod.rs index 529a14101..4289343e3 100644 --- a/crates/plan-selection/src/pass/mod.rs +++ b/crates/plan-selection/src/pass/mod.rs @@ -80,6 +80,20 @@ pub struct PlanningModels<'a> { /// whether it keeps query-time results across evaluations (Stage 2); /// default [`DeploymentCapabilities::default`]. pub deployment_capabilities: &'a DeploymentCapabilities, + /// Whether the deployment's executor compiles a candidate's plan; Stage + /// 3 rejects a candidate it does not + /// ([`RejectionCause::ExecutorCannotCompile`](crate::RejectionCause::ExecutorCannotCompile)). + /// `None` (the default) checks nothing; `asap_planner::plan_installable` + /// sets the Planner's executor when the caller leaves it `None`. + pub executor_compile_check: Option<&'a dyn ExecutorCompileCheck>, +} + +/// The deployment executor's compiler, asked by Stage 3 whether it compiles +/// a candidate's plan, so selection and execution cannot drift. +pub trait ExecutorCompileCheck { + /// `Err` with the executor's reason when it cannot compile the plan of + /// `roots`: one candidate's timed query roots, in workload order. + fn compiles(&self, roots: &[asap_types::ir::QueryRoot]) -> Result<(), String>; } impl<'a> PlanningModels<'a> { @@ -94,6 +108,7 @@ impl<'a> PlanningModels<'a> { evidence, plan_selection: &DEFAULT_PLAN_SELECTION, deployment_capabilities: &DEFAULT_DEPLOYMENT, + executor_compile_check: None, } } @@ -104,6 +119,7 @@ impl<'a> PlanningModels<'a> { evidence: &NO_ACCURACY_EVIDENCE, plan_selection: &DEFAULT_PLAN_SELECTION, deployment_capabilities: &DEFAULT_DEPLOYMENT, + executor_compile_check: None, } } @@ -136,6 +152,12 @@ impl<'a> PlanningModels<'a> { self } + /// These models with the deployment executor's compile check. + pub fn with_executor_compile_check(mut self, check: &'a dyn ExecutorCompileCheck) -> Self { + self.executor_compile_check = Some(check); + self + } + /// These models with the deployment's capabilities (supported /// summaries, raw-data and query-time retention). pub fn with_deployment_capabilities( diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index b581b6c49..cba43961e 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -40,7 +40,9 @@ use std::collections::{HashMap, HashSet}; use std::rc::Rc; use asap_logical_asap_optimization::DeploymentCapabilities; -use asap_plan_selection::{NodePriceSource, OperatorKind, SearchReport, Selection}; +use asap_plan_selection::{ + ExecutorCompileCheck, NodePriceSource, OperatorKind, SearchReport, Selection, +}; use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload; use asap_types::ir::physical_export::{ compile_physical_asap_workload_with_node_ids, PhysicalASAPDAG, PhysicalASAPDAGDocument, @@ -359,10 +361,17 @@ pub async fn plan_installable(input: UserInput<'_>) -> Result = entries.into_iter().map(|entry| entry.query.0).collect(); + // Stage 3 selects only a plan the Planner's executor compiles, unless + // the caller brings its own executor's check. + let models = input.models.with_deployment_capabilities(&capabilities); + let models = match models.executor_compile_check { + Some(_) => models, + None => models.with_executor_compile_check(&PlannerExecutorCompileCheck), + }; let input = UserInput { workload: input.workload, frontend_specific: input.frontend_specific, - models: input.models.with_deployment_capabilities(&capabilities), + models, pass: input.pass, }; let output = e2e_plan(input).await?; @@ -389,7 +398,107 @@ pub async fn plan_installable(input: UserInput<'_>) -> Result Result<(), String> { + let installable = || -> Result { + let timed = apply_materialization_timings_to_roots( + roots, + &MaterializationAssignment::default(), + )?; + let expanded = pane_retention_of_expanded_panes(&timed, None)?; + let queries = vec![String::new(); expanded.dag.roots.len()]; + InstallablePlan::of_dag( + expanded.dag, + queries, + None, + &expanded.retention_ms_by_node, + &HashMap::new(), + ) + }; + match installable() { + Ok(plan) => plan.check_compiles(), + // Not installable for another reason, which `plan_installable` + // reports as such (e.g. a scalar root); not the compiler's to say. + Err(_) => Ok(()), + } + } +} + impl InstallablePlan { + /// Whether the Planner's executor compiles this plan as a runtime runs + /// it: each stored output's build over its raw-sample windows and stored + /// inputs, then the query run over the stored outputs it reads and the + /// query-time raw-sample windows. `Err` holds the executor's error. + pub fn check_compiles(&self) -> Result<(), String> { + use asap_executor::physical_planner::{compile, InputContract}; + use std::collections::BTreeMap; + let dag = &self.dag.dag; + let contract = |id: PhysicalASAPNodeId| -> Result<(u64, InputContract), String> { + let node = dag + .nodes + .iter() + .find(|node| node.id == id) + .ok_or_else(|| format!("node {id} is not in the plan"))?; + Ok(( + id as u64, + InputContract::bounded(std::sync::Arc::new(node.output_schema.clone())), + )) + }; + let stored: HashSet = + self.stored_outputs.iter().map(|o| o.node_id).collect(); + for output in &self.stored_outputs { + let inputs = output + .inputs + .raw_sample_windows + .iter() + .map(|window| window.time_range_node_id) + .chain(output.inputs.stored_outputs.iter().copied()) + .map(contract) + .collect::, _>>()?; + compile(dag, inputs, &[output.node_id as u64]) + .map_err(|error| format!("stored output {}: {error}", output.node_id))?; + } + // The query run reads the stored outputs it reaches from the roots + // without passing another, and the query-time windows. + let roots: Vec = self.queries.iter().map(|q| q.root_node_id).collect(); + let mut read = Vec::new(); + let mut pending = roots.clone(); + let mut seen = HashSet::new(); + while let Some(id) = pending.pop() { + if !seen.insert(id) { + continue; + } + if stored.contains(&id) { + read.push(id); + } else { + pending.extend( + dag.edges + .iter() + .filter(|edge| edge.consumer == id) + .map(|edge| edge.producer), + ); + } + } + let inputs = read + .into_iter() + .chain( + self.query_time_raw_windows + .iter() + .map(|window| window.time_range_node_id), + ) + .map(contract) + .collect::, _>>()?; + let roots: Vec = roots.into_iter().map(|id| id as u64).collect(); + compile(dag, inputs, &roots).map_err(|error| format!("query run: {error}"))?; + Ok(()) + } + /// The plan of `dag`, with what the runtime reads from it derived. fn of_dag( dag: PhysicalASAPDAG, diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs index cfc31e5d7..8404f17de 100644 --- a/crates/planner/src/lib.rs +++ b/crates/planner/src/lib.rs @@ -39,15 +39,16 @@ use asap_frontend_sql::{lower_sql_dialect, SqlError}; // configures the same models and reads the same output whether it goes through // `e2e_plan` or straight to `optimize`. pub use asap_plan_selection::pass::{ - optimize, run_stage1_to_stage3, OptimizationInput, OptimizationPass, OptimizeError, - PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, + optimize, run_stage1_to_stage3, ExecutorCompileCheck, OptimizationInput, OptimizationPass, + OptimizeError, PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, + Stage1ToStage3Run, }; mod installable; pub use installable::{ - plan_installable, InstallablePlan, InstallablePlanError, InstalledQuery, PredictedNodeUsage, - PredictedQueryUsage, PredictedUsage, SelectedPlanReport, StoredCoverage, StoredOutput, - StoredOutputBuild, StoredOutputInputs, + plan_installable, InstallablePlan, InstallablePlanError, InstalledQuery, + PlannerExecutorCompileCheck, PredictedNodeUsage, PredictedQueryUsage, PredictedUsage, + SelectedPlanReport, StoredCoverage, StoredOutput, StoredOutputBuild, StoredOutputInputs, }; // What a deployment runtime needs to build a `UserInput` and read an diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 83d974498..da6c1ebe3 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -1808,3 +1808,166 @@ async fn a_ddsketch_over_signed_samples_runs_within_alpha() { }); assert_p99s_within_alpha(&plan, &samples, &per_group, max); } + +/// The ASAPQuery workloads' counter top-k query `topk(10, sum by (job) +/// ((data_total[5m])))` for `f` `increase` or `rate`, every minute +/// within 0.05 at failure probability 0.3, over 24 `data_total` counter +/// series sampled every 10 s. +fn counter_topk_workload(function: &str) -> PlanningWorkload { + let query = format!("topk(10, sum by (job) ({function}(data_total[5m])))"); + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{{ + "query": {query:?}, + "demand": {{"fixed_interval": 60000}}, + "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 100000.0}}}}, + "sources": {{"data_total": {{"series_count": 10000, "metric_type": "counter", + "sample_interval": 100}}}}}}}}"# + )) + .unwrap() +} + +/// The per-second rate of series `instance` (1 or 2) of job `job` (1..=12) +/// in [`counter_samples`]: distinct for every series and every job's sum. +fn counter_rate(job: i64, instance: i64) -> f64 { + job as f64 + 0.1 * instance as f64 +} + +/// `data_total` samples of 12 jobs with 2 series each, every 10 s from +/// 595 s before `EVALUATION_MS`: each a counter growing linearly at +/// [`counter_rate`], so its 5-minute `increase` is 300 × its rate (the +/// samples 5 s inside each window end extrapolate to them exactly). +fn counter_samples() -> Vec { + let mut samples = Vec::new(); + for job in 1..=12 { + for instance in 1..=2 { + for step in 0..60 { + let mut s = sample( + "data_total", + &format!("{job}-{instance}"), + EVALUATION_MS - 595_000 + step * 10_000, + counter_rate(job, instance) * (5 + step * 10) as f64, + ); + s.labels.insert("job".into(), format!("j{job:02}")); + samples.push(s); + } + } + } + samples +} + +/// Issue: with guarantees ignored and no raw samples, the counter top-k +/// queries selected exact per-series panes followed by a query-time +/// CountSketchWithHeap top-k the executor cannot compile ("invalid keyed +/// evaluation shape or capacity"). The selected plan must be one the +/// executor compiles; it runs, equal to its query-time reference and to the +/// exact top 10 jobs (jobs 3..=12, each the sum of its two series). +#[tokio::test] +async fn ignoring_guarantees_counter_topk_plans_compile_and_answer_exactly() { + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let settings = PlanSelectionSettings { + ignore_accuracy_guarantees: true, + ..PlanSelectionSettings::default() + }; + for (function, seconds) in [("increase", 300.0), ("rate", 1.0)] { + let plan = plan_installable(UserInput::new( + &counter_topk_workload(function), + FrontendInput::Promql { histograms: None }, + PlanningModels::builtin() + .with_deployment_capabilities(&capabilities) + .with_plan_selection(&settings), + )) + .await + .unwrap_or_else(|error| panic!("{function}: {error}")); + // An exact plan: no heap over the per-job sums, which have one row + // per job. + assert!( + !plan.dag.dag.nodes.iter().any(|node| matches!( + &node.payload, + asap_executor::planner::ir::physical_export::PhysicalASAPOperatorPayload::ASAP( + ASAPOp::SummaryAgg { + family: asap_executor::planner::ir::schema::FieldDataType::Sketch(..), + .. + } + ) + )), + "{function}" + ); + // The plan compiles as a runtime compiles it. + plan.check_compiles().unwrap(); + let plan_json = serde_json::to_string(&plan).unwrap(); + let run = run_data_plane( + &plan_json, + &counter_samples(), + EVALUATION_MS, + &mut SummaryStore::new(), + ); + assert_eq!( + run.answers_per_query, run.reference_answers_per_query, + "{function}" + ); + let expected: Vec<(String, f64)> = (3..=12) + .map(|job| { + ( + format!("j{job:02}"), + (counter_rate(job, 1) + counter_rate(job, 2)) * seconds, + ) + }) + .collect(); + let mut answered: Vec<(String, f64)> = run.answers_per_query[0] + .iter() + .map(|(labels, value)| (labels["job"].clone(), *value)) + .collect(); + answered.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!(answered.len(), expected.len(), "{function}: {answered:?}"); + for ((job, value), (want_job, want)) in answered.iter().zip(&expected) { + assert_eq!(job, want_job, "{function}"); + assert!( + (value - want).abs() <= 1e-9 * want, + "{function} {job}: {value} vs {want}" + ); + } + } +} + +/// `InstallablePlan::check_compiles` reports what the Planner's executor +/// refuses: the counter top-k plan with its answer column's type changed +/// from Float64 to Utf8 does not compile, with the executor's reason. +#[tokio::test] +async fn check_compiles_reports_the_executors_error() { + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let mut plan = plan_installable(UserInput::new( + &counter_topk_workload("rate"), + FrontendInput::Promql { histograms: None }, + PlanningModels::builtin().with_deployment_capabilities(&capabilities), + )) + .await + .unwrap(); + plan.check_compiles().unwrap(); + let root = plan.queries[0].root_node_id; + let node = plan + .dag + .dag + .nodes + .iter_mut() + .find(|node| node.id == root) + .unwrap(); + let answer = node + .output_schema + .fields + .iter_mut() + .find(|field| field.name == "topk_10") + .unwrap(); + answer.dtype = asap_executor::planner::ir::schema::FieldDataType::Plain( + asap_executor::planner::ir::schema::DataType::Utf8, + ); + let error = plan.check_compiles().unwrap_err(); + assert!(error.starts_with("query run: "), "{error}"); +}