From 5013dfae58edbfd37595bba0c660299b38ccf43e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 15 May 2026 14:29:17 -0600 Subject: [PATCH] feat(emit): thread workload.group_by_labels into streaming-config labels.grouping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to #244. The typed L5's `BackendAggregation` now carries a `grouping: Vec` field, surfaced under `labels.grouping` in the streaming-config JSON the controller posts to asapquery-backend. The backend's precompute-engine accumulator pipeline keys its per-aggregation state by the projected attribute set, so this is what makes cross-host fan-in merges (sum by zone / etc.) actually behave correctly on the backend side. ## Plumbing rationale The L3 `QueryExpr::Aggregate.by` is `Vec` — positional indexes into a synthesized `Schema` that intentionally does not track open-set labels (see the module doc on `intent_algebra::column_resolution` — label-set resolution is a Step γ TODO). So reverse-resolving ColumnId → label name at L5 emit time isn't tractable. `QueryWorkload.group_by_labels: Vec` carries the names unambiguously, and `handle_plan` already has the workload in scope at the call site for both binder paths (query_string → `bind_query_expr` and explicit-field → `bind_workload_typed`). The pragmatic plumb: 1. Add `grouping: Vec` to `BackendAggregation` (default empty; `#[serde(default)]` keeps existing serialised fixtures parsing). 2. Have the L5 emitter populate `grouping: vec![]` at both construction sites (no behaviour change at emit time). 3. In `handle_plan`, after `split_typed_three_stage` returns the per-stage map, patch every `BackendAggregation.grouping` with `workload.group_by_labels.clone()` before `emit_backend_streaming_config_json`. Every aggregation under a single workload shares the same grouping today, so the patch is uniform. 4. `build_backend_aggregation_json` reads `agg.grouping` into `labels.grouping` (was hard-coded empty before this PR). `labels.rollup` and `labels.aggregated` stay empty — the controller doesn't surface either today. ## Test plan - New: `backend_json_emits_grouping_under_labels` exercises the JSON round-trip of the new field directly. 687 lib tests (+1) + 27 binary tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/stage_config.rs | 52 ++++++++++++++++++- control_plane/src/emit/trait_def.rs | 1 + control_plane/src/main.rs | 20 ++++++- .../src/physical/colored_dag/emitter.rs | 19 +++++++ 4 files changed, 90 insertions(+), 2 deletions(-) diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 48df77436..a5925e486 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -1395,7 +1395,7 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { "aggregationSubType": "", "metric": agg.metric_name, "labels": { - "grouping": Vec::::new(), + "grouping": agg.grouping, "rollup": Vec::::new(), "aggregated": Vec::::new(), }, @@ -1744,6 +1744,7 @@ mod tests { sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }, BackendAggregation { @@ -1753,6 +1754,7 @@ mod tests { sketch_params: SketchParams::Hll(HllParams { precision: 14 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }, ], @@ -1808,6 +1810,7 @@ mod tests { }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }, BackendAggregation { @@ -1817,6 +1820,7 @@ mod tests { sketch_params: SketchParams::Cms(CmsParams { w: 4096, d: 4 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }, ], @@ -1880,6 +1884,7 @@ mod tests { sketch_params: params, window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![BackendReadout { @@ -2225,6 +2230,48 @@ mod tests { } } + /// Grouping labels surface under `labels.grouping` in the emitted + /// JSON. The L5 emitter itself leaves the list empty; `handle_plan` + /// patches it from `workload.group_by_labels` before posting, so + /// here we simulate that by setting `grouping` on the + /// `BackendAggregation` directly and assert the JSON round-trips. + #[test] + fn backend_json_emits_grouping_under_labels() { + let cfg = BackendStageConfig { + aggregations: vec![BackendAggregation { + aggregation_id: "agg0".into(), + metric_name: "http_latency_ms".into(), + sketch_kind: SketchKind::DDSketch, + sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 30, + spatial_filter: String::new(), + grouping: vec!["zone".into(), "service".into()], + aggregation_input: AggregationInput::SketchEnvelope, + }], + readouts: vec![], + }; + let v = emit_backend_streaming_config_json(&cfg).expect("emit ok"); + let grouping = v["aggregations"][0]["labels"]["grouping"] + .as_array() + .expect("grouping array"); + let names: Vec<&str> = grouping.iter().filter_map(|s| s.as_str()).collect(); + assert_eq!( + names, + vec!["zone", "service"], + "labels.grouping must surface BackendAggregation.grouping verbatim\n{v}" + ); + // The other label lists stay empty — the L5 controller doesn't + // yet emit rollup / aggregated. + assert!(v["aggregations"][0]["labels"]["rollup"] + .as_array() + .unwrap() + .is_empty()); + assert!(v["aggregations"][0]["labels"]["aggregated"] + .as_array() + .unwrap() + .is_empty()); + } + /// Snapshot: aggregations + readouts together exhibit the /// id-aliasing the backend uses to wire readouts back to their /// producing aggregation. Pins the sort order + key names. Phase β @@ -2242,6 +2289,7 @@ mod tests { sketch_params: SketchParams::Kll(KllParams { k: 200 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![BackendReadout { @@ -2284,6 +2332,7 @@ mod tests { sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![], @@ -2306,6 +2355,7 @@ mod tests { sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: AggregationInput::Raw, }], readouts: vec![], diff --git a/control_plane/src/emit/trait_def.rs b/control_plane/src/emit/trait_def.rs index 9cd29a911..d6ba202f8 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -223,6 +223,7 @@ mod tests { sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), window_secs: 60, spatial_filter: String::new(), + grouping: Vec::new(), aggregation_input: crate::physical::colored_dag::emitter::AggregationInput::SketchEnvelope, }], diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 838fc70b1..b4a08fe5f 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -558,7 +558,25 @@ async fn handle_plan( Err(e) => warn!(error = %e, "emit_gateway_yaml failed"), } } - crate::physical::colored_dag::StageConfig::Backend(be) => { + crate::physical::colored_dag::StageConfig::Backend(mut be) => { + // Patch grouping label names from the + // workload spec. The typed L5 emitter + // produces `BackendAggregation` with + // `grouping: vec![]` because the canonical + // L3 `QueryExpr::Aggregate.by` is + // positional `ColumnId`s against a + // synthesized schema that has no label + // columns (open-set label naming is a + // Step γ TODO in + // `intent_algebra::column_resolution`). + // `QueryWorkload.group_by_labels` carries + // the names unambiguously, so we patch + // them in here — every aggregation under + // the same workload shares the same + // grouping today. + for agg in &mut be.aggregations { + agg.grouping = workload.group_by_labels.clone(); + } // Phase C: post the typed L5 streaming-config // JSON to ASAPQuery-backend via the shared // BackendClient when configured. Without a diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index e08c2dc97..32087f71d 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -311,6 +311,20 @@ pub struct BackendAggregation { /// `label_filters`). Empty string when no filter applies. #[serde(default)] pub spatial_filter: String, + /// Group-by label names — keys in `labels.grouping` on the backend + /// side, where the precompute engine's accumulator pipeline keys + /// its per-aggregation state by the projected attribute set. + /// + /// The L5 emitter populates this empty (`vec![]`); the caller + /// (`handle_plan`) patches it from `workload.group_by_labels` + /// before posting the streaming-config JSON. The canonical L3 + /// `QueryExpr::Aggregate.by` carries the keys as positional + /// `ColumnId`s against a synthesized schema that has no label + /// columns (open-set label naming is a Step γ TODO in + /// `intent_algebra::column_resolution`), so the workload-spec + /// strings are the only reliable source of the names today. + #[serde(default)] + pub grouping: Vec, /// Phase ε.1 — what shape the backend ingests for this /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the /// default (the wire payload is a sketch state already). Mode 2 @@ -467,6 +481,10 @@ impl Emitter for ThreeStageEmitter { sketch_params: params.clone(), window_secs: edge.window_secs.unwrap_or(0), spatial_filter: spatial_filter_from_label_filters(&edge.label_filters), + // Populated post-emit by the caller (handle_plan) + // from workload.group_by_labels — see the + // struct doc-comment for the rationale. + grouping: Vec::new(), // Mode 1 — sketch built at edge, ships envelope. aggregation_input: AggregationInput::SketchEnvelope, }); @@ -544,6 +562,7 @@ impl Emitter for ThreeStageEmitter { sketch_params: params.clone(), window_secs: edge.window_secs.unwrap_or(0), spatial_filter: spatial_filter_from_label_filters(&edge.label_filters), + grouping: Vec::new(), // Mode 2 — backend builds sketch from raw OTLP. aggregation_input: AggregationInput::Raw, });