From 466002b59cb5efc13d1cbb3439f510dcc38fb8b1 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 15 May 2026 15:37:23 -0600 Subject: [PATCH] fix(emit): propagate metric_name/window_secs through L5 walk (DAG visit order) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #247's belt-and-braces patches (`handle_plan` patching `BackendAggregation.metric_name` / `window_secs` / `grouping` from the workload spec) were load-bearing for `metric_name` and `window_secs`, not no-ops. Root cause was a DAG-visit-order bug in `ThreeStageEmitter::emit_per_stage`: * The colored DAG's node order is depth-first parent-first (`SketchEstimate`, `SketchAgg`, `Logical`) so a SketchAgg appears BEFORE its `Logical(Window{Scan})` child in `dag.nodes`. * Pass 2's SketchAgg arm captures `edge.source_metric` / `edge.window_secs` / `edge.label_filters` *at push time* while building `BackendAggregation`. * But those fields are populated by Pass 2's Logical arm calling `extract_edge_facts(qe, &mut edge)` — which fires LATER in the same loop, after SketchAgg. * Result: every `BackendAggregation` shipped with empty `metric_name` / `window_secs: 0`, which the backend's `AggregationConfig::from_yaml_data` rejects on `Missing metric` / `Missing windowSize`. The Replanner's separate legacy emitter (`generate_streaming_config_yaml`) papered over it by reading directly from `plan.agent_config`; `handle_plan` was papering over it with the workload-spec patch from #247. Fix: add **Pass 0** that pre-walks the DAG looking only for `Logical(qe) @ Edge` nodes and calls `extract_edge_facts` to populate edge facts. Pass 2's Logical arm stays (it's idempotent — only sets on `None` and dedupes label_filters), so the change is safe even if node order changes upstream. Grouping stays empty out of the L5 walk: canonical L3 `QueryExpr::Aggregate.by` is `Vec` against a synthesized schema with no label columns (Step γ TODO in `intent_algebra::column_resolution`). The workload-spec patch in `handle_plan` remains the source of truth for grouping today. ## Diagnostic tests added `physical::stage_split::l5_walk_propagation_tests` — three characterisation tests that pin the L5 walk's behaviour for `bind_workload_typed` output: 1. `l5_walk_surfaces_metric_name_for_bind_workload_typed_output` 2. `l5_walk_surfaces_window_secs_for_bind_workload_typed_output` 3. `l5_walk_leaves_grouping_empty_pending_step_gamma` The first two now pass (the fix). The third pins the current limitation so a future Step γ fix fails it loudly and signals that the handle_plan grouping patch can be retired alongside. ## Tests - `cargo test --lib -p control_plane`: 690 passed (was 687; +3 new diagnostic tests). - `cargo test --tests --bins -p control_plane`: 27 passed. - `cargo test --test e2e_controller_plans_and_backend_serves`: 3 passed (the e2e suite's metric_name / windowSize assertions now pass via the real fix, not the patch). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/physical/colored_dag/emitter.rs | 25 ++++ control_plane/src/physical/stage_split.rs | 132 ++++++++++++++++++ 2 files changed, 157 insertions(+) diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 32087f71..6bfb30c3 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -434,6 +434,31 @@ impl Emitter for ThreeStageEmitter { // through. let mut sketch_agg_ids: HashMap = HashMap::new(); + // Pass 0 — populate edge facts (source_metric, window_secs, + // label_filters) from every `Logical(qe) @ Edge` node before + // anything else reads them. Necessary because the DAG's + // depth-first node order puts SketchAgg BEFORE its + // `Logical(Window{Scan})` child, but the SketchAgg arm of + // Pass 2 captures `edge.source_metric` / `edge.window_secs` / + // `edge.label_filters` *at push time* when building + // `BackendAggregation` (and `EdgeSketchProcessor`'s + // `spatial_filter` etc.). Without this pre-pass, those fields + // see `None` because the child Logical hasn't been visited + // yet — and the resulting `BackendAggregation` ships with an + // empty `metric_name` / `window_secs: 0` / `spatial_filter: + // ""`, which the backend's `AggregationConfig::from_yaml_data` + // rejects on `Missing metric` / `Missing windowSize`. + // + // `extract_edge_facts` is idempotent on `source_metric` and + // `window_secs` (only sets if `None`) and dedupes + // `label_filters` — so the Pass 2 arm that also calls it + // remains correct (no double-counting). + for node in &dag.nodes { + if let (PhysicalExpr::Logical(qe), StageId::Edge) = (&node.expr, node.stage) { + extract_edge_facts(qe, &mut edge); + } + } + // Pass 1 — assign deterministic aggregation_ids to every // SketchAgg up-front so SketchMerge / SketchEstimate emission // (pass 2) can resolve them regardless of node-table order. diff --git a/control_plane/src/physical/stage_split.rs b/control_plane/src/physical/stage_split.rs index 3d128f88..0c9ae71a 100644 --- a/control_plane/src/physical/stage_split.rs +++ b/control_plane/src/physical/stage_split.rs @@ -53,3 +53,135 @@ pub fn split_typed_three_stage( let dag = StageAllocator.allocate(expr, Topology::ThreeStage).ok()?; ThreeStageEmitter.emit_per_stage(&dag).ok() } + +#[cfg(test)] +mod l5_walk_propagation_tests { + //! Characterisation tests for the L5 walk's edge-fact extraction. + //! + //! Established by PR #247: `handle_plan`'s `Backend` stage arm + //! belt-and-braces patches `metric_name`, `window_secs`, and + //! `grouping` on every emitted `BackendAggregation` from the + //! workload spec, on the suspicion that the L5 walk's + //! `extract_edge_facts` doesn't propagate these fields cleanly + //! through every binder's output shape. + //! + //! These tests **measure** what the L5 walk actually produces for + //! the canonical `bind_workload_typed` output — so we know whether + //! the patches are dead weight (the walk works → fields already + //! populated → patches are no-ops) or load-bearing (walk doesn't + //! propagate → patches are the real source of the field values). + //! + //! Result documented in the test assertions: the walk **does** + //! surface `metric_name` and `window_secs` for + //! `bind_workload_typed` output. Grouping stays empty because the + //! canonical L3 `QueryExpr::Aggregate.by` is a `Vec` + //! against a synthesized schema that has no label columns (Step γ + //! TODO in `intent_algebra::column_resolution`). + + use crate::physical::colored_dag::StageConfig; + use crate::types::{AggType, QueryWorkload}; + use std::collections::HashMap; + use std::time::Duration; + + fn workload(metric: &str, group_by: Vec, window: Duration) -> QueryWorkload { + QueryWorkload { + metric_name: metric.to_string(), + label_filters: HashMap::new(), + group_by_labels: group_by, + aggregations: vec![AggType::Quantile], + time_window: window, + repeat_every: None, + accuracy_sla: 0.01, + latency_sla: None, + sketch_type_override: None, + exact_required: false, + quantiles: vec![0.99], + } + } + + #[test] + fn l5_walk_surfaces_metric_name_for_bind_workload_typed_output() { + let w = workload("http_latency_ms", Vec::new(), Duration::from_secs(60)); + let physical_expr = + crate::optimizer::rules::bind_workload_typed(&w).expect("bind produced expr"); + let configs = super::split_typed_three_stage(&physical_expr).expect("split ok"); + let backend_cfg = configs + .into_values() + .find_map(|cfg| match cfg { + StageConfig::Backend(be) => Some(be), + _ => None, + }) + .expect("Backend stage produced"); + let agg = backend_cfg + .aggregations + .first() + .expect("at least one aggregation"); + assert_eq!( + agg.metric_name, "http_latency_ms", + "the L5 walk's `extract_edge_facts` must thread the Scan's \ + metric name through to BackendAggregation.metric_name" + ); + } + + #[test] + fn l5_walk_surfaces_window_secs_for_bind_workload_typed_output() { + let w = workload("http_latency_ms", Vec::new(), Duration::from_secs(120)); + let physical_expr = + crate::optimizer::rules::bind_workload_typed(&w).expect("bind produced expr"); + let configs = super::split_typed_three_stage(&physical_expr).expect("split ok"); + let backend_cfg = configs + .into_values() + .find_map(|cfg| match cfg { + StageConfig::Backend(be) => Some(be), + _ => None, + }) + .expect("Backend stage produced"); + let agg = backend_cfg + .aggregations + .first() + .expect("at least one aggregation"); + assert_eq!( + agg.window_secs, 120, + "the L5 walk's `extract_edge_facts` must thread Window.size \ + through to BackendAggregation.window_secs" + ); + } + + #[test] + fn l5_walk_leaves_grouping_empty_pending_step_gamma() { + // L3 `QueryExpr::Aggregate.by` is positional `ColumnId`s against + // a synthesized schema that has no label columns — so the walk + // CANNOT recover the original label names. handle_plan patches + // grouping from `workload.group_by_labels` for this reason. + // This test pins the current behaviour so a future Step γ fix + // (proper open-set label resolution) will fail it loudly and + // remind whoever's making the change to also retire the patch. + let w = workload( + "http_latency_ms", + vec!["zone".to_string()], + Duration::from_secs(60), + ); + let physical_expr = + crate::optimizer::rules::bind_workload_typed(&w).expect("bind produced expr"); + let configs = super::split_typed_three_stage(&physical_expr).expect("split ok"); + let backend_cfg = configs + .into_values() + .find_map(|cfg| match cfg { + StageConfig::Backend(be) => Some(be), + _ => None, + }) + .expect("Backend stage produced"); + let agg = backend_cfg + .aggregations + .first() + .expect("at least one aggregation"); + assert!( + agg.grouping.is_empty(), + "L5 walk cannot recover label names from canonical L3 \ + ColumnIds (Step γ TODO in column_resolution); \ + BackendAggregation.grouping must come from the workload \ + patch in handle_plan — got {:?}", + agg.grouping + ); + } +}