From 0f87cdc9a5cfc88a2d3b6886ad90261e192bd842 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:02 +0000 Subject: [PATCH 1/7] fix(types): state an aggregate column's metric from its real column layout column_metric assumed group keys first and measures last. A per-entity reduction (count_over_time) keeps every label and replaces the sample value in place, so its value column got the absolute metric and a label after it got the count's. Read the layout aggregate_output_schema produces: the per-entity measure at its input column, and a cross-series count_values label before its count. --- crates/types/src/types.rs | 106 +++++++++++++++++++++++++++++++++++--- 1 file changed, 98 insertions(+), 8 deletions(-) diff --git a/crates/types/src/types.rs b/crates/types/src/types.rs index 6ee25a2b..be6c80aa 100644 --- a/crates/types/src/types.rs +++ b/crates/types/src/types.rs @@ -361,16 +361,70 @@ fn column_metric(node: &OperatorNode, column: usize) -> Option { column_metric(right, column - width) } } - NonASAPOp::Aggregate { measures, .. } => { - // Group keys first, then one column per measure. - let keys = node.schema.fields.len().saturating_sub(measures.len()); - match column.checked_sub(keys).and_then(|m| measures.get(m)) { - Some(AggIntent::Quantile { .. } | AggIntent::HistogramQuantile { .. }) => None, - Some(intent) => Some(intent_metric(intent).unwrap_or(AccuracyMetric::ABSOLUTE)), - None => Some(AccuracyMetric::ABSOLUTE), + NonASAPOp::Aggregate { + reduction, + measures, + child, + .. + } => match aggregate_measure_at(node, reduction, measures, child, column) { + Some(AggIntent::Quantile { .. } | AggIntent::HistogramQuantile { .. }) => None, + Some(intent) => Some(intent_metric(intent).unwrap_or(AccuracyMetric::ABSOLUTE)), + None => Some(AccuracyMetric::ABSOLUTE), + }, + _ => Some(AccuracyMetric::ABSOLUTE), + } +} + +/// The measure the aggregate `node`'s output `column` holds, or `None` for +/// a group key, a kept label or `count_values`' synthesized label, as +/// [`aggregate_output_schema`](crate::ir::aggregate_schema::aggregate_output_schema) +/// lays the columns out. A per-entity reduction replaces its one measure's +/// input column in place and keeps every other column; a cross-series one +/// emits its kept labels first, then one column per measure, with a +/// `count_values` label (unless an earlier column has its name) before its +/// count. +fn aggregate_measure_at<'a>( + node: &OperatorNode, + reduction: &crate::ir::operator_properties::Reduction, + measures: &'a [AggIntent], + child: &OperatorNode, + column: usize, +) -> Option<&'a AggIntent> { + use crate::ir::operator_properties::Reduction; + use crate::pre_asap::{resolve_column_ref, ColumnRef}; + let fields = &node.schema.fields; + match reduction { + Reduction::PerEntity => { + let [measure] = measures else { return None }; + let value = match measure.input_cols().first() { + Some(input) => *input, + None => resolve_column_ref(&ColumnRef::SampleValue, &child.schema).ok()?, + }; + (column == value).then_some(measure) + } + Reduction::Reduce(by) => { + let mut next = if by.is_without() { + fields.len().checked_sub(measures.len())? + } else { + by.len() + }; + for measure in measures { + if let AggIntent::CountValues { label } = measure { + if !by.is_without() + && !fields[..next.min(fields.len())] + .iter() + .any(|f| f.name == *label) + { + next += 1; + } + } + if next == column { + return Some(measure); + } + next += 1; } + None } - _ => Some(AccuracyMetric::ABSOLUTE), } } @@ -478,6 +532,42 @@ mod tests { } } + /// A per-entity count (`count_over_time`) keeps every label column and + /// replaces the sample value in place: the value column carries the + /// count's metric, and a label after it is absolute. + #[test] + fn a_per_entity_count_states_its_value_column_not_its_last_column() { + let scan = node(NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + Field::plain("job", DataType::Utf8, false), + ], + 0, + vec![], + ), + }); + let count = node(NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![AggIntent::Count { + accuracy: AccuracyTarget::max_error(0.01), + }], + output_names: vec![], + filters: vec![], + having: None, + child: scan, + }); + assert_eq!(count.schema.fields[1].name, "value"); + assert_eq!( + column_metric(&count, 1), + Some(AccuracyMetric::ABSOLUTE_STREAM_L1) + ); + assert_eq!(column_metric(&count, 2), Some(AccuracyMetric::ABSOLUTE)); + } + /// A root's default metric follows its computation: counts and top-k /// frequencies are L1-scaled absolute errors, distinct counts and L2 /// frequency moments relative ones, anything else (a sum or an entropy From 6ba05435ed8d9600a95c2be0a60bec28f0cbb647 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:02 +0000 Subject: [PATCH 2/7] fix(dag-viewer): name the arithmetic operator and reject a non-array scalar_roots - stageScalarText printed "0" for an Arithmetic operator: ArithmeticOpKind serializes as a string, whose first key is its first index. - validateStagePipeline treated a scalar_roots that is not an array as empty instead of reporting it. - test_server: the skip for a missing stage_pipeline decorates the endpoint test class again, not the post_plan helper. --- tools/dag-viewer/stages.js | 3 ++- tools/dag-viewer/test_server.py | 2 +- tools/dag-viewer/test_viewer.py | 11 +++++++++++ 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/tools/dag-viewer/stages.js b/tools/dag-viewer/stages.js index 32951d18..53cdf59a 100644 --- a/tools/dag-viewer/stages.js +++ b/tools/dag-viewer/stages.js @@ -54,7 +54,7 @@ function stageScalarText(expr) { case 'PromqlScalarFromVector': return `scalar(#${body})`; case 'ScalarSubquery': return `(subquery #${body})`; case 'Negative': return `-${stageScalarText(body.expr)}`; - case 'Arithmetic': return `(${stageScalarText(body.left)} ${Object.keys(body.op)[0] || body.op} ${stageScalarText(body.right)})`; + case 'Arithmetic': return `(${stageScalarText(body.left)} ${typeof body.op === 'string' ? body.op : Object.keys(body.op)[0]} ${stageScalarText(body.right)})`; default: return `${kind}(…)`; } } @@ -129,6 +129,7 @@ function validatePhysicalDag(dag, where, queryCount) { dag.edges.forEach((edge, index) => { if (!edge || !ids.has(edge.producer) || !ids.has(edge.consumer)) errors.push(`${where}: edge #${index} names a missing producer or consumer`); }); + if (dag.scalar_roots !== undefined && !Array.isArray(dag.scalar_roots)) errors.push(`${where}: scalar_roots must be an array`); const scalars = Array.isArray(dag.scalar_roots) ? dag.scalar_roots : []; if (dag.roots.length + scalars.length !== queryCount) { errors.push(`${where}: ${dag.roots.length} roots and ${scalars.length} scalar roots for ${queryCount} workload queries`); diff --git a/tools/dag-viewer/test_server.py b/tools/dag-viewer/test_server.py index 56e5a7b7..fb0fc346 100644 --- a/tools/dag-viewer/test_server.py +++ b/tools/dag-viewer/test_server.py @@ -219,7 +219,6 @@ def test_regenerate_rewrites_every_document(self): self.assertTrue((root / "out" / "a.json").exists()) -@unittest.skipUnless(server.BINARY.exists(), "needs a built stage_pipeline") def post_plan(payload: dict) -> dict: """POST `payload` to /api/plan on a fresh server and return the document.""" httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler) @@ -237,6 +236,7 @@ def post_plan(payload: dict) -> dict: httpd.server_close() +@unittest.skipUnless(server.BINARY.exists(), "needs a built stage_pipeline") class PlanEndpointTests(unittest.TestCase): def test_plan_endpoint_returns_the_stage_pipeline_document(self): document = post_plan({"queries": [{"name": "job_rate", "language": "promql", "text": "sum by (job) (rate(m[1m]))"}]}) diff --git a/tools/dag-viewer/test_viewer.py b/tools/dag-viewer/test_viewer.py index 1dacf982..643905be 100644 --- a/tools/dag-viewer/test_viewer.py +++ b/tools/dag-viewer/test_viewer.py @@ -121,6 +121,7 @@ def test_contract_violations_are_reported(self): roots=physical(d)[0]["dag"]["roots"][:1], scalar_roots=[[2, "EvalTimestamp"]]), "scalar roots out of order": lambda d: physical(d)[0]["dag"].update( roots=[], scalar_roots=[[1, "EvalTimestamp"], [0, "EvalTimestamp"]]), + "scalar roots not an array": lambda d: physical(d)[0]["dag"].update(scalar_roots={"0": "EvalTimestamp"}), "scalar root reads no node": lambda d: physical(d)[0]["dag"].update( roots=physical(d)[0]["dag"]["roots"][:1], scalar_roots=[[1, {"PromqlScalarFromVector": 99}]]), } @@ -131,6 +132,16 @@ def test_contract_violations_are_reported(self): self.assertNotEqual(self.validate(doc), []) +@unittest.skipIf(py_mini_racer is None, "viewer tests require py_mini_racer") +class ScalarTextTests(unittest.TestCase): + """A scalar expression's short text.""" + + def test_arithmetic_names_its_operator(self): + """`ArithmeticOpKind` serializes as a string, e.g. `"Add"`.""" + expr = {"Arithmetic": {"op": "Add", "left": "EvalTimestamp", "right": {"Literal": {"Float64": 2}}, "semantics": "Sql"}} + self.assertEqual(stages_js().call("stageScalarText", expr), "(time() Add 2)") + + @unittest.skipIf(py_mini_racer is None, "viewer tests require py_mini_racer") class CandidateRowTests(unittest.TestCase): """The Stage 3 list: every candidate the document shows, with its From cdea3bdf57912d8a2f8ecf4aac7a4a25fe927608 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:02 +0000 Subject: [PATCH 3/7] refactor(types)!: name materialization timing errors and helpers by timing - ExecutionDataStateError::EvaluationUnderMaintenance is now QueryTimeValueUnderIngestionSummary and InvalidMaintenanceBinary is InvalidIngestionTimeBinary (now documented); their messages say ingestion time. - split_shared_by_phase is now split_shared_by_timing: it splits by execution timing, not evaluation phase. - MaterializationAssignment's Debug prints each assigned node's kind and address, not its whole sub-DAG. - The edge-rule table states that a query-time MaintainPopulation does not read a MatAtIngestion input's stored output. - Tests: an ingestion-fixed node rejects MatAtQuery, an explicit NotMaterialized keeps a fixed timing, and a fixed-timing conflict survives the split. --- .../src/stage2_physical_asap_optimization.rs | 21 +-- crates/types/src/ir/mod.rs | 2 +- crates/types/src/ir/physical_export.rs | 4 +- crates/types/src/ir/timing.rs | 168 +++++++++++++++--- .../src/post_asap/execution_data_state.rs | 19 +- 5 files changed, 168 insertions(+), 46 deletions(-) diff --git a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs b/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs index edd463b5..f76dffa6 100644 --- a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs +++ b/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs @@ -29,7 +29,7 @@ //! other combination of options is one physical candidate, the //! all-`NotMaterialized` one first: nothing is capped. A node that one //! query reads at ingestion time and another at query time is copied, one -//! copy per phase ([`split_shared_by_phase`]). +//! copy per timing ([`split_shared_by_timing`]). //! //! A build may run at ingestion time (S4) only when: //! @@ -76,7 +76,7 @@ use asap_types::ir::physical_export::{ compile_physical_asap_workload_with_node_ids, PhysicalASAPDAG, PhysicalASAPNodeId, }; use asap_types::ir::{ - apply_materialization_timings, is_filtered_build, split_shared_by_phase, ASAPOp, + apply_materialization_timings, is_filtered_build, split_shared_by_timing, ASAPOp, MaterializationAssignment, Medium, NonASAPOp, Operator, OperatorNode, ProjectItem, QueryRoot, ScalarExpr, SchemaDerivationError, SortKey, TimingMemo, }; @@ -301,7 +301,7 @@ pub fn optimize_physical_asap( .iter() .flat_map(|root| root_operators(root).into_iter().cloned()) .collect(); - let (split, options) = split_shared_by_phase(&operators, &options); + let (split, options) = split_shared_by_timing(&operators, &options); let replaced: HashMap<_, _> = operators .iter() .map(Rc::as_ptr) @@ -1047,11 +1047,12 @@ mod tests { /// `root` with its summary build in 1-min tumbling panes. fn in_panes(root: QueryRoot) -> QueryRoot { let roots = [root]; - let sites = crate::asap_aware_cse::pane_candidates(&roots, &[Some(MINUTE)]); - assert_eq!(sites.len(), 1, "{sites:?}"); - let mut variants = - crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &sites).unwrap(); - variants.pop().unwrap().roots.unwrap().remove(0) + let rule_candidates = crate::asap_aware_cse::pane_candidates(&roots, &[Some(MINUTE)]); + assert_eq!(rule_candidates.len(), 1, "{rule_candidates:?}"); + let mut pass2_candidates = + crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &rule_candidates) + .unwrap(); + pass2_candidates.pop().unwrap().roots.unwrap().remove(0) } const AT_INGESTION: Materialization = Materialization::MatAtIngestion(Medium::Memory); @@ -1399,9 +1400,9 @@ mod tests { /// Two queries share one scan: one reads it through panes maintained at /// ingestion time, the other at query time. The scan is split, one copy - /// per phase; with nothing materialized it stays one node. + /// per timing; with nothing materialized it stays one node. #[test] - fn a_scan_two_queries_read_at_two_phases_is_split() { + fn a_scan_two_queries_read_at_two_timings_is_split() { let QueryRoot::Operator(panes) = in_panes(summary_root("quantile_over_time(0.99, m[5m])")) else { unreachable!() diff --git a/crates/types/src/ir/mod.rs b/crates/types/src/ir/mod.rs index e6d91e98..262aaf90 100644 --- a/crates/types/src/ir/mod.rs +++ b/crates/types/src/ir/mod.rs @@ -25,7 +25,7 @@ pub mod timing; pub use crate::post_asap::execution_data_state::{Materialization, Medium}; pub use timing::{ apply_materialization_timings, data_state, is_filtered_build, planned_data_state, - split_shared_by_phase, validate_at_ingestion, MaterializationAssignment, TimingMemo, + split_shared_by_timing, validate_at_ingestion, MaterializationAssignment, TimingMemo, }; pub mod schema_support; /// Semantic observation coverage, separate from field layout and physical timing. diff --git a/crates/types/src/ir/physical_export.rs b/crates/types/src/ir/physical_export.rs index a9d3f2d4..562399f7 100644 --- a/crates/types/src/ir/physical_export.rs +++ b/crates/types/src/ir/physical_export.rs @@ -453,7 +453,7 @@ pub fn compile_physical_asap_workload_with_node_ids( // Children come before their parents, so every producer is already in `nodes`. for (producer, role) in edge_roles(&flat_node.operator) { let producer_state = nodes[producer].output_state; - let maintenance_dependency = producer_state.timing == ExecutionTiming::IngestionTime + let ingestion_dependency = producer_state.timing == ExecutionTiming::IngestionTime && output_state.timing == ExecutionTiming::IngestionTime; edges.push(PhysicalASAPDAGEdge { producer, @@ -462,7 +462,7 @@ pub fn compile_physical_asap_workload_with_node_ids( intermediate_schema: nodes[producer].output_schema.clone(), data_state: producer_state, grouping: grouping_compatibility(&nodes[producer].payload, &flat_node.operator), - window: if maintenance_dependency { + window: if ingestion_dependency { WindowEdgeCompatibility::RequiresAlignedPanePhaseOrExactWindowEdgeResidual } else { WindowEdgeCompatibility::NotApplicable diff --git a/crates/types/src/ir/timing.rs b/crates/types/src/ir/timing.rs index c496df14..07115b11 100644 --- a/crates/types/src/ir/timing.rs +++ b/crates/types/src/ir/timing.rs @@ -36,7 +36,7 @@ //! error ([`ExecutionDataStateError::UnstoredIngestionRoot`]). //! //! A node reached from two consumers that need different timings cannot be -//! executed once for both; [`split_shared_by_phase`] copies such a sub-DAG +//! executed once for both; [`split_shared_by_timing`] copies such a sub-DAG //! for one side before the assignment is applied, and the pass itself //! rejects a conflict it still finds. //! @@ -45,11 +45,11 @@ //! | Consumer | Accepts from an input | //! |---|---| //! | `SummaryAgg.child`, `PaneChain.child` | Rows, or exact-accumulator state, never a query-time value when the state is maintained | -//! | `SummaryEstimate.summary_input` | Summary state at either phase | -//! | `SummaryMerge.children` | Summary state at either phase; at ingestion time only for a merge at ingestion time | +//! | `SummaryEstimate.summary_input` | Summary state at either timing | +//! | `SummaryMerge.children` | Summary state at either timing; at ingestion time only for a merge at ingestion time | //! | `FinalizeExactAccumulator.child` | Exact-accumulator state | //! | `EvaluatePopulation.child` | A timed `MaintainPopulation` | -//! | `MaintainPopulation.child` | Rows at its own timing matching the population's input | +//! | `MaintainPopulation.child` | Rows at its own timing matching the population's input; unlike a `NonASAP` consumer, a query-time population does not read a `MatAtIngestion` input's stored output | //! | any `NonASAP` consumer | Rows (or exact-accumulator state for a projection-like operator) at the consumer's own timing, or a `MatAtIngestion` output read at query time; ingestion work never reads a query-time value | use std::collections::HashMap; @@ -75,12 +75,33 @@ use crate::pre_asap::schema::{DataType, FieldDataType, Schema}; /// /// The assignment keeps every assigned node alive, so a later allocation /// cannot reuse its address and match a stale entry. -#[derive(Debug, Clone, Default)] +#[derive(Clone, Default)] pub struct MaterializationAssignment { assigned: HashMap<*const OperatorNode, (Rc, Materialization)>, unassigned: Materialization, } +/// Each assigned node as its operator kind and address, not its whole +/// sub-DAG. +impl std::fmt::Debug for MaterializationAssignment { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let assigned: Vec<_> = self + .assigned + .iter() + .map(|(ptr, (node, materialization))| { + ( + format!("{}@{ptr:p}", node.operator.kind_name()), + materialization, + ) + }) + .collect(); + f.debug_struct("MaterializationAssignment") + .field("assigned", &assigned) + .field("unassigned", &self.unassigned) + .finish() + } +} + impl MaterializationAssignment { /// Every `SummaryAgg` or `MaintainPopulation` not [`set`](Self::set) /// explicitly takes `materialization` where the DAG allows it: a @@ -391,7 +412,7 @@ fn placement( } /// The timing `node` takes when its consumer runs at `consumer`, for -/// [`split_shared_by_phase`]: an assignment that cannot be honored is +/// [`split_shared_by_timing`]: an assignment that cannot be honored is /// reported later by [`apply_materialization_timings`]. fn own_timing( node: &Rc, @@ -502,19 +523,23 @@ fn validate_asap( is_exact_accumulator_state(&child.schema)? } other => { - return Err(ExecutionDataStateError::EvaluationUnderMaintenance { - edge: "SummaryAgg.child", - child: other, - }) + return Err( + ExecutionDataStateError::QueryTimeValueUnderIngestionSummary { + edge: "SummaryAgg.child", + child: other, + }, + ) } } if timing == ExecutionTiming::IngestionTime && avail.timing == ExecutionTiming::QueryTime { - return Err(ExecutionDataStateError::EvaluationUnderMaintenance { - edge: "SummaryAgg.child", - child: avail, - }); + return Err( + ExecutionDataStateError::QueryTimeValueUnderIngestionSummary { + edge: "SummaryAgg.child", + child: avail, + }, + ); } Ok(()) } @@ -737,7 +762,7 @@ fn validate_non_asap( || !plain_float_or_ts(&node.schema) || float_count != 1 { - return Err(ExecutionDataStateError::InvalidMaintenanceBinary); + return Err(ExecutionDataStateError::InvalidIngestionTimeBinary); } } } @@ -759,7 +784,7 @@ fn validate_non_asap( /// `Rc`, shared by the roots that read it. Returns the (possibly rewritten) /// roots, in order, and `assignment` remapped onto them: a rebuilt node /// keeps its original's explicit materialization. -pub fn split_shared_by_phase( +pub fn split_shared_by_timing( roots: &[Rc], assignment: &MaterializationAssignment, ) -> (Vec>, MaterializationAssignment) { @@ -855,7 +880,7 @@ pub fn split_shared_by_phase( (split, remapped) } -/// Maintenance arithmetic needs the same per-series population on both sides. +/// Ingestion-time arithmetic needs the same per-series population on both sides. fn per_series_rows(node: &OperatorNode) -> Option<&OperatorNode> { use crate::post_asap::ExactKind; match &node.operator { @@ -1161,8 +1186,8 @@ mod tests { ); } - /// One shared sub-DAG reached as maintenance input and as query-time - /// input cannot be executed once for both; splitting it by phase first + /// One shared sub-DAG reached as ingestion-time input and as query-time + /// input cannot be executed once for both; splitting it by timing first /// makes the plan timeable. #[test] fn a_shared_subtree_reached_at_two_timings_conflicts() { @@ -1183,7 +1208,7 @@ mod tests { }) ); let (split, assignment) = - split_shared_by_phase(std::slice::from_ref(&root), &all_at_ingestion()); + split_shared_by_timing(std::slice::from_ref(&root), &all_at_ingestion()); assert!( apply_materialization_timings(&split[0], &assignment, &mut TimingMemo::new()).is_ok() ); @@ -1228,7 +1253,7 @@ mod tests { .unwrap(); let mut assignment = MaterializationAssignment::default(); assignment.set(&summary, AT_INGESTION); - let (split, assignment) = split_shared_by_phase(std::slice::from_ref(&root), &assignment); + let (split, assignment) = split_shared_by_timing(std::slice::from_ref(&root), &assignment); let timed = apply_materialization_timings(&split[0], &assignment, &mut TimingMemo::new()).unwrap(); let summary = child(&child(&child(&timed))); @@ -1333,7 +1358,7 @@ mod tests { /// over both roots succeeds; a node both queries reach at one timing /// stays one node. #[test] - fn split_across_roots_copies_only_what_two_queries_read_at_two_phases() { + fn split_across_roots_copies_only_what_two_queries_read_at_two_timings() { let shared = scan(); let summary = agg(Rc::clone(&shared), kll()); let by_summary = max(estimate(Rc::clone(&summary))); @@ -1356,7 +1381,7 @@ mod tests { timed_unsplit, Err(ExecutionDataStateError::ConflictingTiming { .. }) )); - let (split, assignment) = split_shared_by_phase(&roots, &assignment); + let (split, assignment) = split_shared_by_timing(&roots, &assignment); assert_eq!(split.len(), 2); let mut memo = TimingMemo::new(); let timed: Vec<_> = split @@ -1434,6 +1459,97 @@ mod tests { ); } + /// The Debug output names each assigned node by its operator kind, not + /// its whole sub-DAG. + #[test] + fn debug_names_each_assigned_node_by_its_kind() { + let mut assignment = MaterializationAssignment::default(); + assignment.set(&agg(scan(), kll()), AT_INGESTION); + let text = format!("{assignment:?}"); + assert!(text.contains("SummaryAgg@"), "{text}"); + assert!(!text.contains("fields"), "{text}"); + } + + /// A node fixed at ingestion time by its candidate rejects a `MatAtQuery` + /// assignment, which needs the other timing. + #[test] + fn ingestion_fixed_node_rejects_mat_at_query() { + let fixed = placed_at_ingestion(max(scan())); + let root = max(Rc::clone(&fixed)); + let at_query = Materialization::MatAtQuery(Medium::Memory); + let mut assignment = MaterializationAssignment::default(); + assignment.set(&fixed, at_query); + assert_eq!( + apply_materialization_timings(&root, &assignment, &mut TimingMemo::new()).err(), + Some( + ExecutionDataStateError::MaterializationConflictsWithFixedTiming { + operator: "Aggregate", + fixed: ExecutionTiming::IngestionTime, + materialization: at_query, + } + ) + ); + } + + /// An explicit `NotMaterialized` on a node whose timing its candidate + /// fixed agrees with either timing: the node keeps its fixed timing and + /// records `NotMaterialized`. + #[test] + fn explicit_not_materialized_keeps_a_fixed_timing() { + let read = Rc::new( + (*finalize(agg(scan(), exact_sum()))) + .clone() + .with_timing(Some(ExecutionTiming::QueryTime)), + ); + let root = max(Rc::clone(&read)); + let mut assignment = MaterializationAssignment::default(); + assignment.set(&read, Materialization::NotMaterialized); + let timed = + apply_materialization_timings(&root, &assignment, &mut TimingMemo::new()).unwrap(); + let read = child(&timed); + assert_eq!(read.timing, Some(ExecutionTiming::QueryTime)); + assert_eq!(read.materialization, Materialization::NotMaterialized); + } + + /// Splitting a workload that also shares a scan at two timings keeps a + /// conflicting assignment on a fixed-timing node: the split roots still + /// fail with the conflict instead of dropping the assignment. + #[test] + fn a_fixed_timing_conflict_survives_the_split() { + let shared = scan(); + let summary = agg(Rc::clone(&shared), kll()); + let pair = concat(vec![ + max(estimate(Rc::clone(&summary))), + max(Rc::clone(&shared)), + ]); + let state = agg(Rc::clone(&shared), exact_sum()); + let read = Rc::new( + (*finalize(Rc::clone(&state))) + .clone() + .with_timing(Some(ExecutionTiming::QueryTime)), + ); + let mut assignment = MaterializationAssignment::default(); + assignment.set(&summary, AT_INGESTION); + assignment.set(&state, AT_INGESTION); + assignment.set(&read, AT_INGESTION); + let (split, assignment) = split_shared_by_timing(&[pair, read], &assignment); + let mut memo = TimingMemo::new(); + let timed: Result, _> = split + .iter() + .map(|root| apply_materialization_timings(root, &assignment, &mut memo)) + .collect(); + assert_eq!( + timed.err(), + Some( + ExecutionDataStateError::MaterializationConflictsWithFixedTiming { + operator: "FinalizeExactAccumulator", + fixed: ExecutionTiming::QueryTime, + materialization: AT_INGESTION, + } + ) + ); + } + /// A root stored at ingestion time is valid: the query reads the stored /// output. A root that runs at ingestion time without being stored there /// is not. @@ -1496,7 +1612,7 @@ mod tests { // Typed derived updates retain one opaque series identity only when both // operands cover the same series; arbitrary labels are never admitted. #[test] - fn maintenance_binary_accepts_only_well_typed_series_identity() { + fn ingestion_time_binary_accepts_only_well_typed_series_identity() { use crate::ir::non_asap::BinaryOperator; use crate::pre_asap::{ArithmeticOpKind, BinaryOpKind}; let identity = crate::pre_asap::schema::PROMQL_SERIES_IDENTITY; @@ -1571,7 +1687,7 @@ mod tests { assert!(validate(schema.clone(), "m").is_ok()); assert_eq!( validate(schema.clone(), "n"), - Err(ExecutionDataStateError::InvalidMaintenanceBinary), + Err(ExecutionDataStateError::InvalidIngestionTimeBinary), "different selectors may cover different series" ); for mutation in 0..4 { @@ -1584,7 +1700,7 @@ mod tests { } assert_eq!( validate(invalid, "m"), - Err(ExecutionDataStateError::InvalidMaintenanceBinary) + Err(ExecutionDataStateError::InvalidIngestionTimeBinary) ); } } diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 03c98225..2f68af73 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -129,13 +129,14 @@ pub enum ExecutionDataStateError { #[error("invalid maintained-population maintenance/evaluation contract")] InvalidMaintainedPopulation, /// A query-time value (a `SummaryEstimate` or query-time operator output) - /// placed beneath a maintained summary — the one shape issue #171's - /// data_state split exists to make unrepresentable. + /// placed beneath a summary built at ingestion time — the one shape issue + /// #171's data_state split exists to make unrepresentable. #[error( - "evaluation value under maintenance: {edge} received a {child} input, but a maintained \ - summary can only consume update-path values (or exact accumulator state)" + "query-time value under an ingestion-time summary: {edge} received a {child} input, but a \ + summary built at ingestion time can only consume ingestion-time rows (or exact \ + accumulator state)" )] - EvaluationUnderMaintenance { + QueryTimeValueUnderIngestionSummary { edge: &'static str, child: ExecutionDataState, }, @@ -190,8 +191,12 @@ pub enum ExecutionDataStateError { /// apply it. #[error("{operator} has a filter, which only a query-time build can apply")] FilteredSummaryAtIngestion { operator: &'static str }, - #[error("unsupported maintenance binary schema or operator")] - InvalidMaintenanceBinary, + /// An ingestion-time `BinaryOp` the update path cannot evaluate: not + /// arithmetic without vector matching, operands that are not the same + /// per-series rows, or a schema other than non-null timestamps, at most + /// one series identity and one Float64 value. + #[error("unsupported ingestion-time binary schema or operator")] + InvalidIngestionTimeBinary, #[error("checked division requires one valid guard on a read-time division operator")] InvalidCheckedDivision, /// An exact operator whose input columns are not all `Plain` at its From 98e9cef3abf20190d1f1a6da3a9ba2b5ed8e2682 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:03 +0000 Subject: [PATCH 4/7] refactor: name PATTERN_A's fields, drop a pass-through helper, compute the front's latencies once - devtools PATTERN_A is an array of PatternAQuery { promql, window_ms, offset_before_t_ms } instead of unnamed tuples. - candidate_labels: same() only forwarded to same_computation. - Stage 3's Pareto front computes each candidate's worst latency once instead of once per pairwise comparison (dominates_at). - Pass 2 test locals named for the old records (groups, variants, sites) are renamed to the *Candidate vocabulary. --- .../asap-aware-mapping/src/asap_aware_cse.rs | 195 +++++++++--------- .../src/stage3_plan_selection.rs | 27 ++- crates/devtools/src/candidate_labels.rs | 33 +-- .../devtools/src/planner_layering_examples.rs | 66 +++--- .../tests/planner_layering_example3.rs | 2 +- .../tests/planner_layering_example4.rs | 4 +- 6 files changed, 186 insertions(+), 141 deletions(-) diff --git a/crates/asap-aware-mapping/src/asap_aware_cse.rs b/crates/asap-aware-mapping/src/asap_aware_cse.rs index 5caee47f..4cb67d5a 100644 --- a/crates/asap-aware-mapping/src/asap_aware_cse.rs +++ b/crates/asap-aware-mapping/src/asap_aware_cse.rs @@ -1544,8 +1544,8 @@ mod tests { .collect() } - fn kinds(group: &SharingCandidate) -> &'static str { - group.nodes[0].operator.kind_name() + fn kinds(sharing_candidate: &SharingCandidate) -> &'static str { + sharing_candidate.nodes[0].operator.kind_name() } /// The positions in `all` (by their nodes) of the sharing candidates @@ -1580,29 +1580,30 @@ mod tests { "sum by (job) (rate(m[1m]))", "topk by (job) (10, sum_over_time(m[1m]))", ]); - let groups = sharing_candidates(&roots); + let sharing = sharing_candidates(&roots); assert_eq!( - groups.iter().map(kinds).collect::>(), + sharing.iter().map(kinds).collect::>(), ["Scan", "TimeRange"] ); - let variants = identical_expression_pass2_candidates(&roots, &sharing_candidates(&roots)); + let pass2_candidates = + identical_expression_pass2_candidates(&roots, &sharing_candidates(&roots)); assert_eq!( - variants + pass2_candidates .iter() - .map(|v| shared_positions(&groups, v)) + .map(|v| shared_positions(&sharing, v)) .collect::>(), [vec![0], vec![0, 1]] ); - for variant in &variants { - assert_eq!(variant.roots, roots, "values are unchanged"); - assert!(distinct_nodes(&variant.roots) < distinct_nodes(&roots)); + for pass2_candidate in &pass2_candidates { + assert_eq!(pass2_candidate.roots, roots, "values are unchanged"); + assert!(distinct_nodes(&pass2_candidate.roots) < distinct_nodes(&roots)); } assert_eq!( - common_nodes(&variants[0].roots[0], &variants[0].roots[1]).len(), + common_nodes(&pass2_candidates[0].roots[0], &pass2_candidates[0].roots[1]).len(), 1 ); assert_eq!( - common_nodes(&variants[1].roots[0], &variants[1].roots[1]).len(), + common_nodes(&pass2_candidates[1].roots[0], &pass2_candidates[1].roots[1]).len(), 2 ); } @@ -1621,8 +1622,9 @@ mod tests { .collect::>(), ["Scan", "TimeRange", "Scan", "TimeRange"] ); - let variants = identical_expression_pass2_candidates(&roots, &sharing_candidates(&roots)); - assert_eq!(variants.len(), 8); + let pass2_candidates = + identical_expression_pass2_candidates(&roots, &sharing_candidates(&roots)); + assert_eq!(pass2_candidates.len(), 8); let shares = |v: &IdenticalExpressionPass2Candidate| { ( !common_nodes(&v.roots[0], &v.roots[1]).is_empty(), @@ -1630,10 +1632,10 @@ mod tests { ) }; assert_eq!( - variants.iter().map(shares).collect::>(), + pass2_candidates.iter().map(shares).collect::>(), HashSet::from([(true, false), (false, true), (true, true)]) ); - assert!(variants.iter().all(|v| v.roots == roots)); + assert!(pass2_candidates.iter().all(|v| v.roots == roots)); } /// Queries with nothing in common have no Pass 2 candidate. @@ -1649,13 +1651,13 @@ mod tests { /// Every non-empty subset of the sharing candidates, applied, kept when it gives a /// distinct candidate: the definition the direct enumeration follows. fn brute_force(roots: &[QueryRoot]) -> Vec> { - let groups = sharing_candidates(roots); - let mut subset = vec![false; groups.len()]; + let sharing = sharing_candidates(roots); + let mut subset = vec![false; sharing.len()]; let mut out = Vec::new(); while next_subset(&mut subset) { - let (_, merged) = apply(roots, &groups, &subset); + let (_, merged) = apply(roots, &sharing, &subset); if merged == subset { - out.push((0..groups.len()).filter(|&g| subset[g]).collect()); + out.push((0..sharing.len()).filter(|&g| subset[g]).collect()); } } out @@ -1675,10 +1677,10 @@ mod tests { &["sum(abs(ceil(a)))", "max(abs(ceil(a)))", "min(ceil(a))"], ] { let roots = roots(queries); - let groups = sharing_candidates(&roots); - let direct: Vec<_> = identical_expression_pass2_candidates(&roots, &groups) + let sharing = sharing_candidates(&roots); + let direct: Vec<_> = identical_expression_pass2_candidates(&roots, &sharing) .iter() - .map(|v| shared_positions(&groups, v)) + .map(|v| shared_positions(&sharing, v)) .collect(); assert_eq!(direct, brute_force(&roots), "{queries:?}"); } @@ -1708,11 +1710,11 @@ mod tests { QueryRoot::Operator(node) }; let roots = [chain(), chain()]; - let groups = sharing_candidates(&roots); - assert_eq!(groups.len(), 20); + let sharing = sharing_candidates(&roots); + assert_eq!(sharing.len(), 20); let start = std::time::Instant::now(); - let variants = identical_expression_pass2_candidates(&roots, &groups); - assert_eq!(variants.len(), 20); + let pass2_candidates = identical_expression_pass2_candidates(&roots, &sharing); + assert_eq!(pass2_candidates.len(), 20); assert!( start.elapsed() < std::time::Duration::from_secs(5), "{:?}", @@ -1869,21 +1871,23 @@ mod tests { let own = kll_builds(&roots); assert_eq!(own.len(), 2); assert!(own[0].1 < own[1].1); - let groups = capability_sharing_candidates(&roots); - assert_eq!(groups.len(), 1); - assert_eq!(groups[0].family, SketchAlgorithm::Kll); - let variants = summary_capability_pass2_candidates(&roots, &groups).unwrap(); - assert_eq!(variants.len(), 1); - let [shared_builds] = variants[0].shared_builds.as_slice() else { - panic!("one shared build: {:?}", variants[0].shared_builds) + let sharing = capability_sharing_candidates(&roots); + assert_eq!(sharing.len(), 1); + assert_eq!(sharing[0].family, SketchAlgorithm::Kll); + let pass2_candidates = summary_capability_pass2_candidates(&roots, &sharing).unwrap(); + assert_eq!(pass2_candidates.len(), 1); + let [shared_builds] = pass2_candidates[0].shared_builds.as_slice() else { + panic!("one shared build: {:?}", pass2_candidates[0].shared_builds) }; - assert_eq!(positions(&groups[0].members, shared_builds, |m| m), [0, 1]); - let shared = kll_builds(variants[0].roots.as_ref().unwrap()); + assert_eq!(positions(&sharing[0].members, shared_builds, |m| m), [0, 1]); + let shared = kll_builds(pass2_candidates[0].roots.as_ref().unwrap()); assert_eq!(shared.len(), 1, "one KLL serves both"); assert_eq!(shared[0].1, own[1].1); - assert!(root_bound(&variants[0].roots.as_ref().unwrap()[0]) < root_bound(&roots[0])); + assert!( + root_bound(&pass2_candidates[0].roots.as_ref().unwrap()[0]) < root_bound(&roots[0]) + ); assert_eq!( - root_bound(&variants[0].roots.as_ref().unwrap()[1]), + root_bound(&pass2_candidates[0].roots.as_ref().unwrap()[1]), root_bound(&roots[1]) ); } @@ -1901,18 +1905,18 @@ mod tests { ], SketchAlgorithm::Kll, ); - let groups = capability_sharing_candidates(&roots); - let variants = summary_capability_pass2_candidates(&roots, &groups).unwrap(); + let sharing = capability_sharing_candidates(&roots); + let pass2_candidates = summary_capability_pass2_candidates(&roots, &sharing).unwrap(); assert_eq!( - variants + pass2_candidates .iter() - .map(|v| positions(&groups[0].members, &v.shared_builds[0], |m| m)) + .map(|v| positions(&sharing[0].members, &v.shared_builds[0], |m| m)) .collect::>(), [vec![0, 1], vec![0, 2], vec![1, 2], vec![0, 1, 2]] ); - for variant in &variants { - let builds = kll_builds(variant.roots.as_ref().unwrap()); - assert_eq!(builds.len(), 4 - variant.shared_builds[0].len()); + for pass2_candidate in &pass2_candidates { + let builds = kll_builds(pass2_candidate.roots.as_ref().unwrap()); + assert_eq!(builds.len(), 4 - pass2_candidate.shared_builds[0].len()); } } @@ -1927,16 +1931,16 @@ mod tests { ], SketchAlgorithm::Kll, ); - let mut groups = capability_sharing_candidates(&roots); - let members = groups[0].members.clone(); - groups[0].members = members + let mut sharing = capability_sharing_candidates(&roots); + let members = sharing[0].members.clone(); + sharing[0].members = members .iter() .cycle() .take(MAX_CAPABILITY_SHARING_MEMBERS + 1) .cloned() .collect(); assert!(matches!( - summary_capability_pass2_candidates(&roots, &groups), + summary_capability_pass2_candidates(&roots, &sharing), Err(PerComputationCandidateError::Unsupported(_)) )); } @@ -1993,16 +1997,16 @@ mod tests { &["sum(abs(a))", "max(abs(a))", "min(a)", "sum(b)", "max(b)"], ] { let roots = roots(queries); - let groups = sharing_candidates(&roots); - for variant in identical_expression_pass2_candidates(&roots, &groups) { - let shared = shared_positions(&groups, &variant); - let label: Vec = (0..groups.len()).map(|g| shared.contains(&g)).collect(); - let (applied, merged) = apply(&roots, &groups, &label); + let sharing = sharing_candidates(&roots); + for pass2_candidate in identical_expression_pass2_candidates(&roots, &sharing) { + let shared = shared_positions(&sharing, &pass2_candidate); + let label: Vec = (0..sharing.len()).map(|g| shared.contains(&g)).collect(); + let (applied, merged) = apply(&roots, &sharing, &label); assert_eq!(merged, label, "{queries:?}"); - assert_eq!(applied, variant.roots, "{queries:?}"); + assert_eq!(applied, pass2_candidate.roots, "{queries:?}"); assert_eq!( distinct_nodes(&applied), - distinct_nodes(&variant.roots), + distinct_nodes(&pass2_candidate.roots), "{queries:?} {:?}", shared ); @@ -2010,7 +2014,8 @@ mod tests { if b < roots.len() { assert_eq!( common_nodes(&applied[a], &applied[b]).len(), - common_nodes(&variant.roots[a], &variant.roots[b]).len() + common_nodes(&pass2_candidate.roots[a], &pass2_candidate.roots[b]) + .len() ); } } @@ -2022,7 +2027,7 @@ mod tests { fn panes(roots: &[QueryRoot], interval_ms: &[Option]) -> Vec<(u64, u32)> { pane_candidates(roots, interval_ms) .iter() - .map(|site| (site.pane_width_ms, site.pane_count)) + .map(|rule_candidate| (rule_candidate.pane_width_ms, rule_candidate.pane_count)) .collect() } @@ -2072,16 +2077,18 @@ mod tests { &[("quantile_over_time(0.99, m[5m])", 0.01)], SketchAlgorithm::Kll, ); - let sites = pane_candidates(&roots, &[Some(60_000)]); - let variants = window_composition_pass2_candidates(&roots, &sites).unwrap(); - let [variant] = variants.as_slice() else { - panic!("one variant: {variants:?}") + let rule_candidates = pane_candidates(&roots, &[Some(60_000)]); + let pass2_candidates = + window_composition_pass2_candidates(&roots, &rule_candidates).unwrap(); + let [pass2_candidate] = pass2_candidates.as_slice() else { + panic!("one Pass 2 candidate: {pass2_candidates:?}") }; assert_eq!( - positions(&sites, &variant.pane_candidates, |c| &c.build), + positions(&rule_candidates, &pass2_candidate.pane_candidates, |c| &c + .build), [0] ); - let panes = variant.roots.as_ref().unwrap(); + let panes = pass2_candidate.roots.as_ref().unwrap(); let QueryRoot::Operator(root) = &panes[0] else { panic!("operator root") }; @@ -2115,14 +2122,14 @@ mod tests { child: range, family: whole_family, .. - }) = sites[0].build.asap() + }) = rule_candidates[0].build.asap() else { unreachable!() }; assert_eq!(family, whole_family); assert_eq!(child.operator, range.children()[0].operator); - assert_eq!(chain.schema, sites[0].build.schema); - assert_eq!(merge.coverage(), sites[0].build.coverage()); + assert_eq!(chain.schema, rule_candidates[0].build.schema); + assert_eq!(merge.coverage(), rule_candidates[0].build.coverage()); } /// Two builds that could be split give three candidates: each alone and @@ -2136,14 +2143,14 @@ mod tests { ], SketchAlgorithm::Kll, ); - let sites = pane_candidates(&roots, &[Some(60_000), Some(60_000)]); - assert_eq!(sites.len(), 2); - let subsets: Vec<_> = window_composition_pass2_candidates(&roots, &sites) + let rule_candidates = pane_candidates(&roots, &[Some(60_000), Some(60_000)]); + assert_eq!(rule_candidates.len(), 2); + let subsets: Vec<_> = window_composition_pass2_candidates(&roots, &rule_candidates) .unwrap() .iter() .map(|v| { assert!(v.roots.is_ok()); - positions(&sites, &v.pane_candidates, |c| &c.build) + positions(&rule_candidates, &v.pane_candidates, |c| &c.build) }) .collect(); assert_eq!(subsets, [vec![0], vec![1], vec![0, 1]]); @@ -2177,16 +2184,16 @@ mod tests { #[test] fn rollup_from_a_finer_exact_sum() { let roots = exact(&["sum by (job) (m)", "sum by (job, instance) (m)"]); - let sites = rollup_candidates(&roots); - let [site] = sites.as_slice() else { - panic!("one site: {sites:?}") + let rule_candidates = rollup_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one candidate: {rule_candidates:?}") }; - assert!(Rc::ptr_eq(&site.coarse, &builds(&roots[0])[0])); - let variants = rollup_pass2_candidates(&roots, &sites).unwrap(); - let [variant] = variants.as_slice() else { - panic!("one variant") + assert!(Rc::ptr_eq(&rule_candidate.coarse, &builds(&roots[0])[0])); + let pass2_candidates = rollup_pass2_candidates(&roots, &rule_candidates).unwrap(); + let [pass2_candidate] = pass2_candidates.as_slice() else { + panic!("one Pass 2 candidate") }; - let rolled = variant.roots.as_ref().unwrap(); + let rolled = pass2_candidate.roots.as_ref().unwrap(); let coarse = builds(&rolled[0]); let fine = builds(&rolled[1]); assert_eq!(fine.len(), 1); @@ -2231,11 +2238,11 @@ mod tests { ] { let finer = query.replace("(job)", "(job, instance)"); let roots = exact(&[query, &finer]); - let sites = rollup_candidates(&roots); - let [site] = sites.as_slice() else { - panic!("{query}: one site: {sites:?}") + let rule_candidates = rollup_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("{query}: one candidate: {rule_candidates:?}") }; - let mut value = Rc::clone(&site.rollup); + let mut value = Rc::clone(&rule_candidate.rollup); if let Some(NonASAPOp::Project { child, .. }) = value.non_asap() { value = Rc::clone(child); } @@ -2250,9 +2257,11 @@ mod tests { panic!("an exact accumulator") }; assert_eq!(*rolled, kind, "{query}"); - let variant = rollup_pass2_candidates(&roots, &sites).unwrap().remove(0); + let pass2_candidate = rollup_pass2_candidates(&roots, &rule_candidates) + .unwrap() + .remove(0); assert_eq!( - operator(&variant.roots.unwrap()[0]).schema, + operator(&pass2_candidate.roots.unwrap()[0]).schema, operator(&roots[0]).schema, "{query}" ); @@ -2339,18 +2348,18 @@ mod tests { #[test] fn exact_topk_reuses_the_larger_ranking() { let roots = exact(&["topk(5, sum by (job) (m))", "topk(10, sum by (job) (m))"]); - let sites = topk_reuse_candidates(&roots); - let [site] = sites.as_slice() else { - panic!("one site: {sites:?}") + let rule_candidates = topk_reuse_candidates(&roots); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one candidate: {rule_candidates:?}") }; - assert_eq!(site.members.len(), 2); - let Some(NonASAPOp::Limit { n: Some(10), .. }) = site.ranking.non_asap() else { - panic!("{:?}", site.ranking.operator) + assert_eq!(rule_candidate.members.len(), 2); + let Some(NonASAPOp::Limit { n: Some(10), .. }) = rule_candidate.ranking.non_asap() else { + panic!("{:?}", rule_candidate.ranking.operator) }; - let variant = topk_reuse_pass2_candidates(&roots, &sites) + let pass2_candidate = topk_reuse_pass2_candidates(&roots, &rule_candidates) .unwrap() .remove(0); - let reused = variant.roots.unwrap(); + let reused = pass2_candidate.roots.unwrap(); let read = |root: &QueryRoot| { let Some(NonASAPOp::Aggregate { child, measures, .. diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index b8ce2ac3..6bfd519b 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -776,13 +776,20 @@ pub fn select_plan( /// order; `None` without candidates. fn choose(ranking: Ranking, costs: &[&CandidateCost]) -> Option<(usize, Vec)> { let front: Vec = match ranking { - Ranking::Front => (0..costs.len()) - .filter(|&i| { - !costs - .iter() - .any(|other| dominates(&other.usage, &costs[i].usage)) - }) - .collect(), + Ranking::Front => { + // Each candidate's worst latency once, not once per comparison. + let latencies: Vec = costs.iter().map(|c| worst_latency_ms(&c.usage)).collect(); + (0..costs.len()) + .filter(|&i| { + !(0..costs.len()).any(|j| { + dominates_at( + (&costs[j].usage, latencies[j]), + (&costs[i].usage, latencies[i]), + ) + }) + }) + .collect() + } _ => Vec::new(), }; // Under the Pareto front, only a front member can be selected. @@ -843,7 +850,11 @@ fn capacity_violation(usage: &ResourceUsage, limits: &ClusterLimits) -> Option bool { - let (la, lb) = (worst_latency_ms(a), worst_latency_ms(b)); + dominates_at((a, worst_latency_ms(a)), (b, worst_latency_ms(b))) +} + +/// [`dominates`] with each usage's worst-query latency already computed. +fn dominates_at((a, la): (&ResourceUsage, f64), (b, lb): (&ResourceUsage, f64)) -> bool { a.peak_memory_bytes <= b.peak_memory_bytes && a.disk_bytes <= b.disk_bytes && !cheaper(lb, la) diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index 65a96129..9e2ec271 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -151,7 +151,9 @@ pub fn pass2_label( .shared_builds .iter() .map(|builds| { - let queries = readers(pass1_roots, |node| builds.iter().any(|b| same(node, b))); + let queries = readers(pass1_roots, |node| { + builds.iter().any(|b| same_computation(node, b)) + }); let family = match builds[0].asap().and_then(ASAPOp::produced_state) { Some(FieldDataType::Sketch(kind, _)) => format!("{:?}", kind.algorithm()), _ => builds[0].operator.kind_name().to_string(), @@ -200,8 +202,14 @@ pub fn pass2_label( format!( "{} for {} rolled up from {}", state_name(&candidate.coarse), - query_names(&readers(pass1_roots, |n| same(n, &candidate.coarse))), - query_names(&readers(pass1_roots, |n| same(n, &candidate.finer))) + query_names(&readers(pass1_roots, |n| same_computation( + n, + &candidate.coarse + ))), + query_names(&readers(pass1_roots, |n| same_computation( + n, + &candidate.finer + ))) ) }) .chain(pass2.topk_reuse_candidates.iter().map(|candidate| { @@ -211,7 +219,10 @@ pub fn pass2_label( }; format!( "top-k for {} from one top-{k}", - query_names(&readers(candidate_roots, |n| same(n, &candidate.ranking))) + query_names(&readers(candidate_roots, |n| same_computation( + n, + &candidate.ranking + ))) ) })) .collect(); @@ -228,7 +239,7 @@ pub fn pass2_label( format!( "{} ({})", node.operator.kind_name(), - query_names(&readers(candidate_roots, |n| same(n, node))) + query_names(&readers(candidate_roots, |n| same_computation(n, node))) ) }) .collect(); @@ -263,13 +274,9 @@ fn readers(roots: &[QueryRoot], matches: impl Fn(&OperatorNode) -> bool) -> Vec< .collect() } -/// Whether `node` is `other` or computes the same thing: the same -/// operators, schemas and children, whatever guarantees, timings and -/// materializations later steps attached. -fn same(node: &OperatorNode, other: &Rc) -> bool { - same_computation(node, other) -} - +/// Whether `a` is `b` or computes the same thing: the same operators, +/// schemas and children, whatever guarantees, timings and materializations +/// later steps attached. fn same_computation(a: &OperatorNode, b: &OperatorNode) -> bool { std::ptr::eq(a, b) || (a.operator.map_children(|_| ()) == b.operator.map_children(|_| ()) @@ -318,7 +325,7 @@ fn is_pane_chain_of(node: &OperatorNode, candidate: &PaneCandidate) -> bool { && range .children() .first() - .is_some_and(|window_input| same(child, window_input)) + .is_some_and(|window_input| same_computation(child, window_input)) } /// Whether two pane candidates become the same pane chain: the same diff --git a/crates/devtools/src/planner_layering_examples.rs b/crates/devtools/src/planner_layering_examples.rs index 7642004d..f1160b36 100644 --- a/crates/devtools/src/planner_layering_examples.rs +++ b/crates/devtools/src/planner_layering_examples.rs @@ -167,26 +167,44 @@ pub fn example2() -> (PlanningWorkload, asap_frontend_sql::SqlCatalog) { /// The metric Example 3's Patterns A and B read. pub const PATTERN_METRIC: &str = "latency_ms"; -/// Pattern A's five queries: (PromQL, its range window in ms, its offset -/// before T in ms); the window and offset restate the query text. -pub const PATTERN_A: [(&str, u64, u64); 5] = [ - ("quantile_over_time(0.99, latency_ms[5y])", 5 * YEAR_MS, 0), - ("quantile_over_time(0.99, latency_ms[1y])", YEAR_MS, 0), - ( - "quantile_over_time(0.99, latency_ms[1y] offset 1y)", - YEAR_MS, - YEAR_MS, - ), - ( - "quantile_over_time(0.99, latency_ms[1y] offset 2y)", - YEAR_MS, - 2 * YEAR_MS, - ), - ( - "quantile_over_time(0.99, latency_ms[3y] offset 2y)", - 3 * YEAR_MS, - 2 * YEAR_MS, - ), +/// One of Pattern A's queries, with the window and offset its text states. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PatternAQuery { + /// The PromQL text. + pub promql: &'static str, + /// Its range window, in ms. + pub window_ms: u64, + /// Its offset before T, in ms. + pub offset_before_t_ms: u64, +} + +/// Pattern A's five queries; the window and offset restate the query text. +pub const PATTERN_A: [PatternAQuery; 5] = [ + PatternAQuery { + promql: "quantile_over_time(0.99, latency_ms[5y])", + window_ms: 5 * YEAR_MS, + offset_before_t_ms: 0, + }, + PatternAQuery { + promql: "quantile_over_time(0.99, latency_ms[1y])", + window_ms: YEAR_MS, + offset_before_t_ms: 0, + }, + PatternAQuery { + promql: "quantile_over_time(0.99, latency_ms[1y] offset 1y)", + window_ms: YEAR_MS, + offset_before_t_ms: YEAR_MS, + }, + PatternAQuery { + promql: "quantile_over_time(0.99, latency_ms[1y] offset 2y)", + window_ms: YEAR_MS, + offset_before_t_ms: 2 * YEAR_MS, + }, + PatternAQuery { + promql: "quantile_over_time(0.99, latency_ms[3y] offset 2y)", + window_ms: 3 * YEAR_MS, + offset_before_t_ms: 2 * YEAR_MS, + }, ]; /// How Pattern A's batch recurs (Example 4 varies it). @@ -215,8 +233,8 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni Some( PATTERN_A .iter() - .map(|&(query, ..)| BatchEntry { - query: Query(query.into()), + .map(|pattern| BatchEntry { + query: Query(pattern.promql.into()), requirements: pattern_a_requirements(), predictability: Predictability::AdHoc, invocations: 1, @@ -231,8 +249,8 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni Some( PATTERN_A .iter() - .map(|&(query, ..)| RepeatingEntry { - query: Query(query.into()), + .map(|pattern| RepeatingEntry { + query: Query(pattern.promql.into()), demand: RepeatedDemand::FixedInterval(RepetitionInterval( (30 * 24 * 60 * MINUTE_MS) as u32, )), diff --git a/crates/integration-tests/tests/planner_layering_example3.rs b/crates/integration-tests/tests/planner_layering_example3.rs index 814fb122..40a097ac 100644 --- a/crates/integration-tests/tests/planner_layering_example3.rs +++ b/crates/integration-tests/tests/planner_layering_example3.rs @@ -67,7 +67,7 @@ fn workload_encodes_example3() { ); a.validate().expect("valid Pattern A"); let queries: Vec<_> = a.query_workload.entries().map(|e| e.query.0).collect(); - assert_eq!(queries, PATTERN_A.map(|(q, ..)| q)); + assert_eq!(queries, PATTERN_A.map(|pattern| pattern.promql)); let b = pattern_b(); b.validate().expect("valid Pattern B"); assert_eq!(evaluation_interval_ms(&b, 0), Some(PATTERN_B_INTERVAL_MS)); diff --git a/crates/integration-tests/tests/planner_layering_example4.rs b/crates/integration-tests/tests/planner_layering_example4.rs index 95d1de17..2c443049 100644 --- a/crates/integration-tests/tests/planner_layering_example4.rs +++ b/crates/integration-tests/tests/planner_layering_example4.rs @@ -148,8 +148,8 @@ fn stage2_physical_asap_optimization_at_rest_runs_nothing_at_ingestion_time() { #[test] fn stage2_physical_asap_optimization_materialized_output_covers_its_consumers() { // Each query's range window and offset, as its PromQL text states them. - let windows: Vec = PATTERN_A.iter().map(|(_, window, _)| *window).collect(); - let offsets: Vec = PATTERN_A.iter().map(|(_, _, before_t)| *before_t).collect(); + let windows: Vec = PATTERN_A.iter().map(|p| p.window_ms).collect(); + let offsets: Vec = PATTERN_A.iter().map(|p| p.offset_before_t_ms).collect(); for (workload, windows, offsets) in [ (once(), windows.clone(), offsets.clone()), (monthly(), windows.clone(), offsets.clone()), From d285781cf0f43fa3b29af7033b8f14423ba315bf Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:03 +0000 Subject: [PATCH 5/7] test: port lost IR regression tests and cover gaps found in reviews - Port three tests lost when the legacy IR was removed: filtered and unfiltered aggregates do not merge, without() drops a renamed sample column, and series identity accepts a direct vector literal. - Pass 1 HAVING: a scalar-reference consumer keeps a key-only HAVING above the output; of stacked Filters only the adjacent one is HAVING. - Materialization: the population test asserts something is stored, and a mixed-placement DAG (one summary stored, one rebuilt) answers alike. - Tumbling panes: the sample set is a parameter instead of a thread-local; new tests for an empty middle pane, counters over panes with an outer offset, and gap/overlap rejection. - Executor: factory and DAG heap states merge interchangeably. - Stage 3: a shared summary takes the smallest failure-probability share; a declared counter prices the CMS+heap candidate per node. - execution_timed_dag: an empty workload is NoRoots, a literal-only one a DAG with no nodes; exact-state roots and their state run at query time. --- crates/asap-aware-mapping/src/pass/mod.rs | 24 +- .../src/stage3_plan_selection.rs | 59 +++++ .../tests/per_computation_candidates.rs | 57 +++++ .../tests/deployment.rs | 44 ++++ .../tests/materialization_e2e.rs | 96 ++++++-- .../tests/tumbling_pane_merge.rs | 224 ++++++++++++++---- crates/planner/tests/summary_sharing.rs | 19 ++ crates/types/src/ir/aggregate_schema.rs | 20 ++ crates/types/src/ir/cse.rs | 32 +++ crates/types/src/ir/schema_support.rs | 18 ++ 10 files changed, 519 insertions(+), 74 deletions(-) diff --git a/crates/asap-aware-mapping/src/pass/mod.rs b/crates/asap-aware-mapping/src/pass/mod.rs index 432ed33c..66f1ab1c 100644 --- a/crates/asap-aware-mapping/src/pass/mod.rs +++ b/crates/asap-aware-mapping/src/pass/mod.rs @@ -281,7 +281,7 @@ impl PlanOutput { pub fn execution_timed_dag( &self, ) -> Result { - // One memo, so a node shared by several roots is timed and exported once. + // One memo, so a node several roots share by `Rc` is timed and exported once. let mut memo = TimingMemo::new(); let assignment = MaterializationAssignment::default(); let timed = self @@ -619,6 +619,28 @@ mod tests { assert!(check_contract(&output, "stub", 0).is_ok()); } + /// An empty workload has no DAG: the export's `NoRoots`, wrapped in + /// `InvalidPhysicalDAG`. A workload of only literal scalars exports a DAG + /// with no nodes, just its `scalar_roots`. + #[test] + fn execution_timed_dag_of_an_empty_and_a_literal_only_workload() { + use asap_types::ir::physical_export::PhysicalASAPDAGValidationError; + use asap_types::ir::ScalarExpr; + assert_eq!( + PlanOutput::new(Vec::new()).execution_timed_dag().err(), + Some(ExecutionDataStateError::InvalidPhysicalDAG( + PhysicalASAPDAGValidationError::NoRoots + )) + ); + let mut literal = PlanOutput::new(Vec::new()); + literal.scalar_roots = vec![(0, ScalarExpr::literal_f64(1.0))]; + let dag = literal.execution_timed_dag().unwrap(); + assert!(dag.nodes.is_empty() && dag.roots.is_empty()); + let [(0, ScalarExpr::Literal(_))] = dag.scalar_roots.as_slice() else { + panic!("one literal scalar root: {:?}", dag.scalar_roots) + }; + } + /// Registering a name twice fails instead of overwriting, so a comparison /// run cannot silently measure one pass twice. #[test] diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index 6bfd519b..5949f898 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -3723,6 +3723,65 @@ mod tests { assert_eq!(kll_params(&select(&per_value).plan), stage1); } + /// With its metric declared a counter, the whole-expression CMS+heap + /// candidate (P2) proves its weights non-negative, is valid and is + /// priced: its per-node costs cover its DAG and sum to its total. + #[test] + fn a_declared_counter_prices_the_count_min_heap_candidate() { + use asap_types::workload::MetricType; + let candidates = candidates(); + let target = AccuracyTarget::max_error(0.01) + .failure_probability(0.001) + .accuracy_metric(AccuracyMetric::ABSOLUTE_STREAM_L1); + let counter = with_metric_types(data(), &[("m", MetricType::Counter)]); + let selection = select_plan( + &candidates, + &[entry_with(target, repeating(10_000))], + &counter, + PlanningModels::builtin().with_evidence(&EveryOutputMeasuredExact), + ) + .unwrap(); + let p2 = candidates.iter().find(|c| c.id == "P2").unwrap(); + let cost = &selection.costs["P2"]; + let mut ids: Vec<_> = p2.dag.nodes.iter().map(|n| n.id).collect(); + ids.sort(); + let mut keys: Vec<_> = cost.per_node.keys().copied().collect(); + keys.sort(); + assert_eq!(keys, ids); + let sum: f64 = cost + .per_node + .values() + .map(|n| n.cost_per_second.unwrap()) + .sum(); + assert_eq!(cost.cost_per_second.unwrap(), sum); + } + + /// One summary two queries read is sized for the stricter one: its + /// failure probability is the smallest share either query needs (the + /// largest `n · G` over its readers). + #[test] + fn a_shared_summary_takes_the_smallest_failure_probability_share() { + let (candidates, lenient) = per_series_kll(AccuracyScope::PerValue); + let root = candidates[0].roots[0].clone(); + let strict = lenient.clone().failure_probability(0.001); + let roots = [root.clone(), root]; + for targets in [[lenient.clone(), strict.clone()], [strict, lenient]] { + let resized = resize_for_failure_probability(&roots, &targets, &data()).unwrap(); + let kll = kll_params(&resized); + assert_eq!(kll.len(), 2, "{kll:?}"); + for params in kll { + let SketchParams::Kll { + failure_probability, + .. + } = params + else { + panic!("{params:?}") + }; + assert_eq!(failure_probability, Some(0.001)); + } + } + } + /// Without a bound on the values one evaluation returns, a per-value /// guarantee does not meet a whole-result target: a PromQL vector whose /// source declares no series count, and a SQL relation (until a bound on diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index cb658f51..03d1f577 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -621,6 +621,63 @@ fn having_on_one_consumer_of_a_shared_aggregate_stays_above() { assert_eq!(targets[0].having.len(), 2); } +/// A scalar reference to a grouped aggregate (`Filter(scan, EXISTS G)` as a +/// second root) is a consumer that reads it without the `HAVING`, so the +/// first root's key-only `HAVING` stays above the output. +#[test] +fn a_scalar_reference_consumer_keeps_a_key_only_having_above() { + let node = grouped_count(None); + let key = equals(0, ScalarValue::Utf8("a".into())); + let filtered = having_filter(node.clone(), key.clone()); + let exists = having_filter( + scan(false), + Predicate(ScalarExpr::Exists { + subquery: node.clone(), + negated: false, + }), + ); + let targets = generate_per_computation_candidates(vec![ + (0, QueryRoot::Operator(filtered.clone())), + (1, QueryRoot::Operator(exists)), + ]) + .unwrap() + .targets; + let target = targets + .iter() + .find(|target| Rc::ptr_eq(&target.aggregate, &node)) + .expect("the grouped count's target"); + assert_eq!( + target.having, + vec![HavingEntry { + filter: Some(filtered), + predicate: key, + placement: HavingPlacement::AboveOutput, + }] + ); +} + +/// Of two stacked `Filter`s above an aggregate, only the adjacent one is +/// its `HAVING`; the outer one filters the inner `Filter`'s output. +#[test] +fn only_the_adjacent_of_stacked_filters_is_having() { + let node = grouped_count(None); + let inner_key = equals(0, ScalarValue::Utf8("a".into())); + let inner = having_filter(node.clone(), inner_key.clone()); + let outer = having_filter(inner.clone(), equals(0, ScalarValue::Utf8("b".into()))); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(outer))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert_eq!( + targets[0].having, + vec![HavingEntry { + filter: Some(inner), + predicate: inner_key, + placement: HavingPlacement::BelowAggregate, + }] + ); +} + /// A global aggregate's `HAVING` always stays above the output, even a /// constant one: the one output row exists with no input rows. #[test] diff --git a/crates/asap-physical-operators/tests/deployment.rs b/crates/asap-physical-operators/tests/deployment.rs index 65e0f175..e1779c07 100644 --- a/crates/asap-physical-operators/tests/deployment.rs +++ b/crates/asap-physical-operators/tests/deployment.rs @@ -294,6 +294,50 @@ fn heap_sketch_states_load_as_dag_summary_values() { } } +// A factory-built top-k state and the state the DAG's keyed build makes +// (`WeightedFrequency::new`, updated with the key's `Utf8` label values) are +// interchangeable: merged either way, they equal one build over both inputs. +#[test] +fn factory_and_dag_heap_states_are_interchangeable() { + use asap_physical_operators::{ + summary_kernels::weighted_frequency::WeightedFrequency, values::Value, + }; + use asap_sketchlib::FrequencyAlgorithm; + let estimate = |state: &dyn AggregateCore, i: usize| { + state + .as_any() + .downcast_ref::() + .expect("the DAG's weighted frequency state") + .point_estimate(&[Value::Utf8(format!("host-{i}").into())]) + .unwrap() + }; + for (algorithm, frequency) in [ + (SketchAlgorithm::CmsWithHeap, FrequencyAlgorithm::Cms), + ( + SketchAlgorithm::CountSketchWithHeap, + FrequencyAlgorithm::CountSketch, + ), + ] { + let family = heap_sketch(algorithm.clone(), 1_024, 5, 4); + let factory = build_weighted_frequency(&family, [(0, 1.0), (1, 2.0)]); + let mut dag = WeightedFrequency::new(frequency, 1_024, 5, 4).unwrap(); + dag.update(&[Value::Utf8("host-0".into())], 3.0).unwrap(); + let union = build_weighted_frequency(&family, [(0, 1.0), (1, 2.0), (0, 3.0)]); + for merged in [ + factory.merge_with(&dag).unwrap(), + dag.merge_with(factory.as_ref()).unwrap(), + ] { + for i in 0..2 { + assert_eq!( + estimate(merged.as_ref(), i), + estimate(union.as_ref(), i), + "{algorithm:?} host-{i}" + ); + } + } + } +} + // Count-Min needs nonnegative weights; a negative input is an error at // validation, never silently dropped. #[test] diff --git a/crates/integration-tests/tests/materialization_e2e.rs b/crates/integration-tests/tests/materialization_e2e.rs index 6f3c8ed6..d687640d 100644 --- a/crates/integration-tests/tests/materialization_e2e.rs +++ b/crates/integration-tests/tests/materialization_e2e.rs @@ -208,6 +208,29 @@ fn job_samples<'a>( } } +/// Values 1..=100 at 300 s, one row each, in every Float64 column of +/// `schema`. +fn values_1_to_100(schema: &SchemaRef, _metric: &str) -> Batch { + let rows = (1..=100) + .map(|value| { + schema + .fields + .iter() + .map(|field| match field.dtype { + asap_types::post_asap::FieldDataType::Plain( + asap_types::pre_asap::DataType::Float64, + ) => Value::Float64(f64::from(value)), + asap_types::post_asap::FieldDataType::Plain( + asap_types::pre_asap::DataType::Timestamp, + ) => Value::Timestamp(300_000), + _ => panic!("unexpected field {field:?}"), + }) + .collect() + }) + .collect(); + Batch::try_new(schema.clone(), rows).unwrap() +} + /// `MatAtIngestion` puts the quantile's state in precompute; /// `NotMaterialized` leaves precompute empty and reads the raw source at /// query time; both answer alike. @@ -217,32 +240,12 @@ fn materialization_decides_precompute_contents() { let mut answers = Vec::new(); for assignment in placements() { let dag = timed_dag(&root, &assignment); - let batch = |schema: &SchemaRef, _: &str| { - let rows = (1..=100) - .map(|value| { - schema - .fields - .iter() - .map(|field| match field.dtype { - asap_types::post_asap::FieldDataType::Plain( - asap_types::pre_asap::DataType::Float64, - ) => Value::Float64(f64::from(value)), - asap_types::post_asap::FieldDataType::Plain( - asap_types::pre_asap::DataType::Timestamp, - ) => Value::Timestamp(300_000), - _ => panic!("unexpected field {field:?}"), - }) - .collect() - }) - .collect(); - Batch::try_new(schema.clone(), rows).unwrap() - }; let ingestion = Scope::Ingestion { window_start_ms: 0, window_end_ms: 300_000, revision: 1, }; - let (values, frontier) = execute(&dag, &batch, ingestion, 300_000); + let (values, frontier) = execute(&dag, &values_1_to_100, ingestion, 300_000); assert_eq!(frontier, states(&dag)); answers.push(values); } @@ -251,6 +254,56 @@ fn materialization_decides_precompute_contents() { assert_eq!(answers[0], answers[1]); } +/// Mixed placement in one DAG: one summary `MatAtIngestion` (in precompute) +/// and the other `NotMaterialized` (built from the raw source at query +/// time) answer like the all-query-time placement. +#[test] +fn mixed_placement_answers_like_query_time() { + let accuracy = AccuracyTarget::max_error(0.01); + let root = lower_promql( + "quantile(0.99, latency) + quantile(0.5, latency)", + accuracy.clone(), + ) + .unwrap(); + let sketches = |candidate: &Rc| -> Vec> { + OperatorNode::reachable(candidate) + .into_iter() + .filter(|node| { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family: asap_types::post_asap::FieldDataType::Sketch(..), + .. + }) + ) + }) + .collect() + }; + let root = logical_candidates(root, accuracy) + .into_iter() + .find(|candidate| sketches(candidate).len() == 2) + .expect("a candidate with two sketches"); + let mut mixed = MaterializationAssignment::default(); + mixed.set(&sketches(&root)[0], AT_INGESTION); + let mut answers = Vec::new(); + for assignment in [MaterializationAssignment::default(), mixed] { + let dag = timed_dag(&root, &assignment); + let ingestion = Scope::Ingestion { + window_start_ms: 0, + window_end_ms: 300_000, + revision: 1, + }; + let (values, frontier) = execute(&dag, &values_1_to_100, ingestion, 300_000); + assert_eq!(frontier, states(&dag)); + answers.push((values, frontier.len())); + } + // Only the assigned sketch is stored. + assert_eq!(answers[0].1, 0); + assert_eq!(answers[1].1, 1); + assert_eq!(answers[1].0.len(), 1); + assert_eq!(answers[0].0, answers[1].0); +} + /// A maintained current-series population: `MatAtIngestion` stores it in /// precompute, `NotMaterialized` rebuilds it from the raw source at query /// time; both rank alike. @@ -269,6 +322,7 @@ fn population_materialization_decides_precompute_contents() { .expect("a maintained population"); let mut stored = MaterializationAssignment::default(); stored.set(&population, AT_INGESTION); + assert!(!states(&timed_dag(&root, &stored)).is_empty()); let mut answers = Vec::new(); for assignment in [MaterializationAssignment::default(), stored] { let dag = timed_dag(&root, &assignment); diff --git a/crates/integration-tests/tests/tumbling_pane_merge.rs b/crates/integration-tests/tests/tumbling_pane_merge.rs index fb0bab0a..633915a8 100644 --- a/crates/integration-tests/tests/tumbling_pane_merge.rs +++ b/crates/integration-tests/tests/tumbling_pane_merge.rs @@ -87,15 +87,19 @@ fn tumbling_panes(root: &Rc) -> (Rc, Vec) -> (Rc, Rc) { let roots = [QueryRoot::Operator(Rc::clone(root))]; - let sites = pane_candidates(&roots, &[Some(PANE_MS as u64)]); - let [site] = sites.as_slice() else { - panic!("one build to split: {sites:?}") + let rule_candidates = pane_candidates(&roots, &[Some(PANE_MS as u64)]); + let [rule_candidate] = rule_candidates.as_slice() else { + panic!("one build to split: {rule_candidates:?}") }; - assert_eq!((site.pane_width_ms, site.pane_count), (PANE_MS as u64, 5)); - let mut variants = window_composition_pass2_candidates(&roots, &sites).unwrap(); - let variant = variants.pop().unwrap(); - assert!(variants.is_empty()); - let QueryRoot::Operator(planned) = variant.roots.unwrap().remove(0) else { + assert_eq!( + (rule_candidate.pane_width_ms, rule_candidate.pane_count), + (PANE_MS as u64, 5) + ); + let mut pass2_candidates = + window_composition_pass2_candidates(&roots, &rule_candidates).unwrap(); + let pass2_candidate = pass2_candidates.pop().unwrap(); + assert!(pass2_candidates.is_empty()); + let QueryRoot::Operator(planned) = pass2_candidate.roots.unwrap().remove(0) else { panic!("an operator root") }; planned.validate_structure().unwrap(); @@ -124,21 +128,17 @@ fn replace( ) } -thread_local! { - /// How far after each 15-s step the samples fall (0: on the pane - /// boundaries). - static PHASE_MS: std::cell::Cell = const { std::cell::Cell::new(0) }; -} +/// One raw sample: series, timestamp (ms) and value. +type Sample = (&'static str, i64, f64); /// Two series, four positive samples per minute in [0, 5m), each -/// [`PHASE_MS`] after its 15-s step. -fn samples() -> Vec<(&'static str, i64, f64)> { - let phase = PHASE_MS.with(|p| p.get()); +/// `phase_ms` after its 15-s step (0: on the pane boundaries). +fn samples(phase_ms: i64) -> Vec { let mut rows = vec![]; for (s, series) in ["a", "b"].into_iter().enumerate() { for minute in 0..PANES { for j in 0..4 { - let ts = minute * PANE_MS + j * 15_000 + phase; + let ts = minute * PANE_MS + j * 15_000 + phase_ms; rows.push((series, ts, (s * 100) as f64 + (minute * 4 + j + 1) as f64)); } } @@ -196,7 +196,7 @@ fn schema_of(dag: &PhysicalASAPDAG, id: u64) -> Arc { /// Raw rows for each `TimeRange` input. A deployment supplies pane `i` with /// the samples in `[T - (i+1)·w, T - i·w)`; a single build reads all samples. -fn raw_inputs(exported: &Exported) -> BTreeMap { +fn raw_inputs(exported: &Exported, samples: &[Sample]) -> BTreeMap { exported .ranges .iter() @@ -208,10 +208,10 @@ fn raw_inputs(exported: &Exported) -> BTreeMap { } else { end - PANE_MS }; - let rows = samples() - .into_iter() + let rows = samples + .iter() .filter(|(_, ts, _)| (start..end).contains(ts)) - .map(|(series, ts, value)| { + .map(|&(series, ts, value)| { let labels = BTreeMap::from([ ("__name__".to_string(), "m".to_string()), ("series".to_string(), series.to_string()), @@ -260,9 +260,9 @@ fn contracts(inputs: &BTreeMap) -> BTreeMap { /// Compile `root` with everything at query time (Pattern B2: panes rebuilt /// from raw rows at query time). -fn compile_at_query_time(root: &Rc) -> Result { +fn compile_at_query_time(root: &Rc, samples: &[Sample]) -> Result { let exported = export(root, &MaterializationAssignment::default()); - let inputs = raw_inputs(&exported); + let inputs = raw_inputs(&exported, samples); compile( &exported.dag, contracts(&inputs), @@ -272,9 +272,9 @@ fn compile_at_query_time(root: &Rc) -> Result { Ok(exported) } -fn execute(root: &Rc) -> Vec> { - let exported = compile_at_query_time(root).unwrap(); - let inputs = raw_inputs(&exported); +fn execute(root: &Rc, samples: &[Sample]) -> Vec> { + let exported = compile_at_query_time(root, samples).unwrap(); + let inputs = raw_inputs(&exported, samples); let plan = compile( &exported.dag, contracts(&inputs), @@ -286,19 +286,23 @@ fn execute(root: &Rc) -> Vec> { /// Build the panes in ingestion runs (stored `MatAtIngestion`, each with its /// own build timestamp), then merge them in the query run (Pattern B1). -fn execute_retained(root: &Rc, panes: &[Rc]) -> Vec> { +fn execute_retained( + root: &Rc, + panes: &[Rc], + samples: &[Sample], +) -> Vec> { let mut assignment = MaterializationAssignment::default(); for pane in panes { assignment.set(pane, Materialization::MatAtIngestion(Medium::Memory)); } - execute_retained_as(root, &assignment) + execute_retained_as(root, &assignment, samples) } /// The planner's pane chain maintained at ingestion time, as Stage 2's B1 /// option assigns it (`MatAtIngestion` on the chain), then expanded: each /// pane carries the chain's timing and materialization, so the panes are /// built in ingestion runs and merged in the query run. -fn execute_maintained_chain(planned: &Rc) -> Vec> { +fn execute_maintained_chain(planned: &Rc, samples: &[Sample]) -> Vec> { let chain = OperatorNode::reachable(planned) .into_iter() .find(|n| matches!(n.asap(), Some(ASAPOp::PaneChain { .. }))) @@ -317,15 +321,16 @@ fn execute_maintained_chain(planned: &Rc) -> Vec> { Materialization::MatAtIngestion(Medium::Memory) ); } - execute_retained_as(&expanded, &MaterializationAssignment::default()) + execute_retained_as(&expanded, &MaterializationAssignment::default(), samples) } fn execute_retained_as( root: &Rc, assignment: &MaterializationAssignment, + samples: &[Sample], ) -> Vec> { let exported = export(root, assignment); - let inputs = raw_inputs(&exported); + let inputs = raw_inputs(&exported, samples); let contracts = contracts(&inputs); let mut retained = BTreeMap::new(); for &(build, pane) in &exported.builds { @@ -376,36 +381,42 @@ fn values(rows: &[Vec]) -> Vec { } /// Returns the per-series values, which every execution path agrees on. -fn assert_pane_merge_matches_single_build(query: &str, accuracy: AccuracyTarget) -> Vec { +fn assert_pane_merge_matches_single_build( + query: &str, + accuracy: AccuracyTarget, + samples: &[Sample], +) -> Vec { let single = single_build(query, accuracy); let (panes, pane_builds) = tumbling_panes(&single); - let expected = execute(&single); + let expected = execute(&single, samples); assert_eq!(expected.len(), 2, "{expected:?}"); assert!(expected.iter().all(|row| has_timestamp(row, EVALUATION_MS))); let result = values(&expected); let expected = render(&expected); assert_eq!( - render(&execute(&panes)), + render(&execute(&panes, samples)), expected, "panes rebuilt at query time" ); assert_eq!( - render(&execute_retained(&panes, &pane_builds)), + render(&execute_retained(&panes, &pane_builds, samples)), expected, "retained panes" ); let (planned, expanded) = planned_panes(&single); assert_eq!( - render(&execute(&expanded)), + render(&execute(&expanded, samples)), expected, "the planner's pane chain, expanded" ); assert_eq!( - render(&execute_maintained_chain(&planned)), + render(&execute_maintained_chain(&planned, samples)), expected, "the planner's pane chain, maintained at ingestion time" ); - let unexpanded = compile_at_query_time(&planned).err().unwrap_or_default(); + let unexpanded = compile_at_query_time(&planned, samples) + .err() + .unwrap_or_default(); assert!(unexpanded.contains("PaneChain"), "{unexpanded}"); result } @@ -414,8 +425,11 @@ fn assert_pane_merge_matches_single_build(query: &str, accuracy: AccuracyTarget) /// build, timestamped at the evaluation time. #[test] fn exact_sum_over_tumbling_panes_matches_single_build() { - let sums = - assert_pane_merge_matches_single_build("sum_over_time(m[5m])", AccuracyTarget::exact()); + let sums = assert_pane_merge_matches_single_build( + "sum_over_time(m[5m])", + AccuracyTarget::exact(), + &samples(0), + ); assert_eq!(sums, vec![210., 2210.]); } @@ -426,6 +440,7 @@ fn quantile_over_tumbling_panes_matches_single_build() { let p99 = assert_pane_merge_matches_single_build( "quantile_over_time(0.99, m[5m])", AccuracyTarget::max_error(0.05).accuracy_metric(AccuracyMetric::Relative), + &samples(0), ); let mut p99 = p99; p99.sort_by(f64::total_cmp); @@ -443,7 +458,8 @@ fn quantile_over_tumbling_panes_matches_single_build() { /// by one pane's minute. #[test] fn exact_rate_over_tumbling_panes_matches_single_build() { - let rates = assert_pane_merge_matches_single_build("rate(m[5m])", AccuracyTarget::exact()); + let rates = + assert_pane_merge_matches_single_build("rate(m[5m])", AccuracyTarget::exact(), &samples(0)); assert_eq!(rates.len(), 2); // 19 counted over 285 s of samples, extrapolated to the 5-min window. for rate in rates { @@ -456,15 +472,17 @@ fn exact_rate_over_tumbling_panes_matches_single_build() { /// of the newest pane. #[test] fn exact_increase_off_the_pane_boundaries_matches_single_build() { - PHASE_MS.with(|p| p.set(7_000)); - let increases = - assert_pane_merge_matches_single_build("increase(m[5m])", AccuracyTarget::exact()); + let increases = assert_pane_merge_matches_single_build( + "increase(m[5m])", + AccuracyTarget::exact(), + &samples(7_000), + ); assert_eq!(increases.len(), 2); } /// The single build of `query`, evaluated at `evaluation_ms` over every /// sample, as values. -fn whole_build_values_at(query: &str, evaluation_ms: i64) -> Vec { +fn whole_build_values_at(query: &str, evaluation_ms: i64, samples: &[Sample]) -> Vec { let root = single_build(query, AccuracyTarget::exact()); let exported = export(&root, &MaterializationAssignment::default()); let inputs: BTreeMap = exported @@ -472,9 +490,9 @@ fn whole_build_values_at(query: &str, evaluation_ms: i64) -> Vec { .iter() .map(|&(id, _)| { let schema = schema_of(&exported.dag, id); - let rows = samples() - .into_iter() - .map(|(series, ts, value)| { + let rows = samples + .iter() + .map(|&(series, ts, value)| { let labels = BTreeMap::from([ ("__name__".to_string(), "m".to_string()), ("series".to_string(), series.to_string()), @@ -506,12 +524,13 @@ fn whole_build_values_at(query: &str, evaluation_ms: i64) -> Vec { /// evaluation time minus the offset, not at the evaluation time. #[test] fn offset_counter_window_ends_at_the_offset() { - PHASE_MS.with(|p| p.set(7_000)); + let samples = samples(7_000); for function in ["rate", "increase"] { - let plain = whole_build_values_at(&format!("{function}(m[5m])"), EVALUATION_MS); + let plain = whole_build_values_at(&format!("{function}(m[5m])"), EVALUATION_MS, &samples); let shifted = whole_build_values_at( &format!("{function}(m[5m] offset 1m)"), EVALUATION_MS + PANE_MS, + &samples, ); assert_eq!(shifted, plain, "{function}"); } @@ -521,7 +540,108 @@ fn offset_counter_window_ends_at_the_offset() { /// build: the merge joins consecutive panes across their boundaries. #[test] fn exact_increase_over_tumbling_panes_matches_single_build() { - let increases = - assert_pane_merge_matches_single_build("increase(m[5m])", AccuracyTarget::exact()); + let increases = assert_pane_merge_matches_single_build( + "increase(m[5m])", + AccuracyTarget::exact(), + &samples(0), + ); assert_eq!(increases.len(), 2); } + +/// A middle pane with no samples, for one series or for every series: the +/// merge joins the panes on either side of the empty one, and Increase +/// matches one 5-minute build. +#[test] +fn exact_increase_over_an_empty_middle_pane_matches_single_build() { + let middle = 2 * PANE_MS..3 * PANE_MS; + for emptied in [&["a"][..], &["a", "b"]] { + let gapped: Vec = samples(0) + .into_iter() + .filter(|(series, ts, _)| !(emptied.contains(series) && middle.contains(ts))) + .collect(); + let increases = assert_pane_merge_matches_single_build( + "increase(m[5m])", + AccuracyTarget::exact(), + &gapped, + ); + assert_eq!(increases.len(), 2, "{emptied:?}"); + } +} + +/// A counter over `[5m] offset 1m` split into the planner's panes, read +/// at query time or maintained at ingestion time, matches one build: the +/// panes' shifts fold the outer offset, and the merged counter window ends +/// at the evaluation time minus the offset. +#[test] +fn a_counter_over_panes_with_an_outer_offset_matches_single_build() { + let samples = samples(7_000); + for function in ["rate", "increase"] { + let single = single_build( + &format!("{function}(m[5m] offset 1m)"), + AccuracyTarget::exact(), + ); + let expected = render(&execute(&single, &samples)); + assert_eq!(expected.len(), 2, "{function}: {expected:?}"); + let (planned, expanded) = planned_panes(&single); + assert_eq!( + render(&execute(&expanded, &samples)), + expected, + "{function}: expanded panes" + ); + assert_eq!( + render(&execute_maintained_chain(&planned, &samples)), + expected, + "{function}: panes maintained at ingestion time" + ); + } +} + +/// `root`'s summary built from panes `(shift, width)` merged by one +/// `SummaryMerge`, both in minutes, or the IR's refusal of that merge. +fn panes_at(root: &Rc, panes: &[(i64, i64)]) -> Result, String> { + let agg = summary_agg(root); + let Some(NonASAPOp::TimeRange { kind, child, .. }) = agg.children()[0].non_asap() else { + panic!("expected a raw time range under the summary"); + }; + let children = panes + .iter() + .map(|&(shift, width)| { + let shifted = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift { + shift: TimeShift { + offset_ms: shift * PANE_MS, + at: None, + }, + child: child.clone(), + })) + .unwrap(); + let range = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: std::time::Duration::from_millis((width * PANE_MS) as u64), + kind: *kind, + child: shifted, + })) + .unwrap(); + OperatorNode::new_shared(agg.operator.map_children(|_| range.clone())).unwrap() + }) + .collect(); + let merged = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children })) + .map_err(|error| format!("{error:?}"))?; + Ok(replace(root, &agg, &merged)) +} + +/// Counter panes form one window only back to back: with a gap between +/// them, compiling the counter evaluation is an error; overlapping panes +/// are already refused by the `SummaryMerge`, whose counter inputs must be +/// disjoint in time. +#[test] +fn counter_panes_with_a_gap_or_an_overlap_are_rejected() { + let single = single_build("rate(m[5m])", AccuracyTarget::exact()); + let gapped = panes_at(&single, &[(0, 1), (2, 1)]).unwrap(); + let error = compile_at_query_time(&gapped, &samples(0)) + .err() + .unwrap_or_default(); + assert!(error.contains("ambiguous logical windows"), "{error}"); + let overlap = panes_at(&single, &[(0, 2), (1, 2)]) + .err() + .unwrap_or_default(); + assert!(overlap.contains("InputsOverlap"), "{overlap}"); +} diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index a1b042df..4585ebb8 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -547,5 +547,24 @@ async fn exact_state_roots_are_finalized() { "{query}: {:?}", root.output_schema ); + // With nothing materialized, the root and the exact state it reads + // run at query time. + use asap_types::post_asap::{DataPrimitive, ExecutionDataState, ExecutionTiming}; + assert_eq!(root.output_state, ExecutionDataState::QUERY_ROWS, "{query}"); + let state = dag + .edges + .iter() + .find(|edge| edge.consumer == root.id) + .map(|edge| edge.producer) + .unwrap(); + let state = dag.nodes.iter().find(|node| node.id == state).unwrap(); + assert_eq!( + state.output_state, + ExecutionDataState { + timing: ExecutionTiming::QueryTime, + primitive: DataPrimitive::SummaryState, + }, + "{query}" + ); } } diff --git a/crates/types/src/ir/aggregate_schema.rs b/crates/types/src/ir/aggregate_schema.rs index 4c868f55..f7476047 100644 --- a/crates/types/src/ir/aggregate_schema.rs +++ b/crates/types/src/ir/aggregate_schema.rs @@ -356,4 +356,24 @@ mod tests { assert!(out.time_index.is_none()); assert_eq!(out.unique_keys, vec![vec![0]]); } + + // A nested aggregate's renamed sample value is not a kept label: + // `sum without (inst) (sum by (inst, job) (m))` over `[inst, job, sum]`. + #[test] + fn without_aggregate_drops_a_renamed_sample_value() { + let inner = Schema::new(vec![ + col("inst", DataType::Utf8, true), + col("job", DataType::Utf8, true), + col("sum", DataType::Float64, false), + ]); + let out = aggregate_output_schema( + &inner, + &Reduction::Reduce(GroupKeys::without(vec![0])), + &[AggIntent::Sum { col: None }], + &[], + ) + .expect("valid excluded column"); + let names: Vec<_> = out.fields.iter().map(|c| c.name.as_str()).collect(); + assert_eq!(names, vec!["job", "sum"]); + } } diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs index 50af4f4c..e7811af9 100644 --- a/crates/types/src/ir/cse.rs +++ b/crates/types/src/ir/cse.rs @@ -365,6 +365,38 @@ mod tests { assert_ne!(ra, rb); } + // Two aggregates that differ only in one measure's `FILTER` predicate + // compute different values, so structural sharing keeps them apart. + #[test] + fn filtered_and_unfiltered_aggregates_do_not_merge() { + use crate::ir::Predicate; + use crate::pre_asap::ScalarValue; + let unfiltered = quantile_agg(vec![1], Some(2), 0.5); + let filtered = node(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![1]), + measures: vec![AggIntent::Quantile { + col: Some(2), + q: 0.5, + accuracy: AccuracyTarget::exact(), + }], + output_names: vec![], + filters: vec![Some(Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(3)), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(1.0))), + semantics: crate::ir::ExprSemantics::Sql, + }))], + having: None, + child: scan(), + }); + let (ra, rb) = two_roots(unfiltered, filtered); + assert!( + !Rc::ptr_eq(&ra, &rb), + "a filtered measure must not be shared" + ); + assert_ne!(ra, rb); + } + #[test] fn no_unique_keys_means_no_merge_even_when_structurally_identical() { let a = quantile_agg(vec![], Some(2), 0.9); diff --git a/crates/types/src/ir/schema_support.rs b/crates/types/src/ir/schema_support.rs index c3ec9931..8717c828 100644 --- a/crates/types/src/ir/schema_support.rs +++ b/crates/types/src/ir/schema_support.rs @@ -100,3 +100,21 @@ pub fn with_promql_series_identity( } visit(root, &mut HashMap::new()) } + +#[cfg(test)] +mod tests { + use super::with_promql_series_identity; + use crate::ir::{NonASAPOp, Operator, OperatorNode, ScalarExpr}; + use crate::pre_asap::ScalarValue; + + // A direct scalar literal (`vector(1)`) stays a valid vector input when + // series identity is realized. + #[test] + fn series_identity_accepts_direct_vector_literal() { + let root = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::PromqlVectorFromScalar( + ScalarExpr::Literal(ScalarValue::Float64(1.0)), + ))) + .unwrap(); + assert!(with_promql_series_identity(&root).is_ok()); + } +} From b16ded2b9c616b2d6b19a1ea3adad8bd9bb06346 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:12:03 +0000 Subject: [PATCH 6/7] docs: fix stale comments found in accumulated reviews Read-out wording becomes evaluation; AbsoluteValue/RelativeValue become absolute/relative errors; the HLL merge's idempotence is stated next to KMV's (and the stale missing-KMV-kernel note dropped); the misplaced compile_node doc moves back above it; compile_exact_evaluation states its offset-0 counter window; Example 1's 27 is 3 x 9; parsed_workload names the query-language frontends and Stage 1; the factory updater's DAG-state claim is limited to the heap families; a SQL fold test no longer claims its target is unmet; "dag" becomes "DAG". --- crates/asap-aware-mapping/src/accuracy/composition.rs | 8 ++++---- .../src/accuracy/estimators/univmon.rs | 2 +- .../src/physical_planner/mod.rs | 6 +++--- .../src/physical_planner/promql_values.rs | 3 +++ .../src/summary_kernels/factory.rs | 11 ++++++----- crates/devtools/tests/stage_pipeline.rs | 4 ++-- crates/frontend-promql/tests/promql_lowering.rs | 2 +- .../tests/per_computation_candidates_sql.rs | 6 +++--- crates/types/src/ir/asap.rs | 2 +- crates/types/src/ir/scalar.rs | 2 +- crates/types/src/parsed_workload.rs | 4 ++-- crates/types/src/post_asap/family_capability.rs | 7 +++++-- crates/types/src/post_asap/query_time/mod.rs | 2 +- crates/types/src/post_asap/sketch.rs | 2 +- crates/types/src/pre_asap/schema.rs | 2 +- 15 files changed, 35 insertions(+), 28 deletions(-) diff --git a/crates/asap-aware-mapping/src/accuracy/composition.rs b/crates/asap-aware-mapping/src/accuracy/composition.rs index 17f2796f..5871cb1f 100644 --- a/crates/asap-aware-mapping/src/accuracy/composition.rs +++ b/crates/asap-aware-mapping/src/accuracy/composition.rs @@ -263,7 +263,7 @@ impl DefaultAccuracyModel { operator: op.clone(), input_metrics: inputs.iter().map(|g| g.spec.accuracy_metric).collect(), local_metric: None, - reason: "exact average requires AbsoluteValue input guarantees".into(), + reason: "exact average requires absolute-error input guarantees".into(), }); } let mut provenance = Vec::new(); @@ -315,7 +315,7 @@ impl DefaultAccuracyModel { operator: op.clone(), input_metrics: inputs.iter().map(|g| g.spec.accuracy_metric).collect(), local_metric: None, - reason: "exact max/min requires AbsoluteValue input guarantees".into(), + reason: "exact max/min requires absolute-error input guarantees".into(), }); } let mut provenance = Vec::new(); @@ -869,12 +869,12 @@ pub(super) fn propagate( } if inputs.len() != 1 || !same_metric(AccuracyMetric::ABSOLUTE) { return Err(unsupported( - "Lipschitz rule is registered for exactly one AbsoluteValue input".into(), + "Lipschitz rule is registered for exactly one absolute-error input".into(), )); } if local.is_some_and(|g| g.spec.accuracy_metric != AccuracyMetric::ABSOLUTE) { return Err(unsupported( - "Lipschitz rule needs an AbsoluteValue local guarantee".into(), + "Lipschitz rule needs an absolute-error local guarantee".into(), )); } Ok(DefaultAccuracyModel::lipschitz( diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs index 8ac8ff31..13c5dcb1 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs @@ -415,7 +415,7 @@ mod tests { (0.2, 0.2), ]; - /// Each L2 shape meets its target as `RelativeValue`, and one fewer + /// Each L2 shape meets its target as a relative error, and one fewer /// row or column does not. #[test] fn l2_shape_is_the_smallest_meeting_its_target() { diff --git a/crates/asap-physical-operators/src/physical_planner/mod.rs b/crates/asap-physical-operators/src/physical_planner/mod.rs index c6ef2195..9501007b 100644 --- a/crates/asap-physical-operators/src/physical_planner/mod.rs +++ b/crates/asap-physical-operators/src/physical_planner/mod.rs @@ -519,7 +519,7 @@ fn compile_internal( } } if let Payload::ASAP(ASAPOp::FinalizeExactAccumulator { .. }) = &node.payload { - // Exact counts read out as Int64; PromQL declares a Float64 sample. + // Exact counts evaluate to Int64; PromQL declares a Float64 sample. let evaluation = bind_operation(node, &schemas) .map_err(|error| invalid(format!("node {id}: {error}")))?; let actual = evaluation.schema(); @@ -678,8 +678,6 @@ fn temporal_evaluation_drops_name(node: &PhysicalASAPDAGNode) -> bool { ) } -/// Bind a Planner node against the schemas supplied by its deployment edges. -/// This is the same checked path used by complete DAG binding. /// The IR accepts a `SummaryMerge` over any state family; the executor runs /// one only when the family-capability table gives the family a sound merge. pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error> { @@ -692,6 +690,8 @@ pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error> { Ok(()) } +/// Bind a Planner node against the schemas supplied by its deployment edges. +/// This is the same checked path used by complete DAG binding. pub fn compile_node(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result { for schema in inputs { crate::values::validate_schema(schema)?; diff --git a/crates/asap-physical-operators/src/physical_planner/promql_values.rs b/crates/asap-physical-operators/src/physical_planner/promql_values.rs index a7c102d1..b5522116 100644 --- a/crates/asap-physical-operators/src/physical_planner/promql_values.rs +++ b/crates/asap-physical-operators/src/physical_planner/promql_values.rs @@ -221,6 +221,9 @@ pub fn exact_state_schema(family: SummaryFamilyType) -> Result } /// Retain exact evaluation semantics before any deployment state is opened. +/// A counter (`Rate`, `Increase`) is evaluated over the `lookback_ms` +/// window ending at the evaluation time: this entry point takes no +/// offset, so its counter window has offset 0. pub fn compile_exact_evaluation( family: SummaryFamilyType, lookback_ms: u64, diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index 2827a777..4e6c06c1 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/src/summary_kernels/factory.rs @@ -245,11 +245,12 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { // --------------------------------------------------------------------------- /// Keyed frequency updater backed by sketchlib's per-row-seeded weighted -/// frequency kernel, the state the DAG's keyed summary build produces: plain -/// CountSketch (no candidate heap), Count-Min with a top-k heap, or -/// CountSketch with a top-k heap. Each row hashes with its own seed, so rows -/// stay independent at any depth. Each label value of the key is one `Utf8` -/// identity, as in the DAG's keyed build. +/// frequency kernel: plain CountSketch (no candidate heap), Count-Min with a +/// top-k heap, or CountSketch with a top-k heap. For the two heap families +/// this is the state the DAG's keyed summary build produces; plain +/// CountSketch uses the same kernel without a heap. Each row hashes with +/// its own seed, so rows stay independent at any depth. Each label value of +/// the key is one `Utf8` identity, as in the DAG's keyed build. /// /// Each raw Prometheus sample contributes its `value` as a Float64 weight, /// kept exactly (no rounding). CountSketch accepts signed weights; Count-Min diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 65f5be04..553d5708 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -152,8 +152,8 @@ fn display_cap_is_recorded_and_keeps_the_selection() { let logical = &document["stage1_logical_asap_optimization"]; assert_eq!(logical["capped"], true); assert!(logical["candidates"].as_array().unwrap().len() < 27); - // 3 Pass 1 candidates, each with 8 Pass 2 candidates: shared inputs, - // tumbling panes, and both. + // 3 Pass 1 candidates, each kept as it is and with 8 Pass 2 candidates + // (shared inputs, tumbling panes, and both): 3 × 9. assert_eq!(logical["combinations"], 27); assert_eq!( document["stage3_plan_selection"]["selected"], diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index b23fb174..0b15e521 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -928,7 +928,7 @@ fn collect_intents(e: &OperatorNode, out: &mut Vec) { } } -/// True if any `AggIntent` anywhere in the dag satisfies `pred`. +/// True if any `AggIntent` anywhere in the DAG satisfies `pred`. fn has_intent bool>(e: &OperatorNode, pred: F) -> bool { all_intents(e).iter().any(pred) } diff --git a/crates/integration-tests/tests/per_computation_candidates_sql.rs b/crates/integration-tests/tests/per_computation_candidates_sql.rs index 5742bcd3..5c1e0823 100644 --- a/crates/integration-tests/tests/per_computation_candidates_sql.rs +++ b/crates/integration-tests/tests/per_computation_candidates_sql.rs @@ -151,9 +151,9 @@ fn reads_ddsketch(node: &std::rc::Rc) -> bool { } /// A SQL fold over the groups of an approximate subquery counts rows the -/// series count of its source does not bound, so -/// it leaves its own row-count statistic, which plan selection does not -/// bind: the target stays unmet. +/// series count of its source does not bound, so its δ names its own +/// row-count statistic (`group_row_count`), not the source's +/// `input_row_count`. #[tokio::test] async fn a_sql_fold_over_estimates_leaves_its_row_count_unbound() { let catalog = SqlCatalog::new().with_table( diff --git a/crates/types/src/ir/asap.rs b/crates/types/src/ir/asap.rs index 67b5b221..7ddd2ff1 100644 --- a/crates/types/src/ir/asap.rs +++ b/crates/types/src/ir/asap.rs @@ -33,7 +33,7 @@ pub enum ASAPOp> { #[serde(default)] filter: Option>, }, - /// Read out a query result from built summary state. Output is a + /// Evaluate a query result from built summary state. Output is a /// row-shaped schema. SummaryEstimate { summary_input: C, diff --git a/crates/types/src/ir/scalar.rs b/crates/types/src/ir/scalar.rs index 2315cde9..eec779f4 100644 --- a/crates/types/src/ir/scalar.rs +++ b/crates/types/src/ir/scalar.rs @@ -358,7 +358,7 @@ impl ScalarExpr { /// Infer the `(DataType, nullable)` this expression produces against the /// input schema its owner evaluates it in. Unregistered functions are /// rejected. A reference to a field carrying summary state is an error: - /// state must be read out before an expression can use it. + /// state must be evaluated before an expression can use it. pub fn scalar_type(&self, schema: &Schema) -> Result<(DataType, bool), SchemaDerivationError> { Ok(match self { ScalarExpr::CurrentTimestamp => (DataType::Timestamp, false), diff --git a/crates/types/src/parsed_workload.rs b/crates/types/src/parsed_workload.rs index bed0a18a..81218a9e 100644 --- a/crates/types/src/parsed_workload.rs +++ b/crates/types/src/parsed_workload.rs @@ -1,8 +1,8 @@ //! [`ParsedWorkload`] — a [`PlanningWorkload`] whose queries have been lowered //! to the operator IR. //! -//! This is the boundary between the frontend stage and the optimization stage -//! (issues #429, #430). Everything downstream of lowering consumes this type +//! This is the boundary between the query-language frontends and Stage 1, +//! logical ASAP-aware optimization (issues #429, #430). Everything downstream of lowering consumes this type //! and never sees query text, a catalog, or a query language; everything the //! optimizer still needs about *demand* — recurrence, predictability, //! execution time, accuracy requirement — is read off the retained diff --git a/crates/types/src/post_asap/family_capability.rs b/crates/types/src/post_asap/family_capability.rs index fec977db..2309367c 100644 --- a/crates/types/src/post_asap/family_capability.rs +++ b/crates/types/src/post_asap/family_capability.rs @@ -203,10 +203,13 @@ impl FamilyCapability { (FrequencyEntropy, Approximate), ], ), + // HLL's merge keeps each register's maximum, which is + // idempotent: an item both inputs saw counts once, so + // overlapping inputs merge to the state of their union. // `asap_sketchlib`'s KMV hashes every item with one // shared seed (`CANONICAL_HASH_SEED`), and `KMV::merge` - // keeps the k smallest distinct hashes: the state of - // the union. The executor has no KMV kernel yet (#697). + // keeps the k smallest distinct hashes: again the state + // of the union. S::Hll | S::Kmv => (Some(MayOverlap), &[(Distinct, Union)]), // Theta is not offered, and has no merge. S::Theta => (None, &[(Distinct, Approximate)]), diff --git a/crates/types/src/post_asap/query_time/mod.rs b/crates/types/src/post_asap/query_time/mod.rs index 511e72f8..ecb89935 100644 --- a/crates/types/src/post_asap/query_time/mod.rs +++ b/crates/types/src/post_asap/query_time/mod.rs @@ -15,7 +15,7 @@ //! itself, not just in prose: it holds pure, sketch-object-agnostic *math* //! (given a sketch's real counter values, compute a tighter posterior //! error bound) that only makes sense to invoke *after* a real sketch has -//! ingested data and is being read out — a downstream runtime's job, not +//! ingested data and is being evaluated — a downstream runtime's job, not //! this crate's. Nothing in `asap-types` or `asap-aware-mapping` calls //! into this module today; it is a library waiting for a runtime that //! doesn't exist yet in this workspace (see [`error_estimation`]'s own diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index 1a1edd96..a6005f73 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -654,7 +654,7 @@ impl Default for GroupingStrategy { } } -// ── Sketch read-out queries ─────────────────────────────────────────────────── +// ── Sketch evaluation queries ───────────────────────────────────────────────── /// One summary-state update. `item` identifies a keyed item when the family /// is keyed; `weight` is the observation/update expression applied to state. diff --git a/crates/types/src/pre_asap/schema.rs b/crates/types/src/pre_asap/schema.rs index 7eec788b..16b1cb17 100644 --- a/crates/types/src/pre_asap/schema.rs +++ b/crates/types/src/pre_asap/schema.rs @@ -132,7 +132,7 @@ pub enum FieldDataType { /// Exact, mergeable accumulator state (`Sum`/`Count`/`Min`/`Max`/`Rate`/ /// `Increase`). Value consumers require an explicit finalization boundary. ExactAggregate(ExactKind, ExactParams), - /// Approximate sketch state (KLL/CMS/HLL/…), read out via a + /// Approximate sketch state (KLL/CMS/HLL/…), evaluated by a /// `SummaryEstimate`. A [`SketchKind`] already carries the concrete /// algorithm, params, and grouping layout committed to, not just its /// category — a bound node needs to know it's specifically independent From e8b84afa5186053a7dbce0351ea3e1abc30f15ac Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:14:02 +0000 Subject: [PATCH 7/7] refactor: rename the Stage 3 test's pane rule candidates --- .../src/stage3_plan_selection.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index 5949f898..b2e3a9c8 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -5433,13 +5433,15 @@ mod tests { .into_iter() .map(|(_, root)| root) .collect(); - let sites = crate::asap_aware_cse::pane_candidates(&roots, &[Some(u64::from(MINUTE))]); - let roots = crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &sites) - .unwrap() - .pop() - .unwrap() - .roots - .unwrap(); + let rule_candidates = + crate::asap_aware_cse::pane_candidates(&roots, &[Some(u64::from(MINUTE))]); + let roots = + crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &rule_candidates) + .unwrap() + .pop() + .unwrap() + .roots + .unwrap(); let mut candidates = optimize_physical_asap( "L1", &roots,