feat(http): surface §7 Partial results on Prometheus warnings

asap-query-engine/src/drivers/query/adapters/prometheus_http.rs

@@ -24,6 +24,13 @@ pub struct PrometheusResponse {
     pub error_type: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,
+    /// Non-error advisories — maps to Prometheus's top-level
+    /// `warnings: []` field. Phase 3b-2-b uses this to surface
+    /// partial results from the §7 schema-timeline dispatcher
+    /// (query spans a reconfigure boundary with a non-combinable
+    /// statistic or a Purged segment).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub warnings: Vec<String>,
 }
 
 impl PrometheusResponse {
@@ -33,6 +40,19 @@ impl PrometheusResponse {
             data: Some(data),
             error_type: None,
             error: None,
+            warnings: Vec::new(),
+        }
+    }
+
+    /// `success` + a non-empty `warnings` list attached. Used by the
+    /// query adapters when the engine returned a `CombinedResult::Partial`.
+    pub fn success_with_warnings(data: Value, warnings: Vec<String>) -> Self {
+        Self {
+            status: "success".to_string(),
+            data: Some(data),
+            error_type: None,
+            error: None,
+            warnings,
         }
     }
 
@@ -42,6 +62,7 @@ impl PrometheusResponse {
             data: None,
             error_type: Some(error_type.to_string()),
             error: Some(error.to_string()),
+            warnings: Vec::new(),
         }
     }
 }
@@ -200,7 +221,15 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
                 StatusCode::INTERNAL_SERVER_ERROR
             })?;
 
-        let response = PrometheusResponse::success(prometheus_data);
+        // Thread through any Phase 3 timeline-dispatch warnings so
+        // they land on the top-level `warnings` field, matching
+        // Prometheus's native API.
+        let warnings = result.query_result.warnings().to_vec();
+        let response = if warnings.is_empty() {
+            PrometheusResponse::success(prometheus_data)
+        } else {
+            PrometheusResponse::success_with_warnings(prometheus_data, warnings)
+        };
         Ok(Json(serde_json::to_value(response).unwrap()).into_response())
     }
 
@@ -215,7 +244,12 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
             error!("Failed to convert range result: {}", e);
             StatusCode::INTERNAL_SERVER_ERROR
         })?;
-        let response = PrometheusResponse::success(prometheus_data);
+        let warnings = result.warnings().to_vec();
+        let response = if warnings.is_empty() {
+            PrometheusResponse::success(prometheus_data)
+        } else {
+            PrometheusResponse::success_with_warnings(prometheus_data, warnings)
+        };
         Ok(Json(serde_json::to_value(response).unwrap()).into_response())
     }
 
@@ -512,4 +546,34 @@ mod tests {
         let parsed = result.unwrap();
         assert_eq!(parsed.query, "sum(metric)");
     }
+
+    #[test]
+    fn success_response_without_warnings_omits_field() {
+        let r = PrometheusResponse::success(json!({"resultType": "vector", "result": []}));
+        let s = serde_json::to_string(&r).unwrap();
+        assert!(s.contains("\"status\":\"success\""));
+        assert!(
+            !s.contains("\"warnings\""),
+            "empty warnings must be skip-serialised for wire compatibility"
+        );
+    }
+
+    #[test]
+    fn success_response_with_warnings_serialises_the_top_level_field() {
+        // This is the Phase 3b-2-b contract: a Partial result coming
+        // out of the §7 timeline dispatcher lands on Prometheus's
+        // native `warnings: []` field at the top of the response,
+        // matching upstream behaviour for warning-carrying queries.
+        let r = PrometheusResponse::success_with_warnings(
+            json!({"resultType": "vector", "result": []}),
+            vec![
+                "partial result: query spans 2 schemas".to_string(),
+                "1 group(s) dropped".to_string(),
+            ],
+        );
+        let s = serde_json::to_string(&r).unwrap();
+        assert!(s.contains("\"warnings\":["));
+        assert!(s.contains("partial result: query spans 2 schemas"));
+        assert!(s.contains("1 group(s) dropped"));
+    }
 }

asap-query-engine/src/engines/query_result.rs

@@ -19,11 +19,47 @@ impl QueryResult {
     }
 
     pub fn vector(values: Vec<InstantVectorElement>, timestamp: u64) -> Self {
-        QueryResult::Vector(InstantVector { values, timestamp })
+        QueryResult::Vector(InstantVector {
+            values,
+            timestamp,
+            warnings: Vec::new(),
+        })
+    }
+
+    /// Phase 3b-2-b: construct an instant vector with a non-empty
+    /// warnings list. Used by the timeline dispatcher when the query
+    /// spans a reconfigure boundary and one or more segments could
+    /// not contribute to the answer (non-combinable statistic, purged
+    /// data, or agg_id removed from config mid-flight). Prometheus's
+    /// native JSON surface carries these back to the caller via the
+    /// top-level `warnings` field, matching the upstream contract.
+    pub fn vector_with_warnings(
+        values: Vec<InstantVectorElement>,
+        timestamp: u64,
+        warnings: Vec<String>,
+    ) -> Self {
+        QueryResult::Vector(InstantVector {
+            values,
+            timestamp,
+            warnings,
+        })
     }
 
     pub fn matrix(values: Vec<RangeVectorElement>) -> Self {
-        QueryResult::Matrix(RangeVector { values })
+        QueryResult::Matrix(RangeVector {
+            values,
+            warnings: Vec::new(),
+        })
+    }
+
+    /// Accumulated per-result warnings. Empty for single-schema
+    /// queries; populated by the §7 timeline dispatcher when a query
+    /// spans a reconfigure boundary.
+    pub fn warnings(&self) -> &[String] {
+        match self {
+            QueryResult::Vector(iv) => &iv.warnings,
+            QueryResult::Matrix(m) => &m.warnings,
+        }
     }
 }
 
@@ -32,6 +68,15 @@ impl QueryResult {
 pub struct InstantVector {
     pub values: Vec<InstantVectorElement>,
     pub timestamp: u64,
+    /// Non-error advisories attached to this result, surfaced on
+    /// Prometheus's top-level `warnings` field. Empty for
+    /// single-schema queries; populated by the Phase 3b-2-b timeline
+    /// dispatcher when one or more segments produced a
+    /// [`crate::engines::timeline_dispatch::CombinedResult::Partial`]
+    /// (non-combinable statistic, purged coverage, or agg_id
+    /// missing from the current config).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub warnings: Vec<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -50,6 +95,9 @@ impl InstantVectorElement {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RangeVector {
     pub values: Vec<RangeVectorElement>,
+    /// See [`InstantVector::warnings`].
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub warnings: Vec<String>,
 }
 
 /// Individual element in a range vector
@@ -235,4 +283,40 @@ mod tests {
         assert_eq!(sample.timestamp, 12345);
         assert_eq!(sample.value, 99.9);
     }
+
+    #[test]
+    fn vector_without_warnings_returns_empty_slice_and_omits_field_in_json() {
+        let labels = create_test_labels();
+        let el = InstantVectorElement::new(labels, 1.0);
+        let qr = QueryResult::vector(vec![el], 100);
+        assert!(qr.warnings().is_empty());
+        let json = serde_json::to_string(&qr).unwrap();
+        assert!(
+            !json.contains("\"warnings\""),
+            "default-empty warnings must be skip-serialised for wire compatibility"
+        );
+    }
+
+    #[test]
+    fn vector_with_warnings_round_trips_through_serde() {
+        let labels = create_test_labels();
+        let el = InstantVectorElement::new(labels, 1.0);
+        let warnings = vec!["partial result: 2 schemas".to_string()];
+        let qr = QueryResult::vector_with_warnings(vec![el], 100, warnings.clone());
+        assert_eq!(qr.warnings(), warnings.as_slice());
+
+        let json = serde_json::to_string(&qr).unwrap();
+        assert!(json.contains("\"warnings\""));
+        let back: QueryResult = serde_json::from_str(&json).unwrap();
+        assert_eq!(back.warnings(), warnings.as_slice());
+    }
+
+    #[test]
+    fn matrix_warnings_default_empty_and_skip_serialised() {
+        let el = RangeVectorElement::new(create_test_labels());
+        let qr = QueryResult::matrix(vec![el]);
+        assert!(qr.warnings().is_empty());
+        let json = serde_json::to_string(&qr).unwrap();
+        assert!(!json.contains("\"warnings\""));
+    }
 }

asap-query-engine/src/engines/simple_engine.rs

@@ -3007,30 +3007,38 @@ impl SimpleEngine {
     }
 
     /// Phase 3b-2-b: per-segment dispatch across the §7 schema
-    /// timeline for combinable statistics.
+    /// timeline.
     ///
-    /// Returns `Some(result)` when:
-    /// * `SchemaRegistry::timeline_for_metric` yields two or more
-    ///   segments for the query's metric within its time range
-    ///   (i.e. the query spans a reconfigure boundary), AND
-    /// * the query's statistic is one of Count / Sum / Min / Max,
-    ///   which `timeline_dispatch::combine_statistic` can stitch
-    ///   cleanly at the scalar level.
+    /// Returns `Some(result)` when `SchemaRegistry::timeline_for_metric`
+    /// yields two or more segments for the query's metric within its
+    /// time range (i.e. the query spans a reconfigure boundary).
+    /// Returns `None` otherwise (single-schema range, unparseable
+    /// query, unresolved probe aggregation) so the caller falls back
+    /// to the default single-agg path — that path is still correct
+    /// whenever the timeline doesn't actually span a boundary.
     ///
-    /// Returns `None` otherwise (single-schema range, non-combinable
-    /// statistic, unparseable query, unresolved probe aggregation).
-    /// The caller falls back to the default single-agg path — that
-    /// path is still correct whenever the timeline doesn't actually
-    /// span a boundary. Non-combinable statistics (quantile / topk /
-    /// cardinality / rate / increase) are routed through the
-    /// default path here too; PR B2 will surface
-    /// [`crate::engines::timeline_dispatch::CombinedResult::Partial`]
-    /// to the HTTP response so users can see "covered" + "missing"
-    /// segments explicitly instead of the single-agg data cliff.
+    /// ## Combinable vs non-combinable statistics
+    ///
+    /// For combinable scalar statistics (Count / Sum / Min / Max)
+    /// every segment contributes and the result is a clean `Full`
+    /// value the user can trust without caveat.
+    ///
+    /// For non-combinable statistics (Quantile / Topk / Cardinality /
+    /// Rate / Increase) — or for any combinable run that includes a
+    /// `Purged` / config-missing segment — `combine_statistic`
+    /// returns `Partial`. This method surfaces Partial on the
+    /// Prometheus HTTP response's `warnings` field: the top-level
+    /// result carries whatever combinable prefix we could compute
+    /// (for additive stats) or an empty vector (for non-combinable),
+    /// plus one or more `warnings` strings explaining the schema
+    /// boundary, the dropped groups, and the unresolved segments.
     ///
     /// Delivers the user-visible Phase 3 outcome documented in
     /// `docs/design-sketch-db.md` §7: queries spanning a reconfigure
-    /// boundary no longer see a data cliff for additive statistics.
+    /// boundary no longer see a silent data cliff — additive stats
+    /// get the combined answer, and non-combinable stats get an
+    /// explicit Partial notice instead of the arbitrary single-agg
+    /// single-segment result.
     fn try_handle_query_promql_via_timeline(
         &self,
         query: &str,
@@ -3070,15 +3078,6 @@ impl SimpleEngine {
             return None;
         }
 
-        // Phase 3: only activate for combinable statistics. See the
-        // module doc on `timeline_dispatch` §7.3 combinability table.
-        if !matches!(
-            stat,
-            Statistic::Count | Statistic::Sum | Statistic::Min | Statistic::Max
-        ) {
-            return None;
-        }
-
         debug!(
             metric = %metric_name,
             segments = segments.len(),
@@ -3156,7 +3155,14 @@ impl SimpleEngine {
         // combiner folds per-segment scalars into one final scalar
         // per group. Groups that only appear in `unresolved` (no
         // segment ever produced a value for them) are skipped.
+        //
+        // `any_partial` tracks whether any group came back non-`Full`
+        // — drives the Prometheus `warnings` surface below so the
+        // caller sees "this answer is partial" explicitly instead of
+        // a silent cliff.
         let mut output: Vec<InstantVectorElement> = Vec::new();
+        let mut any_partial = false;
+        let mut groups_with_no_value = 0usize;
         for (label_key, segment_values) in per_group {
             match combine_statistic(stat, &segment_values, &unresolved) {
                 CombinedResult::Full(v) => {
@@ -3165,22 +3171,68 @@ impl SimpleEngine {
                 CombinedResult::Partial {
                     covered: Some(v), ..
                 } => {
-                    // Best-effort: emit `covered` for combinable
-                    // stats so the user sees the partial sum. PR B2
-                    // will add a first-class Partial response surface
-                    // carrying the `missing` list.
+                    // Emit `covered` for combinable stats so the user
+                    // sees the partial sum. The warning below tells
+                    // them not to trust the scalar as a full range
+                    // answer.
                     output.push(InstantVectorElement::new(label_key.unwrap_or_default(), v));
+                    any_partial = true;
                 }
                 CombinedResult::Partial { covered: None, .. } => {
-                    // No segment produced a value for this group —
-                    // drop it rather than emit a misleading 0.
+                    // Non-combinable stat (quantile / topk / rate /
+                    // increase / cardinality) OR a group the
+                    // combiner couldn't reduce. Drop the group —
+                    // there is no meaningful scalar to show — but
+                    // flag the whole response partial.
+                    any_partial = true;
+                    groups_with_no_value += 1;
                 }
             }
         }
 
+        // Phase 6: build Prometheus `warnings` when the combiner
+        // returned any Partial. One line summarising the schema
+        // boundary, plus up-to-three per-segment lines with agg_id
+        // + clipped range so operators can correlate against the
+        // `GET /api/v1/db/timeline` surface. We cap at three to
+        // keep responses bounded; the full set is still inspectable
+        // via the timeline endpoint.
+        let warnings = if any_partial {
+            let mut w = Vec::with_capacity(2 + unresolved.len().min(3));
+            w.push(format!(
+                "partial result: query spans {} schemas for metric '{}' over [{}, {}] and the requested statistic {:?} is not cleanly combinable across schema boundaries — see `GET /api/v1/db/timeline?metric={}&start_ms={}&end_ms={}` for the full segment map",
+                segments.len(),
+                metric_name,
+                t1,
+                t2,
+                stat,
+                metric_name,
+                t1,
+                t2,
+            ));
+            if groups_with_no_value > 0 {
+                w.push(format!(
+                    "{} group(s) dropped because no segment could answer the statistic",
+                    groups_with_no_value,
+                ));
+            }
+            for seg in unresolved.iter().take(3) {
+                w.push(format!(
+                    "segment agg_id={} [{}, {}) status={:?} coverage={:?} unresolved",
+                    seg.agg_id, seg.start_ms, seg.end_ms, seg.status, seg.coverage,
+                ));
+            }
+            if unresolved.len() > 3 {
+                w.push(format!("... and {} more", unresolved.len() - 3));
+            }
+            w
+        } else {
+            Vec::new()
+        };
+
         Some((
             probe_context.metadata.query_output_labels,
-            QueryResult::vector(output, probe_context.query_time),
+            QueryResult::vector_with_warnings(output, probe_context.query_time, warnings),
         ))
     }