feat(query): warm-tier fallback for /api/v1/query_range

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()}

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<u64> = 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<u64> = 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,

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<i64> = (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()
+    );
+}