Merge ticket 042 (sum/avg timestamp merge fix), resolve conflicts with 041

Rebase 042's Count-aggregation and windowed-merge test helpers onto 041's
Result-based evaluate_promql signature and 037's declared-param-by-name
metric shape.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o

src/engine.rs

@@ -1267,14 +1267,30 @@ fn windowed_rate_series(points: &[(i64, f64)], range_ms: i64) -> Vec<(i64, f64)>
         .collect()
 }
 
-/// Merges several series by exact timestamp match: values sharing a timestamp are summed (or averaged).
+/// Fixed window width (ms) used by `merge_series_windowed` to combine multiple raw-occurrence
-/// This is the plain `sum()`/`avg()` behavior — appropriate when the underlying points are occurrence
+/// sub-series into one time series — the same fixed-width bucketing style `histogram()`/the
-/// markers rather than a continuous reading, so there's nothing to carry forward between timestamps.
+/// `Timeline` query already use elsewhere in this file, just applied along the time axis instead of
-fn merge_series_exact(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
+/// the value axis. Occurrence timestamps across different label groups (e.g. different hosts serving
+/// the same path) essentially never land on the same millisecond, so summing/averaging only requires
+/// matching *this* window to combine into "total over time" rather than going sparse (ticket 042).
+const MERGE_WINDOW_MS: i64 = 1_000;
+
+/// Merges several series into one combined time series (ticket 042). A single sub-series is returned
+/// untouched — there's only one contributor, so there's nothing to merge and no reason to lose
+/// resolution. With two or more sub-series, points are bucketed into fixed `MERGE_WINDOW_MS`-wide
+/// windows (rather than requiring an exact-millisecond timestamp match) before summing/averaging
+/// within each window, so raw occurrence markers from different label groups combine into a
+/// meaningful "total over time" instead of each window reflecting only whichever sub-series happened
+/// to report at that exact millisecond.
+fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
+    if sub_series.len() <= 1 {
+        return sub_series.first().cloned().unwrap_or_default();
+    }
     let mut grouped: BTreeMap<i64, Vec<f64>> = BTreeMap::new();
     for points in sub_series {
         for (timestamp_ms, value) in points {
-            grouped.entry(*timestamp_ms).or_default().push(*value);
+            let window_start = timestamp_ms.div_euclid(MERGE_WINDOW_MS) * MERGE_WINDOW_MS;
+            grouped.entry(window_start).or_default().push(*value);
         }
     }
     grouped
@@ -1289,7 +1305,7 @@ fn merge_series_exact(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64,
 
 /// Merges several already-windowed rate series by sweeping the union of their timestamps, carrying each
 /// sub-series' last-known value forward (0.0 before its first point) so every step reflects every
-/// sub-series' most recent reading — unlike `merge_series_exact`, which only combines exact-timestamp matches.
+/// sub-series' most recent reading — unlike `merge_series_windowed`, which buckets by fixed time windows instead.
 fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
     let mut all_timestamps: std::collections::BTreeSet<i64> = std::collections::BTreeSet::new();
     for points in sub_series {
@@ -1398,7 +1414,7 @@ fn evaluate_promql(
             groups
                 .into_iter()
                 .map(|(group_labels, sub_series)| {
-                    let points = merge_series_exact(&sub_series, false);
+                    let points = merge_series_windowed(&sub_series, false);
                     PromqlSeriesDto { labels: group_labels, points }
                 })
                 .collect()
@@ -1426,7 +1442,7 @@ fn evaluate_promql(
                     let points = if rate_range_ms.is_some() {
                         merge_series_locf(&sub_series, is_avg)
                     } else {
-                        merge_series_exact(&sub_series, is_avg)
+                        merge_series_windowed(&sub_series, is_avg)
                     };
                     PromqlSeriesDto { labels: group_labels, points }
                 })
@@ -2081,6 +2097,32 @@ mod tests {
         .unwrap();
     }
 
+    fn declare_requests_by_host_metric(index: &mut BTreeMap<IndexKey, DataValue>) {
+        declare_test_param(
+            index,
+            "path",
+            r"request path=(?P<path>\S+) host=(?P<host>\S+)",
+            "path",
+            MetricPrimitiveKind::String,
+        );
+        declare_test_param(
+            index,
+            "host",
+            r"request path=(?P<path>\S+) host=(?P<host>\S+)",
+            "host",
+            MetricPrimitiveKind::String,
+        );
+        declare_metric(
+            index,
+            metric_request(
+                "requests",
+                r"request path=(?P<path>\S+) host=(?P<host>\S+)",
+                vec!["path", "host"],
+            ),
+        )
+        .unwrap();
+    }
+
     #[test]
     fn sum_over_string_kind_metric_is_rejected_with_an_actionable_error() {
         let mut index = BTreeMap::new();
@@ -2160,6 +2202,68 @@ mod tests {
         assert!(series.is_ok());
     }
 
+    #[test]
+    fn sum_by_merges_staggered_raw_timestamps_across_label_groups_instead_of_going_sparse() {
+        // Regression test for ticket 042: `sum(requests) by (path)` groups together several
+        // underlying (path, host) series. Their raw occurrence timestamps are staggered (as real
+        // occurrence timestamps essentially always are), so a merge that only combines exact-
+        // timestamp matches (the old `merge_series_exact` behavior) would leave each point
+        // reflecting just one host instead of the combined total across hosts.
+        let mut index = BTreeMap::new();
+        declare_requests_by_host_metric(&mut index);
+        ingest_log(&mut index, 0, "request path=/a host=web1");
+        ingest_log(&mut index, 1, "request path=/a host=web2");
+        ingest_log(&mut index, 2, "request path=/a host=web3");
+
+        let parsed = promql::parse_query("sum(requests) by (path)").unwrap();
+        let series = evaluate_promql(&index, &parsed, None, None).unwrap();
+
+        assert_eq!(series.len(), 1);
+        assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a"));
+        // The three staggered per-host occurrences merge into one combined-total point rather than
+        // three separate points each showing a lone host's count of 1.
+        let total: f64 = series[0].points.iter().map(|(_, v)| v).sum();
+        assert_eq!(total, 3.0, "expected all three staggered occurrences to sum to 3.0, got {:?}", series[0].points);
+        assert!(
+            series[0].points.len() < 3,
+            "expected staggered timestamps to be bucketed together rather than staying sparse/near-empty, got {:?}",
+            series[0].points
+        );
+    }
+
+    #[test]
+    fn avg_by_merges_staggered_raw_timestamps_across_label_groups() {
+        let mut index = BTreeMap::new();
+        declare_requests_by_host_metric(&mut index);
+        ingest_log(&mut index, 0, "request path=/a host=web1");
+        ingest_log(&mut index, 1, "request path=/a host=web2");
+
+        let parsed = promql::parse_query("avg(requests) by (path)").unwrap();
+        let series = evaluate_promql(&index, &parsed, None, None).unwrap();
+
+        assert_eq!(series.len(), 1);
+        assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a"));
+        // Both occurrences (value 1.0 each) land in the same merged bucket, so the average stays 1.0
+        // rather than reporting two separate points that each only saw one host.
+        assert_eq!(series[0].points, vec![(0, 1.0)]);
+    }
+
+    #[test]
+    fn sum_by_still_gives_a_single_label_group_its_own_line_at_full_resolution() {
+        // Existing single-label-group behavior (only one underlying series feeds each output group)
+        // must stay untouched by the multi-series windowing fix.
+        let mut index = BTreeMap::new();
+        declare_requests_metric(&mut index);
+        ingest_log(&mut index, 0, "request path=/a");
+        ingest_log(&mut index, 1, "request path=/a");
+
+        let parsed = promql::parse_query("sum(requests) by (path)").unwrap();
+        let series = evaluate_promql(&index, &parsed, None, None).unwrap();
+
+        assert_eq!(series.len(), 1);
+        assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]);
+    }
+
     #[test]
     fn bare_rate_produces_a_windowed_point_per_timestamp_not_a_single_point() {
         let mut index = BTreeMap::new();