diff --git a/Cargo.lock b/Cargo.lock index 5f1fb4b7..ab159a2f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -980,6 +980,7 @@ dependencies = [ "anyhow", "arc-swap", "arrow", + "asap-aware-mapping", "asap-precompute-rs", "asap-types", "asap_otel_proto", diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 787d43b5..ecaefb29 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -54,6 +54,7 @@ enum NodeKind<'a> { SketchAgg { sketch_type: &'a SketchAlgorithm, params: &'a SketchParams, + input: &'a planner_types::post_asap::SummaryUpdate, }, /// An exact accumulator — the old `PhysicalExpr::ExactAgg`. This /// emitter has never had a match arm for it (falls through to the @@ -100,10 +101,12 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> { } => NodeKind::ExactAgg, SummaryExpr::SummaryAgg { family: planner_types::post_asap::SummaryFamilyType::Sketch(kind, _), + input, .. } => NodeKind::SketchAgg { sketch_type: kind.algorithm(), params: kind.params(), + input, }, // `Plain`/`Sample`/`Wavelet`/`StatModel` never occur on a real // `SummaryAgg` (never `Plain` by construction; `Sample`/ @@ -156,6 +159,8 @@ pub enum EmitError { /// the legacy planner output rather than POST an empty payload. #[error("backend has no sketch consumers; nothing to wire")] BackendEmpty, + #[error("unsupported heap update weight")] + UnsupportedHeapUpdate, } /// Generic emitter trait — Phase E ships only [`ThreeStageEmitter`]; future @@ -886,6 +891,7 @@ impl Emitter for ThreeStageEmitter { NodeKind::SketchAgg { sketch_type, params, + input, }, StageId::Edge, ) => { @@ -902,7 +908,20 @@ impl Emitter for ThreeStageEmitter { }); backend_aggregations.push(BackendAggregation { item_label: None, - heap_update_mode: None, + heap_update_mode: if matches!( + sketch_type, + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ) { + use planner_types::post_asap::SummaryInputExpr; + use planner_types::pre_asap::ColumnRef; + Some(match &input.weight { + SummaryInputExpr::Constant(value) if *value == 1.0 => "count", + SummaryInputExpr::Column(ColumnRef::SampleValue) => "value", + _ => return Err(EmitError::UnsupportedHeapUpdate), + }) + } else { + None + }, aggregation_id, metric_name: edge.source_metric.clone().unwrap_or_default(), family: SummaryFamilyType::Sketch( diff --git a/control_plane/src/physical/workload_planner.rs b/control_plane/src/physical/workload_planner.rs index 5ab88c7f..ea43d465 100644 --- a/control_plane/src/physical/workload_planner.rs +++ b/control_plane/src/physical/workload_planner.rs @@ -70,7 +70,18 @@ fn mvp_deployment_policy( /// `PhysicalExpr` is then fed into `planner::stage_split::split_typed_three_stage` /// + the per-stage emitters in `config::stage_config`. pub fn bind_workload_typed(w: &QueryWorkload) -> Option { - bind_workload_typed_with_item_filter(w, None) + bind_workload_typed_with_evidence(w, None, None) +} + +/// Bind a legacy workload with an explicit Top-K membership certificate. +/// Approximate Top-K fails closed through [`bind_workload_typed`] when this +/// evidence is absent; callers that have validated a fresh certificate use +/// this entry point instead. +pub fn bind_workload_typed_with_topk_evidence( + w: &QueryWorkload, + evidence: &crate::physical::compiler::TopKMembershipEvidence, +) -> Option { + bind_workload_typed_with_evidence(w, None, Some(evidence)) } /// Like [`bind_workload_typed`], but for a `Frequency` statistic, `item_filter` @@ -87,6 +98,14 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option, +) -> Option { + bind_workload_typed_with_evidence(w, item_filter, None) +} + +fn bind_workload_typed_with_evidence( + w: &QueryWorkload, + item_filter: Option<(&str, &str)>, + topk_evidence: Option<&crate::physical::compiler::TopKMembershipEvidence>, ) -> Option { use crate::physical::post_asap::cost_model::ForcedFamilyCostModel; use crate::types_v2::AccuracyTarget; @@ -217,6 +236,12 @@ pub fn bind_workload_typed_with_item_filter( nullable: false, table: None, }, + Column { + name: "endpoint".into(), + dtype: DataType::Utf8, + nullable: false, + table: None, + }, ], 0, vec![vec![0]], @@ -226,17 +251,39 @@ pub fn bind_workload_typed_with_item_filter( range: w.time_window, child: Box::new(scan).into(), }; + // Planner's weighted Top-K contract deliberately accepts only an + // additive ranking input. The legacy workload vocabulary has no TopK + // aggregation variant: the `top_endpoint_qps` contract row arrives as + // `Frequency`, meaning that each observed series occurrence contributes + // one to its rank. Make that previously implicit update semantics + // explicit as an inner Count aggregate. Besides satisfying the typed + // contract, this supplies the PromQL label-set entity identity used as + // the heap item. + let aggregate_child = if statistic == DeploymentIntent::TopK { + QueryExpr::Aggregate { + reduction: planner_types::pre_asap::Reduction::by(vec![2]), + measures: vec![L3AggIntent::Count { + accuracy: accuracy.clone(), + }], + output_names: Vec::new(), + having: None, + child: Box::new(windowed).into(), + } + } else { + windowed + }; let aggregate = QueryExpr::Aggregate { - // Synthetic probe only -- `boundary::implementation_for` (what - // this shape actually drives) keys off `AggIntent`/accuracy alone, - // never `Reduction`, so this value doesn't affect the family pick. - // `PerEntity` is the representative choice for a windowed shape - // with no `by` (ASAPController#163/#165). - reduction: planner_types::pre_asap::Reduction::PerEntity, + // Top-K is a genuine full reduction over the per-item counts above; + // other synthetic probes retain the representative per-entity shape. + reduction: if statistic == DeploymentIntent::TopK { + planner_types::pre_asap::Reduction::by(vec![]) + } else { + planner_types::pre_asap::Reduction::PerEntity + }, measures: vec![intent], output_names: Vec::new(), having: None, - child: Box::new(windowed).into(), + child: Box::new(aggregate_child).into(), }; // ── Drive the picked family directly, bypassing selection ───────── @@ -271,7 +318,35 @@ pub fn bind_workload_typed_with_item_filter( other => other, }; let cost_model = ForcedFamilyCostModel::new(accuracy.clone(), forced); - let node = crate::planner_selection::select_summary(&aggregate, &cost_model).ok()?; + struct Evidence<'a>(Option<&'a crate::physical::compiler::TopKMembershipEvidence>); + impl asap_aware_mapping::AccuracyEvidenceProvider for Evidence<'_> { + fn propagation_stats( + &self, + op: &planner_types::post_asap::CompositionOperator, + _family: &planner_types::post_asap::SummaryFamilyType, + _query: Option<&planner_types::post_asap::SketchQuery>, + ) -> asap_aware_mapping::PropagationStats { + match (op, self.0) { + (planner_types::post_asap::CompositionOperator::TopKSelection, Some(e)) => { + asap_aware_mapping::PropagationStats { + topk_selected_lower_bound: Some(e.selected_lower_bound), + topk_excluded_upper_bound: Some(e.excluded_upper_bound), + topk_interval_failure_probability: Some(e.interval_failure_probability), + ..Default::default() + } + } + _ => Default::default(), + } + } + } + let node = crate::planner_selection::select_summary_with_evidence( + &aggregate, + &cost_model, + &asap_aware_mapping::DefaultAccuracyModel, + &asap_aware_mapping::EqualSplitAllocator, + &Evidence(topk_evidence), + ) + .ok()?; // `implement_tree_with` never *errors* on "nothing bound" — an // intent `boundary::implementation_for`/`CostModel::realize_extension` // can't realize (e.g. `TopK { accuracy: Exact }`, ASAPController#151, @@ -662,6 +737,16 @@ mod tests { } } + fn topk_evidence() -> crate::physical::compiler::TopKMembershipEvidence { + crate::physical::compiler::TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.001, + observed_at_unix_ms: 1, + source: "workload-planner-test".into(), + } + } + #[test] fn typed_binding_http_requests_total_is_raw_passthrough() { // Contract: `http_requests_total` → raw passthrough (no sketch). @@ -715,13 +800,32 @@ mod tests { } #[test] - fn typed_binding_top_endpoint_qps_requires_membership_evidence() { + fn typed_binding_top_endpoint_qps_picks_count_sketch_with_heap() { // Contract: `top_endpoint_qps` → CountSketch (TopK). // The metric-name reclassification reroutes from the AggType // default (Frequency → CMS) to the contract row (TopK → // CountSketch). let w = workload_for("top_endpoint_qps", AggType::Frequency); assert!(bind_workload_typed(&w).is_none()); + let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence()) + .expect("evidenced top_endpoint_qps must bind"); + // Count-ranked producers must not revert to value-weighted runtime defaults. + let configs = crate::physical::stage_split::split_typed_three_stage(&bound).unwrap(); + let backend = configs + .get(&crate::physical::colored_dag::StageId::Backend) + .unwrap(); + let crate::physical::colored_dag::StageConfig::Backend(backend) = backend else { + panic!("expected backend config"); + }; + assert_eq!(backend.aggregations[0].heap_update_mode, Some("count")); + assert_eq!( + extract_family(&bound), + Some(SketchAlgorithm::CountSketchWithHeap), + ); + assert!(matches!( + extract_query(&bound), + Some(planner_types::post_asap::SketchQuery::TopK { k: 10, .. }) + )); } #[test] @@ -811,7 +915,7 @@ mod tests { } #[test] - fn planner_rejects_topk_override_without_membership_evidence() { + fn planner_honors_cms_topk_override() { // CMS-Heap pattern (Cormode & Muthukrishnan 2005): when a // workload's `sketch_family_override` (= // `sketch_type_override`) selects CountMinSketch for a TopK @@ -819,16 +923,23 @@ mod tests { // back to the canonical CountSketch default. let mut w = workload_for("top_endpoint_qps", AggType::Frequency); w.sketch_type_override = Some(SketchType::CountMinSketch); - assert!(bind_workload_typed(&w).is_none()); + let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence()) + .expect("CMS Top-K override must bind with evidence"); + assert_eq!(extract_family(&bound), Some(SketchAlgorithm::CmsWithHeap)); } #[test] - fn planner_default_topk_requires_membership_evidence() { + fn planner_default_topk_uses_count_sketch_with_heap() { // Without any override, the canonical pick for a TopK metric // stays CountSketch(-with-heap) — CMS-Heap is opt-in via // override only. let w = workload_for("top_endpoint_qps", AggType::Frequency); - assert!(bind_workload_typed(&w).is_none()); + let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence()) + .expect("default Top-K must bind with evidence"); + assert_eq!( + extract_family(&bound), + Some(SketchAlgorithm::CountSketchWithHeap), + ); } #[test] diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index f8b43fbe..a56b61e9 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -129,6 +129,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 57791460..99e04e9b 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -117,6 +117,266 @@ fn is_warm(response: &Value) -> bool { }) } +// Both heap implementations must execute registered temporal counts through +// an installed QueryPlan, retaining all three ranked identities and values. +#[tokio::test] +async fn registered_temporal_topk_cms_heap() { + registered_temporal_topk(planner_types::post_asap::SketchAlgorithm::CmsWithHeap).await; +} + +#[tokio::test] +async fn registered_temporal_topk_count_sketch_heap() { + registered_temporal_topk(planner_types::post_asap::SketchAlgorithm::CountSketchWithHeap).await; +} + +async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlgorithm) { + use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler}; + use planner_types::post_asap::{CompositionOperator, SketchQuery, SummaryFamilyType}; + const QUERY: &str = "topk(3, count_over_time(top_endpoint_qps[5s]))"; + struct Evidence; + impl asap_aware_mapping::AccuracyEvidenceProvider for Evidence { + fn propagation_stats( + &self, + op: &CompositionOperator, + _: &SummaryFamilyType, + _: Option<&SketchQuery>, + ) -> asap_aware_mapping::PropagationStats { + if matches!(op, CompositionOperator::TopKSelection) { + asap_aware_mapping::PropagationStats { + topk_selected_lower_bound: Some(95.0), + topk_excluded_upper_bound: Some(80.0), + topk_interval_failure_probability: Some(0.001), + ..Default::default() + } + } else { + Default::default() + } + } + } + let mut fixture: Value = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let mut entry = fixture["query_workload"]["repeating_queries"][5].clone(); + entry["query"] = QUERY.into(); + entry["requirements"]["accuracy"] = + serde_json::json!({"explicit": {"EpsilonDelta": {"epsilon": 0.05, "delta": 0.05}}}); + fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); + fixture["implementation"]["topk_evidence"] = serde_json::json!({ + QUERY: { + "selected_lower_bound": 95.0, "excluded_upper_bound": 80.0, + "interval_failure_probability": 0.001, "observed_at_unix_ms": 9500, + "source": "deterministic-count-ranking-fixture" + } + }); + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); + let (mut request, environment) = snapshot.planning_request().unwrap(); + let query = &mut request.queries[0]; + let expr = + control_plane::query_parser::parse_query_expr_canonical(QUERY, query.accuracy.clone()) + .unwrap(); + let model = control_plane::physical::post_asap::cost_model::ForcedFamilyCostModel::new( + query.accuracy.clone(), + algorithm.clone(), + ); + query.post_asap = control_plane::planner_selection::select_summary_with_evidence( + &expr, + &model, + &asap_aware_mapping::DefaultAccuracyModel, + &asap_aware_mapping::EqualSplitAllocator, + &Evidence, + ) + .unwrap(); + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + use data_plane::storage_engines::types::AggregationType; + let expected_type = match algorithm { + planner_types::post_asap::SketchAlgorithm::CmsWithHeap => { + AggregationType::CountMinSketchWithHeap + } + planner_types::post_asap::SketchAlgorithm::CountSketchWithHeap => { + AggregationType::CountSketchWithHeap + } + _ => panic!("fixture requires a heap implementation"), + }; + assert_eq!( + plan.precompute_plan.materializations[0].aggregation_type, + expected_type + ); + assert_eq!( + plan.precompute_plan.materializations[0].parameters["weight_mode"], + "count" + ); + let artifact = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { + precompute_plan: plan.precompute_plan, + transmission_plan: plan.transmission_plan, + backend_plan: plan.backend_plan.encode_to_vec(), + query_plan: plan.query_plan, + storage_routing: None, + adaptation_evidence: vec![], + }; + let output = tempfile::tempdir().unwrap(); + let mut artifact_file = tempfile::NamedTempFile::new().unwrap(); + serde_json::to_writer(&mut artifact_file, &artifact).unwrap(); + let port = unused_port(); + let fallback = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_url = format!("http://{}", fallback.local_addr().unwrap()); + let fallback_task = tokio::spawn(async move { + axum::serve(fallback, Router::new() + .route("/-/healthy", get(|| async { "healthy" })) + .route("/api/v1/query", get(|| async { Json(serde_json::json!({ + "status": "error", "errorType": "execution", "error": "fixture exact backend unavailable" + })) }))).await.unwrap(); + }); + let mut child = ChildGuard( + Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args(["--profile", "asapquery", "--physical-plan"]) + .arg(artifact_file.path()) + .args([ + "--forward-unsupported-queries", + "--prometheus-server", + &fallback_url, + ]) + .args(["--http-port", &port.to_string(), "--output-dir"]) + .arg(output.path()) + .args([ + "--precompute-allowed-lateness-ms", + "0", + "--precompute-flush-interval-ms", + "25", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + let client = reqwest::Client::new(); + let backend = format!("http://127.0.0.1:{port}"); + wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + let base = now - now.rem_euclid(5000) - 20000; + let counts = [ + ("alpha", 100), + ("beta", 50), + ("gamma", 200), + ("delta", 75), + ("epsilon", 10), + ("zeta", 150), + ]; + let samples = WriteRequest { + timeseries: counts + .iter() + .map(|(item, count)| { + // Non-unit values distinguish count updates from accidental weighted sums. + let points = (0..2) + .flat_map(|window| { + (0..*count).map(move |i| (base + window * 5000 + 10 + i * 20, 17.0)) + }) + .collect::>(); + series_with_labels("top_endpoint_qps", &[("endpoint", item)], &points) + }) + .collect(), + }; + assert_eq!(remote_write(&client, &backend, &samples).await, 204); + let watermark = WriteRequest { + timeseries: vec![series_with_labels( + "top_endpoint_qps", + &[("endpoint", "gamma")], + &[(base + 10500, 17.0)], + )], + }; + assert_eq!(remote_write(&client, &backend, &watermark).await, 204); + assert_eq!(remote_write(&client, &backend, &samples).await, 204); + let timestamp = (base + 5000) as f64 / 1000.0; + let instant = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let response: Value = client + .get(format!("{backend}/api/v1/query")) + .query(&[ + ("query", QUERY.to_string()), + ("time", timestamp.to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + if response["status"] == "success" && is_warm(&response) { + break response; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("registered TopK must become warm within 30s"); + let expected = [("gamma", 200.0), ("zeta", 150.0), ("alpha", 100.0)]; + let assert_ranks = |response: &Value, range: bool| { + assert_eq!(response["status"], "success", "{response}"); + assert!(is_warm(response), "{response}"); + let rows = response["data"]["result"].as_array().unwrap(); + assert_eq!(rows.len(), 3, "{response}"); + for (item, count) in expected { + let row = rows + .iter() + .find(|row| { + row["metric"]["item"].as_str() + == Some(format!("top_endpoint_qps{{endpoint=\"{item}\"}}").as_str()) + }) + .unwrap_or_else(|| panic!("missing {item}: {response}")); + let points = if range { + let points = row["values"].as_array().unwrap(); + assert_eq!(points.len(), 2); + points.clone() + } else { + vec![row["value"].clone()] + }; + for (index, point) in points.iter().enumerate() { + assert_eq!(point[0].as_f64(), Some(timestamp + index as f64 * 5.0)); + assert_eq!( + point[1].as_str().unwrap().parse::().unwrap(), + count, + "{response}" + ); + } + } + }; + assert_ranks(&instant, false); + let range: Value = client + .get(format!("{backend}/api/v1/query_range")) + .query(&[ + ("query", QUERY.to_string()), + ("start", timestamp.to_string()), + ("end", (timestamp + 5.0).to_string()), + ("step", "5".into()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_ranks(&range, true); + let unregistered: Value = client + .get(format!("{backend}/api/v1/query")) + .query(&[("query", "topk(3, top_endpoint_qps)")]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + unregistered["status"], "error", + "unregistered query must not replan: {unregistered}" + ); + assert_eq!(unregistered["error"], "fixture exact backend unavailable"); + fallback_task.abort(); +} + // Three registered consumers must observe one raw SUM/count producer, including // uneven instance sample counts and Remote Write retries. #[tokio::test] diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 878f1fc9..56dcc934 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -117,8 +117,21 @@ fn build_workload( fn plan_backend_stage_config( workload: &QueryWorkload, ) -> control_plane::physical::colored_dag::BackendStageConfig { - let deployment_expr = control_plane::physical::workload_planner::bind_workload_typed(workload) - .expect("bind_workload_typed produced a PhysicalExpr"); + let deployment_expr = if workload.metric_name == "top_endpoint_qps" { + let evidence = control_plane::physical::compiler::TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.001, + observed_at_unix_ms: 1, + source: "self-contained-e2e-fixture".into(), + }; + control_plane::physical::workload_planner::bind_workload_typed_with_topk_evidence( + workload, &evidence, + ) + } else { + control_plane::physical::workload_planner::bind_workload_typed(workload) + } + .expect("typed workload binding produced a PhysicalExpr"); let configs = control_plane::physical::stage_split::split_typed_three_stage(&deployment_expr) .expect("split_typed_three_stage produced per-stage configs"); let mut backend_cfg = configs @@ -1398,11 +1411,11 @@ async fn controller_plan_to_query_full_roundtrip_hll() { // is the canonical TopK metric, so the planner pins // `with_heap: true` and the controller emits `CountSketchWithHeap` // (regardless of override). To match, the wire DP carries a -// msgpack-encoded heap envelope (mirroring Test 9), but the query +// msgpack-encoded heap envelope, but the query // uses `count_over_time(...)` instead of `topk(...)` — the // reducer's `decode_frequency_total` reads row-0 of the underlying // matrix for heap-bearing variants too, so FrequencyEstimate works -// on the same sid that Test 9 queries for top-k. +// on a heap-bearing SID. // // **Strict-success: `count_over_time(top_endpoint_qps[1s])`** binds // to `Capability::FrequencyEstimate(Any)`, which @@ -1434,7 +1447,7 @@ async fn controller_plan_to_query_full_roundtrip_count_sketch() { // Use the planner-picked `(w, d)` so the OTLP DP's wire-level // `rows`/`cols` line up with the policy's `parameters.{d, w}` — - // same shape constraint as Test 9. + // dimension mismatches prevent physical-policy binding. let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json); let rows = d as usize; let cols = w as usize; @@ -1661,62 +1674,6 @@ fn extract_w_d_from_streaming_config(streaming_config_json: &JsonValue) -> (u32, (w, d) } -/// OTLP `ExportMetricsServiceRequest` with a single `CountMinSketch` DP -/// carrying msgpack-encoded heap-bearing bytes. `encoding=MSGPACK` (3) -/// triggers `sketch_algorithm_for`'s auto-promotion to `CmsWithHeap`. -/// `rows`/`cols` on the parent `CountMinSketch` MUST match the policy's -/// `parameters.{d,w}` for the policy_fp content match to bind. -fn build_cms_with_heap_msgpack_export( - metric_name: &str, - attrs: &[(&str, &str)], - time_unix_nano: u64, - sketch_bytes: Vec, - wire_rows: i32, - wire_cols: i32, -) -> ExportMetricsServiceRequest { - let attributes = attrs - .iter() - .map(|(k, v)| KeyValue { - key: k.to_string(), - value: Some(AnyValue { - value: Some(any_value::Value::StringValue(v.to_string())), - }), - }) - .collect(); - let start_t_ns = time_unix_nano.saturating_sub(1_000_000_000); - let dp = CountMinSketchDataPoint { - attributes, - start_time_unix_nano: start_t_ns, - time_unix_nano, - sketch: sketch_bytes, - encoding: CountMinSketchEncoding::Msgpack as i32, - flags: 0, - series_id: 0, - }; - ExportMetricsServiceRequest { - resource_metrics: vec![ResourceMetrics { - resource: None, - scope_metrics: vec![ScopeMetrics { - scope: None, - metrics: vec![Metric { - name: metric_name.to_string(), - description: String::new(), - unit: String::new(), - metadata: Vec::new(), - data: Some(Data::Countminsketch(CountMinSketch { - data_points: vec![dp], - aggregation_temporality: 0, - rows: wire_rows, - cols: wire_cols, - })), - }], - schema_url: String::new(), - }], - schema_url: String::new(), - }], - } -} - /// OTLP `ExportMetricsServiceRequest` with a single `CountSketch` DP /// carrying msgpack-encoded heap-bearing bytes. `encoding=MSGPACK` (3) /// triggers `sketch_algorithm_for`'s auto-promotion to @@ -1773,307 +1730,9 @@ fn build_count_sketch_with_heap_msgpack_export( } } -// ── Test 8 — heap-bearing CMS + topk strict-success ───────────────────────── -// -// The full top-k roundtrip with `CmsWithHeap`. Workload pins CMS via -// `sketch_type_override: Some(SketchType::CountMinSketch)` on the -// `top_endpoint_qps` metric (TopK statistic class), which the planner -// binds via `bind_cms_with_heap_on_topk` (CMS-Heap pattern from -// Cormode & Muthukrishnan 2005). -// -// The OTLP DP carries a msgpack-encoded `CountMinSketchWithHeap` -// payload (`encoding=MSGPACK`); the receiver's `sketch_algorithm_for` -// peeks at the bytes and auto-promotes the sid to `CmsWithHeap`, -// registering it under `Capability::FrequencyTopk(CmsWithHeap)`. -// -// The reducer's `topk` family decodes the heap directly via -// `decode_cms_with_heap_from_msgpack` and emits one output series -// per top-k item with the item key in the `item` label. - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn controller_plan_to_query_full_roundtrip_cms_with_heap_topk() { - let stack = start_full_stack(19_571, 19_572).await; - let client = reqwest::Client::new(); - - let workload = build_workload_with_override( - "top_endpoint_qps", - vec![AggType::Frequency], - 0.05, - Duration::from_secs(1), - vec!["service".to_string()], - Vec::new(), - Some(SketchType::CountMinSketch), - ); - let streaming_config_json = plan_streaming_config_json(&workload); - // The controller now emits `CountMinSketchWithHeap` directly when - // the planner-set `with_heap: true` flag on `CmsParams` fires - // (see `sketch_kind_to_backend_type`). No in-test JSON patch is - // needed — the analyzer ↔ policy match binds against the - // controller-emitted aggregation type as-is. - assert_eq!( - streaming_config_json["aggregations"][0]["aggregationType"], "CountMinSketchWithHeap", - "controller must emit CountMinSketchWithHeap when bind_cms_with_heap_on_topk fires\n{streaming_config_json}" - ); - assert_eq!( - streaming_config_json["readouts"][0]["op"], "topk", - "controller must emit a topk readout for CMS-with-heap binding\n{streaming_config_json}" - ); - post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; - - // Use the planner-picked `(w, d)` so the OTLP DP's wire-level - // `rows`/`cols` line up with the policy's `parameters.{d, w}` — - // `derive_sketch_policy_fp` content-matches on these keys, so a - // dimension mismatch leaves the sid registered with `policy_fp` - // = UNSET (unreachable through `sids_for_policy`). - let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json); - let rows = d as usize; - let cols = w as usize; - let wire_rows = d as i32; - let wire_cols = w as i32; - - // Heap items with deterministic count ordering. `gamma` is the - // unambiguous top-1 (count=200); the heap (top_k=10) keeps all six. - let items: &[(&str, u64)] = &[ - ("alpha", 100), - ("beta", 50), - ("gamma", 200), - ("delta", 75), - ("epsilon", 10), - ("zeta", 150), - ]; - let sketch_bytes = build_heap_bearing_msgpack(rows, cols, 10, items); - - let now_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("system time before UNIX epoch") - .as_nanos() as u64; - let sketch_t_ns = now_ns.saturating_sub(3_000_000_000); - let watermark_t_ns = now_ns.saturating_sub(1_000_000_000); - - let req = build_cms_with_heap_msgpack_export( - "top_endpoint_qps", - &[("service", "e2e-test")], - sketch_t_ns, - sketch_bytes.clone(), - wire_rows, - wire_cols, - ); - post_otlp_http(&client, stack.otlp_http_port, req).await; - - // Watermark MUST also carry the heap — the reducer reads - // `samples.iter().next_back()` and decodes the latest sample's - // bytes; an empty-heap watermark would shadow the real payload. - let watermark_req = build_cms_with_heap_msgpack_export( - "top_endpoint_qps", - &[("service", "e2e-test")], - watermark_t_ns, - sketch_bytes, - wire_rows, - wire_cols, - ); - post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; - - tokio::time::sleep(Duration::from_millis(800)).await; - - let response: JsonValue = client - .get(format!( - "http://127.0.0.1:{}/api/v1/query", - stack.backend_port - )) - .query(&[("query", "topk(3, top_endpoint_qps)")]) - .send() - .await - .expect("query failed") - .json() - .await - .expect("response not JSON"); - let status = response["status"].as_str().unwrap_or("(missing)"); - assert_eq!( - status, - "success", - "topk(...) on CmsWithHeap must succeed end-to-end. Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); - // Top-1 must be `gamma` (count=200), surfaced via the - // `item: ` synthesized label on each top-k series. The - // `InstantVectorElement::label_keys_override` field (added - // alongside this assertion's tightening) carries the synthesized - // key through the Prometheus adapter. - let result = &response["data"]["result"]; - let arr = result - .as_array() - .expect("result must be an array of vector elements"); - assert!( - !arr.is_empty() && arr.len() <= 3, - "topk(3) must return between 1 and 3 series. Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); - let mut values: Vec = arr - .iter() - .filter_map(|e| e["value"][1].as_str().and_then(|s| s.parse::().ok())) - .collect(); - values.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)); - let mut found_gamma = false; - for elem in arr { - if let Some(item) = elem["metric"]["item"].as_str() { - if item == "gamma" { - found_gamma = true; - break; - } - } - } - assert!( - found_gamma, - "topk(3) must surface `gamma` via the `item` label on at least \ - one series. Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); - assert!( - values - .first() - .map(|v| (v - 200.0).abs() < 1.0) - .unwrap_or(false), - "topk(3) on heap-bearing CMS must surface `gamma`'s count (200) as \ - the top value (received {values:?}). Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); -} - -// ── Test 9 — heap-bearing CountSketch + topk strict-success ───────────────── -// -// Same shape as Test 8 but the workload defaults the `top_endpoint_qps` -// metric to `CountSketch` (the canonical TopK pick — unbiased -// estimator, see `BindCountSketchOnTopK`). The OTLP DP carries the -// SAME msgpack heap envelope (the wire shape is shared); only the -// outer DP type changes (`CountSketchDataPoint` instead of -// `CountMinSketchDataPoint`). -// -// `sketch_algorithm_for` was extended in this PR to peek at -// CountSketch DPs the same way it does for CountMin — a -// non-empty heap in a msgpack-encoded payload promotes the sid to -// `CountSketchWithHeap`, which the analyzer's `is_satisfied_by` -// recognises as a valid `FrequencyTopk(CountSketchWithHeap)` provider. - -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn controller_plan_to_query_full_roundtrip_count_sketch_with_heap_topk() { - let stack = start_full_stack(19_573, 19_574).await; - let client = reqwest::Client::new(); - - let workload = build_workload_with_override( - "top_endpoint_qps", - vec![AggType::Frequency], - 0.05, - Duration::from_secs(1), - vec!["service".to_string()], - Vec::new(), - None, // default → CountSketch (canonical TopK pick) - ); - let streaming_config_json = plan_streaming_config_json(&workload); - // The controller now emits `CountSketchWithHeap` directly when - // the planner-set `with_heap: true` flag on `CountSketchParams` - // fires (see `sketch_kind_to_backend_type`). No in-test JSON - // patch is needed. - assert_eq!( - streaming_config_json["aggregations"][0]["aggregationType"], "CountSketchWithHeap", - "controller must emit CountSketchWithHeap for default top_endpoint_qps TopK binding\n{streaming_config_json}" - ); - assert_eq!( - streaming_config_json["aggregations"][0]["parameters"]["with_heap"], true, - "controller must set parameters.with_heap=true for CountSketch TopK binding\n{streaming_config_json}" - ); - post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; - - let items: &[(&str, u64)] = &[ - ("alpha", 100), - ("beta", 50), - ("gamma", 200), - ("delta", 75), - ("epsilon", 10), - ("zeta", 150), - ]; - let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json); - let rows = d as usize; - let cols = w as usize; - let wire_rows = d as i32; - let wire_cols = w as i32; - let sketch_bytes = build_heap_bearing_msgpack(rows, cols, 10, items); - - let now_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("system time before UNIX epoch") - .as_nanos() as u64; - let sketch_t_ns = now_ns.saturating_sub(3_000_000_000); - let watermark_t_ns = now_ns.saturating_sub(1_000_000_000); - - let req = build_count_sketch_with_heap_msgpack_export( - "top_endpoint_qps", - &[("service", "e2e-test")], - sketch_t_ns, - sketch_bytes.clone(), - wire_rows, - wire_cols, - ); - post_otlp_http(&client, stack.otlp_http_port, req).await; - - let watermark_req = build_count_sketch_with_heap_msgpack_export( - "top_endpoint_qps", - &[("service", "e2e-test")], - watermark_t_ns, - sketch_bytes, - wire_rows, - wire_cols, - ); - post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; - - tokio::time::sleep(Duration::from_millis(800)).await; - - let response: JsonValue = client - .get(format!( - "http://127.0.0.1:{}/api/v1/query", - stack.backend_port - )) - .query(&[("query", "topk(3, top_endpoint_qps)")]) - .send() - .await - .expect("query failed") - .json() - .await - .expect("response not JSON"); - let status = response["status"].as_str().unwrap_or("(missing)"); - assert_eq!( - status, - "success", - "topk(...) on CountSketchWithHeap must succeed end-to-end. Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); - // Same shape-only assertion as Test 8 — `InstantVectorElement` - // currently drops per-element labels, so we can't check for - // `item: "gamma"`. Verify the strongest invariants the wire - // surfaces today: 1..=3 series and gamma's count (200) leads. - let result = &response["data"]["result"]; - let arr = result - .as_array() - .expect("result must be an array of vector elements"); - assert!( - !arr.is_empty() && arr.len() <= 3, - "topk(3) must return between 1 and 3 series. Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); - let mut values: Vec = arr - .iter() - .filter_map(|e| e["value"][1].as_str().and_then(|s| s.parse::().ok())) - .collect(); - values.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)); - assert!( - values - .first() - .map(|v| (v - 200.0).abs() < 1.0) - .unwrap_or(false), - "topk(3) on heap-bearing CountSketch must surface `gamma`'s count (200) \ - as the top value (received {values:?}). Response:\n{}", - serde_json::to_string_pretty(&response).unwrap_or_default() - ); -} +// Heap TopK serving acceptance lives in asapquery_compatibility_process_e2e: +// registered_temporal_topk_{cms_heap,count_sketch_heap} install the selected +// physical QueryPlan and verify raw count updates through the production binary. // ── Test 10 — range-query warm-tier fallback (CMS + count_over_time) ──────── //