fix(querier): keep the runningDiff lookback step in the cache

The cache window is the query window widened by the step the statement
reads before it, so a hit carries that step too.

Assisted-by: Claude Fable 5.1

pkg/querier/bucket_cache.go

@@ -60,7 +60,7 @@ func (bc *bucketCache) GetMissRanges(
 ) (cached *qbtypes.Result, missing []*qbtypes.TimeRange) {
 
 	// Get query window
-	startMs, endMs := q.Window()
+	startMs, endMs := cacheWindow(q, uint64(step.Milliseconds()))
 
 	bc.logger.DebugContext(ctx, "getting miss ranges", slog.String("fingerprint", q.Fingerprint()), slog.Uint64("start", startMs), slog.Uint64("end", endMs))
 
@@ -127,7 +127,7 @@ func (bc *bucketCache) GetMissRanges(
 // Put stores fresh query results in the cache.
 func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result) {
 	// Get query window
-	startMs, endMs := q.Window()
+	startMs, endMs := cacheWindow(q, uint64(step.Milliseconds()))
 
 	// Calculate the flux boundary - data after this point should not be cached
 	currentMs := uint64(time.Now().UnixMilli())
@@ -152,7 +152,8 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 	}
 
 	// Trim the result to exclude data within flux interval
-	trimmedResult := bc.trimResultToFluxBoundary(fresh, cachableEndMs)
+	requestStartMs, _ := q.Window()
+	trimmedResult := bc.trimResultToFluxBoundary(fresh, requestStartMs, cachableEndMs)
 	if trimmedResult == nil {
 		// Result type is not cacheable (raw or scalar)
 		return
@@ -222,6 +223,30 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 	}
 }
 
+// lookbackSteps is how many steps before its window a query's statement
+// reads. runningDiff drops its first point, so the metrics statement reads
+// one step before the window to give the first interval a difference.
+func lookbackSteps(q qbtypes.Query) int {
+	bq, ok := q.(*builderQuery[qbtypes.MetricAggregation])
+	if !ok {
+		return 0
+	}
+	for _, fn := range bq.spec.Functions {
+		if fn.Name == qbtypes.FunctionNameRunningDiff {
+			return 1
+		}
+	}
+	return 0
+}
+
+// cacheWindow is the query window widened by its lookback, so a hit carries
+// the steps the statement reads before the window.
+func cacheWindow(q qbtypes.Query, stepMs uint64) (uint64, uint64) {
+	startMs, endMs := q.Window()
+	lookbackMs := uint64(lookbackSteps(q)) * stepMs
+	return startMs - min(lookbackMs, startMs), endMs
+}
+
 // generateCacheKey names the cache entry holding one grid cell of a query.
 func (bc *bucketCache) generateCacheKey(q qbtypes.Query, cellIdx uint64) string {
 	return fmt.Sprintf("v6:query:%s:cellidx:%d", q.Fingerprint(), cellIdx)
@@ -760,7 +785,7 @@ func (bc *bucketCache) deduplicateWarnings(warnings []string) []string {
 }
 
 // trimResultToFluxBoundary trims the result to exclude data points beyond the flux boundary.
-func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, fluxBoundary uint64) *qbtypes.Result {
+func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, requestStartMs, fluxBoundary uint64) *qbtypes.Result {
 	trimmedResult := &qbtypes.Result{
 		Type:     result.Type,
 		Stats:    result.Stats,
@@ -790,8 +815,10 @@ func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, fluxBoun
 
 					// Filter values to exclude those beyond flux boundary and partial values
 					for _, value := range series.Values {
-						// Skip partial values - they cannot be cached
-						if value.Partial {
+						// Skip partial values - they cannot be cached. A lookback
+						// step before the request window is flagged partial
+						// against that window but is a whole step, so it stays.
+						if value.Partial && uint64(value.Timestamp) >= requestStartMs {
 							continue
 						}
 						if uint64(value.Timestamp) <= fluxBoundary {

pkg/querier/heatmap_test.go

@@ -71,7 +71,7 @@ func TestTrimResultToFluxBoundaryKeepsTheHeatmapAxis(t *testing.T) {
 		},
 	}
 
-	trimmed := cache.trimResultToFluxBoundary(result, 1710000060000)
+	trimmed := cache.trimResultToFluxBoundary(result, 0, 1710000060000)
 
 	tsData, ok := trimmed.Value.(*qbtypes.TimeSeriesData)
 	require.True(t, ok)

pkg/querier/querier.go

@@ -814,7 +814,7 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 
 	// If entire range is missing, execute normally
 	if cachedResult == nil && len(missingRanges) == 1 {
-		startMs, endMs := query.Window()
+		startMs, endMs := cacheWindow(query, uint64(step.Milliseconds()))
 		if missingRanges[0].From == startMs && missingRanges[0].To == endMs {
 			sem <- struct{}{}
 			result, err := query.Execute(ctx)
@@ -898,6 +898,11 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 
 	// Merge cached and fresh results
 	mergedResult := q.mergeResults(cachedResult, freshResults)
+	// A gap statement at the start of the cache window reads its own lookback
+	// step before it, which the request's statement would not read.
+	if cacheStartMs, _ := cacheWindow(query, uint64(step.Milliseconds())); lookbackSteps(query) > 0 {
+		dropValuesBefore(mergedResult, cacheStartMs)
+	}
 	mergedResult.Stats.RowsScanned += totalStats.RowsScanned
 	mergedResult.Stats.BytesScanned += totalStats.BytesScanned
 	mergedResult.Stats.DurationMS += totalStats.DurationMS
@@ -908,6 +913,18 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 	return mergedResult, nil
 }
 
+func dropValuesBefore(result *qbtypes.Result, fromMs uint64) {
+	data, ok := result.Value.(*qbtypes.TimeSeriesData)
+	if !ok || data == nil {
+		return
+	}
+	for _, aggBucket := range data.Aggregations {
+		for _, series := range aggBucket.Series {
+			series.Values = slices.DeleteFunc(series.Values, func(value *qbtypes.TimeSeriesValue) bool { return uint64(value.Timestamp) < fromMs })
+		}
+	}
+}
+
 // createRangedQuery creates a copy of the query with a different time range.
 func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query, timeRange qbtypes.TimeRange) qbtypes.Query {
 	// this is called in a goroutine, so we create a copy of the query to avoid race conditions

tests/integration/tests/queriercommon/13_cache_running_diff.py

@@ -0,0 +1,69 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+from uuid import uuid4
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import Metrics
+from fixtures.querier import (
+    build_builder_query,
+    build_function,
+    get_all_series,
+    index_series_by_label,
+    make_query_request,
+)
+
+MINUTE = timedelta(minutes=1)
+
+
+def test_running_diff_full_hit_matches_no_cache(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    """
+    Setup:
+    A gauge that grows by 10 every minute over four minutes.
+
+    Tests:
+    runningDiff over the last three of those minutes, requested twice. The
+    metrics statement reads one step before the window so the first
+    interval has a difference. The second request, a full cache hit, must
+    return the same three differences as the request with noCache.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # the cache outlives the run, so a fixed name would serve the previous
+    # run's points back to this one
+    metric_name = f"cache_running_diff_{uuid4().hex[:8]}"
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "svc-a"},
+                timestamp=now - (14 - i) * MINUTE,
+                value=100.0 + 10 * i,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for i in range(4)
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query("A", metric_name, "avg", "avg", group_by=["service"], functions=[build_function("runningDiff")])
+    start_ms = int((now - 13 * MINUTE).timestamp() * 1000)
+    end_ms = int((now - 10 * MINUTE).timestamp() * 1000)
+
+    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+    assert warm.status_code == HTTPStatus.OK, warm.text
+    cached = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+    assert cached.status_code == HTTPStatus.OK, cached.text
+    fresh = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=True)
+    assert fresh.status_code == HTTPStatus.OK, fresh.text
+
+    fresh_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(fresh.json(), "A"), "service")["svc-a"]["values"]}
+    assert sorted(fresh_points.values()) == [10, 10, 10], fresh_points
+
+    cached_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(cached.json(), "A"), "service")["svc-a"]["values"]}
+    assert cached_points == fresh_points