diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index a9d292ca..ed2d4df2 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -1698,7 +1698,65 @@ async fn process_range_query_request( Err(status) => status.into_response()} } None => { - debug!("Range query returned None - query not supported"); + // Legacy `handle_range_query_promql` returned None — try + // the modern warm-tier path. Mirrors the instant-query + // fallback in `process_via_simple_engine` that PR #253 + // wired through the trait's `execute(&str)`; this site + // uses the range-aware sibling + // `execute_range_promql_modern(query, start, end, step)` + // which returns Matrix per the + // `/api/v1/query_range` wire-format requirement. + // + // Shapes that go through this fallback: anything the + // legacy path doesn't know (notably the modified-OTLP + // sketch-backed sids — count_over_time / quantile_over_time + // / etc. against CMS / CountSketch / KLL / DDSketch / HLL + // policies). Shapes still unsupported in the warm tier + // (topk_over_time — not standard PromQL anyway) fall + // through this branch too and continue to the unsupported- + // query response, which the EngineRouter can route to a + // cold-tier fallback if one is configured. + let start_ms = (parsed_request.start * 1000.0) as u64; + let end_ms = (parsed_request.end * 1000.0) as u64; + let step_ms = (parsed_request.step * 1000.0) as u64; + let modern_result = state + .query_engine + .execute_range_promql_modern( + &parsed_request.query, + start_ms, + end_ms, + step_ms, + ) + .await; + if let Ok(query_result) = modern_result { + debug!( + "Modern execute_range_promql_modern handled what legacy \ + handle_range_query_promql missed (query='{}')", + parsed_request.query + ); + let total_duration = start_time.elapsed(); + debug!( + "Total range query processing took (modern fallback): {:.2}ms", + total_duration.as_secs_f64() * 1000.0 + ); + return match state + .adapter + .format_range_success_response( + &query_result, + &promql_utilities::data_model::KeyByLabelNames::default(), + ) + .await + { + Ok(response) => response.into_response(), + Err(status) => status.into_response(), + }; + } + + debug!( + "Both legacy and modern range-query paths returned None/Err \ + for query='{}', falling through to unsupported", + parsed_request.query + ); match state.adapter.format_unsupported_query_response().await { Ok(json) => json.into_response(), Err(status) => status.into_response()} diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index e814b2e4..178f28bb 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -3172,6 +3172,149 @@ impl ASAPQueryEngine { )) } + /// Modern warm-tier path for `/api/v1/query_range` — the range- + /// query equivalent of the `QueryEngine::execute(&str)` trait + /// surface. Used by the HTTP server as a fallback when the legacy + /// `handle_range_query_promql` returns `None`. + /// + /// Time semantics follow Prometheus's + /// `/api/v1/query_range?start&end&step` spec: the result is a + /// `matrix` (one row per series, each row carrying multiple + /// (timestamp, value) samples). The warm-tier reducer naturally + /// produces one sample per window_close in `[start, end]`, so + /// the matrix is sampled at the underlying aggregation's window + /// boundaries — typically a finer grid than the user's `step` + /// when window_size < step. (The Prometheus spec says + /// evaluate at each step `t = start, start+step, …, end`; the + /// warm tier returns at native window-close granularity instead. + /// This is more data, not less — clients that expect exact step + /// timestamps can downsample, or route step-precise queries to + /// the cold tier via the EngineRouter.) + /// + /// `step` is currently accepted for API compatibility but unused + /// — see the granularity-mismatch note above. + pub async fn execute_range_promql_modern( + &self, + query: &str, + start_ms: u64, + end_ms: u64, + _step_ms: u64, + ) -> Result< + crate::query_engines::query_result::QueryResult, + crate::query_engines::EngineError, + > { + let Some(idx) = self.sketch_index.as_ref() else { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!("ASAPQueryEngine: no sketch index for `{query}` — failing over"), + )); + }; + + let analysis = + control_plane::asap_tier_analysis::analyze_promql_for_asap_tier(query); + + if let Some(reason) = &analysis.unsupported { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore analyzer rejected `{query}` for range query: \ + {reason:?} — failing over to archive" + ), + )); + } + if analysis.candidates.is_empty() { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore analyzer produced no ASAP-tier candidates for \ + `{query}` — failing over to archive" + ), + )); + } + + let streaming_snap = self.streaming_config_snapshot(); + let policy_registry = streaming_snap.policy_registry(); + let reducer = crate::storage_engines::sketch_db::query::SketchReducer::new(idx); + let mut combined_result: Option< + crate::storage_engines::sketch_db::query::ASAPTierResult, + > = None; + + for candidate in &analysis.candidates { + let policy_fps = control_plane::asap_tier_analysis::find_matching_policies( + &policy_registry, + candidate, + ); + let mut sids: Vec = Vec::new(); + for fp in &policy_fps { + sids.extend(idx.sids_for_policy(*fp)); + } + if sids.is_empty() { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore has no policy for metric `{}` satisfying \ + capability {:?} — failing over to archive", + candidate.metric_name, candidate.required_capability, + ), + )); + } + + let required: crate::storage_engines::sketch_db::index::Capability = + candidate.required_capability.clone(); + let mut hit_sids: Vec = Vec::with_capacity(sids.len()); + for sid in &sids { + let meta = match idx.instance(*sid) { + Some(m) => m, + None => continue, + }; + if let Some(cap) = meta.capability.as_ref() { + if required.is_satisfied_by(cap) { + hit_sids.push(*sid); + } + } + } + if hit_sids.is_empty() { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore has no sid satisfying capability {:?} for \ + metric `{}` — failing over to archive", + candidate.required_capability, candidate.metric_name + ), + )); + } + + let result = reducer + .evaluate( + &hit_sids, + &candidate.function, + &candidate.function_args, + start_ms, + end_ms, + ) + .map_err(|e| { + crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore reducer failed for `{query}` over \ + [{start_ms}, {end_ms}]: {e:?} — failing over to archive" + ), + ) + })?; + combined_result = Some(result); + } + + let result = combined_result.ok_or_else(|| { + crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!("SketchStore reducer produced no result for `{query}`"), + ) + })?; + + // Matrix shape — the range_query wire format requires it. + Ok(asap_tier_result_to_query_result(result, end_ms, true)) + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, 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 689ab6ea..e8d96e88 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -1903,3 +1903,121 @@ async fn controller_plan_to_query_full_roundtrip_count_sketch_with_heap_topk() { serde_json::to_string_pretty(&response).unwrap_or_default() ); } + +// ── Test 10 — range-query warm-tier fallback (CMS + count_over_time) ──────── +// +// `/api/v1/query_range` previously had no warm-tier fallback — +// when the legacy `handle_range_query_promql` returned `None` (which +// it does for sketch-backed sids), the handler immediately fell +// through to `format_unsupported_query_response` ("No result for +// query"). This PR adds an `execute_range_promql_modern` modern +// path that mirrors PR #253's `process_via_simple_engine` fallback +// for instant queries. +// +// Same wire setup as Test 7 (heap-less CMS, `endpoint_request_freq` +// with `AggType::Frequency`), but the query goes through +// `/api/v1/query_range?query=count_over_time(metric[10s])` instead +// of the instant endpoint. The result `resultType` is `matrix` +// (Prometheus spec for range queries). + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn controller_plan_to_range_query_count_over_time_cms() { + let stack = start_full_stack(19_575, 19_576).await; + let client = reqwest::Client::new(); + + let workload = build_workload_with_override( + "endpoint_request_freq", + 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); + post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; + + let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json); + let rows = d; + let cols = w; + let counts: Vec = (0..(rows * cols) as i64).map(|i| (i % 11).abs()).collect(); + let cms_state = build_count_min_state(rows, cols, counts); + let sketch_bytes = cms_state.encode_to_vec(); + + 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_min_export( + "endpoint_request_freq", + &[("service", "e2e-test")], + sketch_t_ns, + sketch_bytes, + rows as i32, + cols as i32, + ); + post_otlp_http(&client, stack.otlp_http_port, req).await; + + let watermark_state = build_count_min_state(rows, cols, vec![0i64; (rows * cols) as usize]); + let watermark_req = build_count_min_export( + "endpoint_request_freq", + &[("service", "e2e-test")], + watermark_t_ns, + watermark_state.encode_to_vec(), + rows as i32, + cols as i32, + ); + post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; + + tokio::time::sleep(Duration::from_millis(800)).await; + + // Query range covering the watermark + sketch windows. Prometheus's + // /api/v1/query_range expects epoch-second floats for start/end/step. + let now_secs = now_ns as f64 / 1e9; + let start_secs = now_secs - 10.0; + let end_secs = now_secs; + let step_secs = 1.0; + let response: JsonValue = client + .get(format!( + "http://127.0.0.1:{}/api/v1/query_range", + stack.backend_port + )) + .query(&[ + ("query", "count_over_time(endpoint_request_freq[10s])"), + ("start", &format!("{start_secs}")), + ("end", &format!("{end_secs}")), + ("step", &format!("{step_secs}")), + ]) + .send() + .await + .expect("range query failed") + .json() + .await + .expect("response not JSON"); + let status = response["status"].as_str().unwrap_or("(missing)"); + assert_eq!( + status, "success", + "count_over_time(...) range-query against heap-less CMS must succeed \ + end-to-end via the modern execute_range_promql_modern fallback. \ + Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); + assert_eq!( + response["data"]["resultType"], "matrix", + "range-query result must carry resultType=matrix per the Prometheus \ + /api/v1/query_range wire spec. Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); + let result = &response["data"]["result"]; + let arr = result + .as_array() + .expect("data.result must be an array of matrix elements"); + assert!( + !arr.is_empty(), + "matrix result must contain at least one series. Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); +}