feat(emit): thread workload.group_by_labels into streaming-config labels.grouping - #245
Merged
Merged
Conversation
…els.grouping Follow-up to #244. The typed L5's `BackendAggregation` now carries a `grouping: Vec<String>` 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<ColumnId>` — 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<String>` 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<String>` 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) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 15, 2026
…g plumb (#247) First end-to-end test of the gateway-less control-plane ↔ data-plane contract. Lives in `data_plane/tests/` since data_plane already owns the OTLP/precompute/HTTP wiring; control_plane is already a workspace dep so no Cargo.toml changes were needed. ## What it tests Two `#[tokio::test]`s: 1. `controller_plans_ddsketch_quantile_and_backend_parses_streaming_config` — workload `{ metric: http_latency_ms, agg: Quantile, sla: 0.01, window: 60s }` flows through `bind_workload_typed` → `split_typed_three_stage` → `emit_backend_streaming_config_json`, the JSON is POSTed to the in-process backend's `/api/v1/streaming-config`, and the GET endpoint reflects the parsed aggregation. Asserts the emitted JSON's shape (#244 content fields present, #244/#246 `aggregationId` absent). 2. `controller_plans_with_grouping_and_backend_parses_grouping_labels` — same shape but with `group_by_labels: ["zone"]`. Asserts the emitter surfaces `labels.grouping: ["zone"]` (#245), the POST succeeds, and the active-config snapshot contains `"zone"` in the parsed `AggregationConfig.grouping_labels`. ## Findings the test caught `bind_workload_typed`-produced PhysicalExpr doesn't surface `metric_name` or `window_secs` to `BackendAggregation` via `extract_edge_facts`. The underlying cause is likely that the allocator paints the `Logical(Scan{...})` chain at a stage other than Edge in some binder outputs, so the walk's `(Logical, Edge)` match arm doesn't fire — Step γ open-set label work (also blocking grouping fidelity) probably needs a coordinated fix. Pragmatic workaround mirroring the #245 grouping patch: control_plane/src/main.rs (handle_plan, Backend stage match arm) for agg in &mut be.aggregations { if agg.metric_name.is_empty() { agg.metric_name = workload.metric_name.clone(); } if agg.window_secs == 0 { agg.window_secs = workload.time_window.as_secs(); } agg.grouping = workload.group_by_labels.clone(); } This is belt-and-braces — if the L5 walk DID surface the field, the conditional `if … is_empty()` / `== 0` checks preserve it; otherwise the workload spec wins. Closes a real bug where fresh-deploy `handle_plan` POSTs would have failed the backend parser on `Missing metric` / `Missing windowSize` (PR #244 caught the JSON shape gap; this test caught the value-population gap). ## Other change `HttpServer::start_test_server` had `#[cfg(test)]` gating it to the lib's own unit tests — integration tests under `data_plane/tests/` are compiled separately and couldn't see it. Dropped the gate; the method's name + doc-comment make the test-only intent explicit, and production code uses `start()` regardless. ## Test plan - [x] `cargo test --test e2e_controller_plans_and_backend_serves`: **2 passed**. - [x] `cargo test --lib -p control_plane`: **687 passed**. - [x] `cargo test --tests --bins -p control_plane`: **27 passed**. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Follow-up to #244. The typed L5's
BackendAggregationnow carries agrouping: Vec<String>field, surfaced underlabels.groupingin 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.) behave correctly server-side.Plumbing rationale
The L3
QueryExpr::Aggregate.byisVec<ColumnId>— positional indexes into a synthesizedSchemathat intentionally does not track open-set labels (see the module doc onintent_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<String>carries the names unambiguously, andhandle_planalready has the workload in scope for both binder paths (query_string →bind_query_exprand explicit-field →bind_workload_typed). The pragmatic plumb:grouping: Vec<String>toBackendAggregation(#[serde(default)]keeps existing serialised fixtures parsing).grouping: vec![]at both construction sites (no behaviour change at emit time).handle_plan, aftersplit_typed_three_stage, patches everyBackendAggregation.groupingwithworkload.group_by_labels.clone()before posting. Every aggregation under one workload shares the same grouping today, so the patch is uniform.build_backend_aggregation_jsonreadsagg.groupingintolabels.grouping(was hard-coded empty before this PR).labels.rollup/labels.aggregatedstay empty — the controller doesn't surface either today.Test plan
cargo checkcleancargo test --lib: 687 passed; 0 failed (+1:backend_json_emits_grouping_under_labels)cargo test --tests --bins: 27 passed; 0 failed🤖 Generated with Claude Code