Merge branch 'main' into nv/exec-stats

.github/workflows/build-community.yaml

@@ -63,7 +63,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
       GO_NAME: signoz-community
       GO_INPUT_ARTIFACT_CACHE_KEY: community-jsbuild-${{ github.sha }}
       GO_INPUT_ARTIFACT_PATH: frontend/build

.github/workflows/build-enterprise.yaml

@@ -89,7 +89,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
       GO_INPUT_ARTIFACT_CACHE_KEY: enterprise-jsbuild-${{ github.sha }}
       GO_INPUT_ARTIFACT_PATH: frontend/build
       GO_BUILD_CONTEXT: ./cmd/enterprise

.github/workflows/build-staging.yaml

@@ -95,7 +95,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
       GO_INPUT_ARTIFACT_CACHE_KEY: staging-jsbuild-${{ github.sha }}
       GO_INPUT_ARTIFACT_PATH: frontend/build
       GO_BUILD_CONTEXT: ./cmd/enterprise

.github/workflows/goci.yaml

@@ -22,7 +22,7 @@ jobs:
     with:
       PRIMUS_REF: main
       GO_TEST_CONTEXT: ./...
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
   fmt:
     if: |
       github.event_name == 'merge_group' ||
@@ -32,7 +32,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
   lint:
     if: |
       github.event_name == 'merge_group' ||
@@ -42,7 +42,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
   deps:
     if: |
       github.event_name == 'merge_group' ||
@@ -52,7 +52,7 @@ jobs:
     secrets: inherit
     with:
       PRIMUS_REF: main
-      GO_VERSION: 1.25
+      GO_VERSION: 1.24
   semconv-generated:
     if: |
       github.event_name == 'merge_group' ||
@@ -65,7 +65,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: check-semconv-generated-files
         run: make semconv-check
   build:
@@ -80,7 +80,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: qemu-install
         uses: docker/setup-qemu-action@v3
       - name: aarch64-install
@@ -116,7 +116,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: generate-openapi
         run: |
           go run cmd/enterprise/*.go generate openapi
@@ -133,7 +133,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: generate-authz
         run: |
           go run cmd/enterprise/*.go generate authz
@@ -150,7 +150,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: generate-web-settings
         run: |
           go run cmd/enterprise/*.go generate config web-settings
@@ -167,7 +167,7 @@ jobs:
       - name: go-install
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: generate-transaction-groups
         run: |
           go run cmd/enterprise/*.go generate config transaction-groups

.github/workflows/gor-signoz-community.yaml

@@ -66,7 +66,7 @@ jobs:
       - name: setup-go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: cross-compilation-tools
         if: matrix.os == 'ubuntu-latest'
         run: |
@@ -130,7 +130,7 @@ jobs:
       - name: setup-go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
 
       # copy the caches from build
       - name: get-sha

.github/workflows/gor-signoz.yaml

@@ -75,7 +75,7 @@ jobs:
       - name: setup-go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
       - name: cross-compilation-tools
         if: matrix.os == 'ubuntu-latest'
         run: |
@@ -138,7 +138,7 @@ jobs:
       - name: setup-go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.25"
+          go-version: "1.24"
 
       # copy the caches from build
       - name: get-sha

go.mod

@@ -20,7 +20,6 @@ require (
 	github.com/evanphx/json-patch/v5 v5.9.11
 	github.com/gin-gonic/gin v1.11.0
 	github.com/go-co-op/gocron v1.30.1
-	github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875
 	github.com/go-openapi/runtime v0.29.2
 	github.com/go-openapi/strfmt v0.26.1
 	github.com/go-playground/validator/v10 v10.27.0

go.sum

@@ -328,8 +328,6 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
 github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
-github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875 h1:XGPtjxqp0FS8jj1Z7QGcSOSgyq1fUPjNyJXvi3KfI4A=
-github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875/go.mod h1:uNVvRXArCGbZ508SxYYTC5v1JWoz2voff5pm25jU1Ok=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=

pkg/alertmanager/alertmanagernotify/email/email_test.go

@@ -11,7 +11,6 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
-	"net/textproto"
 	"net/url"
 	"os"
 	"strconv"
@@ -707,9 +706,7 @@ func TestEmailRejected(t *testing.T) {
 
 	// Send the alert to mock SMTP server.
 	retry, err := e.Notify(context.Background(), firingAlert)
-	var smtpErr *textproto.Error
-	require.ErrorAs(t, err, &smtpErr)
-	assert.Equal(t, 501, smtpErr.Code)
+	require.ErrorContains(t, err, "501 5.5.4 Rejected!")
 	assert.True(t, retry)
 	require.NoError(t, srv.Shutdown(ctx))
 

pkg/http/render/render.go

@@ -5,12 +5,9 @@ import (
 	"net/http"
 	"strconv"
 
-	jsonv2 "github.com/go-json-experiment/json"
-	jsonv1 "github.com/go-json-experiment/json/v1"
+	"github.com/SigNoz/signoz/pkg/errors"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/tidwall/gjson"
-
-	"github.com/SigNoz/signoz/pkg/errors"
 )
 
 const (
@@ -32,11 +29,6 @@ type ErrorResponse struct {
 }
 
 func Success(rw http.ResponseWriter, httpCode int, data interface{}) {
-	if _, ok := data.(jsonv2.MarshalerTo); ok {
-		successStream(rw, httpCode, data)
-		return
-	}
-
 	body, err := json.Marshal(&SuccessResponse{Status: StatusSuccess.s, Data: data})
 	if err != nil {
 		Error(rw, err)
@@ -142,32 +134,3 @@ func Error(rw http.ResponseWriter, cause error) {
 	rw.WriteHeader(httpCode)
 	_, _ = rw.Write(body)
 }
-
-// successStream writes the body as it is encoded. It can still render an error if encoding fails
-// before the first write; after that the status is sent and a failure leaves the body truncated.
-func successStream(rw http.ResponseWriter, httpCode int, data any) {
-	if httpCode == 0 {
-		httpCode = http.StatusOK
-	}
-
-	writer := &successStreamWriter{rw: rw, httpCode: httpCode}
-	if err := jsonv2.MarshalWrite(writer, &SuccessResponse{Status: StatusSuccess.s, Data: data}, jsonv1.DefaultOptionsV1()); err != nil && !writer.started {
-		Error(rw, err)
-	}
-}
-
-// successStreamWriter sends the headers on the first write.
-type successStreamWriter struct {
-	rw       http.ResponseWriter
-	httpCode int
-	started  bool
-}
-
-func (writer *successStreamWriter) Write(p []byte) (int, error) {
-	if !writer.started {
-		writer.started = true
-		writer.rw.Header().Set("Content-Type", "application/json")
-		writer.rw.WriteHeader(writer.httpCode)
-	}
-	return writer.rw.Write(p)
-}

pkg/http/render/render_test.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/go-json-experiment/json/jsontext"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -211,26 +210,3 @@ func TestErrorRetryAfterHeader(t *testing.T) {
 		})
 	}
 }
-
-type streamedData []int
-
-func (data streamedData) MarshalJSONTo(enc *jsontext.Encoder) error {
-	if err := enc.WriteToken(jsontext.BeginArray); err != nil {
-		return err
-	}
-	for _, v := range data {
-		if err := enc.WriteToken(jsontext.Int(int64(v))); err != nil {
-			return err
-		}
-	}
-	return enc.WriteToken(jsontext.EndArray)
-}
-
-func TestSuccess_Stream(t *testing.T) {
-	recorder := httptest.NewRecorder()
-	Success(recorder, http.StatusAccepted, streamedData{1, 2, 3})
-
-	assert.Equal(t, http.StatusAccepted, recorder.Code)
-	assert.Equal(t, "application/json", recorder.Header().Get("Content-Type"))
-	assert.Equal(t, `{"status":"success","data":[1,2,3]}`, recorder.Body.String())
-}

pkg/querier/bucket_cache.go

@@ -26,8 +26,9 @@ const bucketCacheCellSteps = 60
 type BucketCache interface {
 	// cached portion + list of gaps to fetch
 	GetMissRanges(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step) (cached *qbtypes.Result, missing []*qbtypes.TimeRange)
-	// store fresh buckets for future hits
-	Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result)
+	// store fresh buckets for future hits; only the ranges in the map were
+	// queried for fresh, the rest of it came from the cache
+	Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats)
 }
 
 // bucketCache implements the BucketCache interface.
@@ -116,7 +117,7 @@ func (bc *bucketCache) GetMissRanges(
 	}
 
 	// Merge buckets into a single result
-	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, cachedWarnings)
+	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, cachedWarnings, startMs, endMs)
 
 	// Filter the merged result to only include values within the requested time range
 	mergedResult = bc.filterResultToTimeRange(mergedResult, q, startMs, endMs, stepMs)
@@ -125,7 +126,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) {
+func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) {
 	// Get query window
 	startMs, endMs := q.Window()
 
@@ -188,7 +189,7 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 
 	// Convert trimmed result to buckets, one per grid cell it spans
 	cellWidthMs := bucketCacheCellSteps * stepMs
-	freshBuckets := bc.resultToBuckets(ctx, trimmedResult, cachableStartMs, cachableEndMs, cellWidthMs)
+	freshBuckets := bc.resultToBuckets(ctx, trimmedResult, cachableStartMs, cachableEndMs, cellWidthMs, execStatsForQueriedTimeRanges)
 
 	cellKeys := make([]string, len(freshBuckets))
 	cellEntries := make([]cachetypes.Cacheable, len(freshBuckets))
@@ -210,7 +211,7 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 			cellEntry = cellEntries[i].(*qbtypes.CachedData)
 		}
 
-		cellBuckets, changed := bc.mergeFreshBucketIntoCell(ctx, cellEntry.Buckets, freshBucket)
+		cellBuckets, changed := bc.mergeFreshBucketIntoCell(ctx, cellEntry.Buckets, freshBucket, execStatsForQueriedTimeRanges)
 		if !changed {
 			continue
 		}
@@ -463,7 +464,7 @@ func (bc *bucketCache) filterRelevantBuckets(buckets []*qbtypes.CachedBucket, st
 }
 
 // mergeBuckets combines multiple cached buckets into a single result.
-func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.CachedBucket, warnings []string) *qbtypes.Result {
+func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.CachedBucket, warnings []string, startMs, endMs uint64) *qbtypes.Result {
 	if len(buckets) == 0 {
 		return &qbtypes.Result{}
 	}
@@ -471,12 +472,13 @@ func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.Cach
 	// All buckets should have the same type
 	resultType := buckets[0].Type
 
-	// Aggregate stats
+	// A bucket can reach past the window; only the part inside it is served
 	var totalStats qbtypes.ExecStats
 	for _, bucket := range buckets {
-		totalStats.RowsScanned += bucket.Stats.RowsScanned
-		totalStats.BytesScanned += bucket.Stats.BytesScanned
-		totalStats.DurationMS += bucket.Stats.DurationMS
+		servedStats := bucket.ProrateStatsToWindow(startMs, endMs)
+		totalStats.RowsScanned += servedStats.RowsScanned
+		totalStats.BytesScanned += servedStats.BytesScanned
+		totalStats.DurationMS += servedStats.DurationMS
 	}
 
 	// Merge values based on type
@@ -673,7 +675,7 @@ func (bc *bucketCache) isEmptyResult(result *qbtypes.Result) (isEmpty bool, isFi
 }
 
 // resultToBuckets converts a query result into time-based buckets.
-func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Result, startMs, endMs, cellWidthMs uint64) []*qbtypes.CachedBucket {
+func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Result, startMs, endMs, cellWidthMs uint64, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) []*qbtypes.CachedBucket {
 	// Check if result is empty
 	isEmpty, isFiltered := bc.isEmptyResult(result)
 
@@ -685,7 +687,7 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
 	}
 
 	// Empty filtered results still become buckets, so the range is not re-queried
-	perCellBuckets, err := result.SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs)
+	perCellBuckets, err := result.SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs, execStatsForQueriedTimeRanges)
 	if err != nil {
 		bc.logger.ErrorContext(ctx, "failed to convert result to buckets", errors.Attr(err))
 		return nil
@@ -697,7 +699,7 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
 // and reports whether the cell changed: a bucket already covering the fresh
 // one leaves it as is, buckets inside the fresh one are replaced, buckets
 // overlapping or edge to edge with it are unioned, and the rest are kept apart.
-func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets []*qbtypes.CachedBucket, freshBucket *qbtypes.CachedBucket) ([]*qbtypes.CachedBucket, bool) {
+func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets []*qbtypes.CachedBucket, freshBucket *qbtypes.CachedBucket, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) ([]*qbtypes.CachedBucket, bool) {
 	keptBuckets := make([]*qbtypes.CachedBucket, 0, len(cellBuckets)+1)
 	touchingBuckets := []*qbtypes.CachedBucket{}
 	for _, cellBucket := range cellBuckets {
@@ -710,7 +712,7 @@ func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets
 			keptBuckets = append(keptBuckets, cellBucket)
 		}
 	}
-	keptBuckets = append(keptBuckets, bc.unionBuckets(ctx, freshBucket, touchingBuckets))
+	keptBuckets = append(keptBuckets, bc.unionBuckets(ctx, freshBucket, touchingBuckets, execStatsForQueriedTimeRanges))
 
 	slices.SortFunc(keptBuckets, func(a, b *qbtypes.CachedBucket) int {
 		return cmp.Compare(a.StartMs, b.StartMs)
@@ -721,7 +723,16 @@ func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets
 // unionBuckets widens the fresh bucket by the buckets it touches. Buckets
 // inside it are replaced outright, so only a genuine partial overlap decodes
 // anything.
-func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.CachedBucket, touchingBuckets []*qbtypes.CachedBucket) *qbtypes.CachedBucket {
+func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.CachedBucket, touchingBuckets []*qbtypes.CachedBucket, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) *qbtypes.CachedBucket {
+	// Data queried again takes the new query's cost, which the fresh bucket
+	// carries; a touching bucket keeps its cost only for what was not queried again
+	for _, touchingBucket := range touchingBuckets {
+		keptStats := touchingBucket.ProrateStatsToUnqueried(execStatsForQueriedTimeRanges)
+		freshBucket.Stats.RowsScanned += keptStats.RowsScanned
+		freshBucket.Stats.BytesScanned += keptStats.BytesScanned
+		freshBucket.Stats.DurationMS += keptStats.DurationMS
+	}
+
 	bucketsToMerge := []*qbtypes.CachedBucket{freshBucket}
 	for _, touchingBucket := range touchingBuckets {
 		if !freshBucket.Contains(touchingBucket) {
@@ -742,14 +753,11 @@ func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.Ca
 		EndMs:   freshBucket.EndMs,
 		Type:    freshBucket.Type,
 		Value:   value,
-		Stats:   qbtypes.ExecStats{StepIntervals: freshBucket.Stats.StepIntervals},
+		Stats:   freshBucket.Stats,
 	}
 	for _, mergedBucket := range bucketsToMerge {
 		unionBucket.StartMs = min(unionBucket.StartMs, mergedBucket.StartMs)
 		unionBucket.EndMs = max(unionBucket.EndMs, mergedBucket.EndMs)
-		unionBucket.Stats.RowsScanned += mergedBucket.Stats.RowsScanned
-		unionBucket.Stats.BytesScanned += mergedBucket.Stats.BytesScanned
-		unionBucket.Stats.DurationMS += mergedBucket.Stats.DurationMS
 	}
 	return unionBucket
 }

pkg/querier/bucket_cache_bench_test.go

@@ -30,7 +30,7 @@ func BenchmarkBucketCache_GetMissRanges(b *testing.B) {
 			endMs:       uint64((i + 1) * 10000),
 		}
 		result := createBenchmarkResult(query.startMs, query.endMs, 1000)
-		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 	}
 
 	// Create test queries with varying cache hit patterns
@@ -121,7 +121,7 @@ func BenchmarkBucketCache_Put(b *testing.B) {
 
 			for i := 0; i < b.N; i++ {
 				for j := 0; j < tc.numQueries; j++ {
-					bc.Put(ctx, orgID, queries[j], qbtypes.Step{Duration: 1000 * time.Millisecond}, results[j])
+					bc.Put(ctx, orgID, queries[j], qbtypes.Step{Duration: 1000 * time.Millisecond}, results[j], map[qbtypes.TimeRange]qbtypes.ExecStats{{From: queries[j].startMs, To: queries[j].endMs}: results[j].Stats})
 				}
 			}
 		})
@@ -259,7 +259,7 @@ func BenchmarkBucketCache_ConcurrentOperations(b *testing.B) {
 			endMs:       uint64((i + 1) * 10000),
 		}
 		result := createBenchmarkResult(query.startMs, query.endMs, 1000)
-		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 	}
 
 	b.ResetTimer()
@@ -284,7 +284,7 @@ func BenchmarkBucketCache_ConcurrentOperations(b *testing.B) {
 					endMs:       uint64((i + 1) * 10000),
 				}
 				result := createBenchmarkResult(query.startMs, query.endMs, 1000)
-				bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+				bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 			case 2: // Partial read
 				query := &mockQuery{
 					fingerprint: fmt.Sprintf("concurrent-query-%d", i%100),

pkg/querier/bucket_cache_step_test.go

@@ -59,7 +59,7 @@ func TestBucketCacheStepAlignment(t *testing.T) {
 	}
 
 	// Put result in cache
-	bc.Put(ctx, orgID, query, step, result)
+	bc.Put(ctx, orgID, query, step, result, nil)
 
 	// Get cached data
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
@@ -109,7 +109,7 @@ func TestBucketCacheNoStepInterval(t *testing.T) {
 	}
 
 	// Should cache the entire range when step is 0
-	bc.Put(ctx, orgID, query, step, result)
+	bc.Put(ctx, orgID, query, step, result, nil)
 
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
 	assert.NotNil(t, cached)

pkg/querier/bucket_cache_test.go

@@ -156,7 +156,7 @@ func TestBucketCache_Put_And_Get(t *testing.T) {
 	}
 
 	// Store in cache
-	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 
 	// Wait a bit for cache to be written
 	time.Sleep(10 * time.Millisecond)
@@ -189,7 +189,7 @@ func TestBucketCache_PartialHit(t *testing.T) {
 		Type:  qbtypes.RequestTypeTimeSeries,
 		Value: createTestTimeSeries("A", 1000, 3000, 1000),
 	}
-	bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1)
+	bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1, nil)
 
 	// Wait for cache write
 	time.Sleep(10 * time.Millisecond)
@@ -225,7 +225,7 @@ func TestBucketCache_MultipleBuckets(t *testing.T) {
 	bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 100 * time.Millisecond}, &qbtypes.Result{
 		Type:  qbtypes.RequestTypeTimeSeries,
 		Value: createTestTimeSeries("A", 1000, 2000, 100),
-	})
+	}, nil)
 
 	query2 := &mockQuery{
 		fingerprint: "test-query",
@@ -235,7 +235,7 @@ func TestBucketCache_MultipleBuckets(t *testing.T) {
 	bc.Put(context.Background(), valuer.UUID{}, query2, qbtypes.Step{Duration: 100 * time.Millisecond}, &qbtypes.Result{
 		Type:  qbtypes.RequestTypeTimeSeries,
 		Value: createTestTimeSeries("A", 3000, 4000, 100),
-	})
+	}, nil)
 
 	// Wait for cache writes
 	time.Sleep(10 * time.Millisecond)
@@ -280,7 +280,7 @@ func TestBucketCache_FluxInterval(t *testing.T) {
 	}
 
 	// This should not be cached due to flux interval
-	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 
 	// Wait a bit
 	time.Sleep(10 * time.Millisecond)
@@ -358,7 +358,7 @@ func TestBucketCache_MergeTimeSeriesResults(t *testing.T) {
 				{Series: []*qbtypes.TimeSeries{series1}},
 			},
 		},
-	})
+	}, nil)
 
 	// Cache second part
 	query2 := &mockQuery{
@@ -374,7 +374,7 @@ func TestBucketCache_MergeTimeSeriesResults(t *testing.T) {
 				{Series: []*qbtypes.TimeSeries{series2}},
 			},
 		},
-	})
+	}, nil)
 
 	// Wait for cache writes
 	time.Sleep(10 * time.Millisecond)
@@ -441,7 +441,7 @@ func TestBucketCache_RawData(t *testing.T) {
 		Value: rawData,
 	}
 
-	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	cached, missing := bc.GetMissRanges(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
@@ -481,7 +481,7 @@ func TestBucketCache_ScalarData(t *testing.T) {
 		Value: scalarData,
 	}
 
-	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	cached, missing := bc.GetMissRanges(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
@@ -509,7 +509,7 @@ func TestBucketCache_EmptyFingerprint(t *testing.T) {
 		Value: createTestTimeSeries("A", 1000, 5000, 1000),
 	}
 
-	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	// Should still be able to retrieve
@@ -564,7 +564,7 @@ func TestBucketCache_ConcurrentAccess(t *testing.T) {
 				Type:  qbtypes.RequestTypeTimeSeries,
 				Value: createTestTimeSeries(fmt.Sprintf("Q%d", id), query.startMs, query.endMs, 100),
 			}
-			bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 100 * time.Microsecond}, result)
+			bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 100 * time.Microsecond}, result, nil)
 			done <- true
 		}(i)
 	}
@@ -624,7 +624,7 @@ func TestBucketCache_GetMissRanges_FluxInterval(t *testing.T) {
 		},
 	}
 
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, cachedResult)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, cachedResult, nil)
 
 	// Get miss ranges
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
@@ -686,7 +686,7 @@ func TestBucketCache_Put_FluxIntervalTrimming(t *testing.T) {
 	}
 
 	// Put the result
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 
 	// Retrieve cached data
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
@@ -756,7 +756,7 @@ func TestBucketCache_Put_EntireRangeInFluxInterval(t *testing.T) {
 	}
 
 	// Put the result - should not cache anything
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 
 	// Try to get cached data - should have no cached data
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
@@ -862,7 +862,7 @@ func TestBucketCache_EmptyDataHandling(t *testing.T) {
 			}
 
 			// Put the result
-			bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, tt.result)
+			bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, tt.result, nil)
 
 			// Wait a bit for cache to be written
 			time.Sleep(10 * time.Millisecond)
@@ -927,7 +927,7 @@ func TestBucketCache_PartialValues(t *testing.T) {
 	}
 
 	// Put the result
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
 
 	// Wait for cache to be written
 	time.Sleep(10 * time.Millisecond)
@@ -997,7 +997,7 @@ func TestBucketCache_AllPartialValues(t *testing.T) {
 	}
 
 	// Put the result
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 
 	// Wait for cache to be written
 	time.Sleep(10 * time.Millisecond)
@@ -1058,7 +1058,7 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
 	}
 
 	// Cache the wide range
-	bc.Put(ctx, orgID, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1)
+	bc.Put(ctx, orgID, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	// Now query for a smaller range (2000-3500ms)
@@ -1125,7 +1125,7 @@ func TestBucketCache_PromQLKeepsTheValueAtTheWindowEnd(t *testing.T) {
 				}},
 			}},
 		},
-	})
+	}, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
@@ -1291,7 +1291,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
 		}
 
 		// Put the result
-		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 		time.Sleep(10 * time.Millisecond)
 
 		// Get cached data
@@ -1345,7 +1345,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
 		}
 
 		// Put the result
-		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 		time.Sleep(10 * time.Millisecond)
 
 		// Get cached data
@@ -1397,7 +1397,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
 		}
 
 		// Put the result
-		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+		bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 		time.Sleep(10 * time.Millisecond)
 
 		// Get cached data
@@ -1454,7 +1454,7 @@ func TestBucketCache_NoCache(t *testing.T) {
 	}
 
 	// Put the result in cache
-	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
+	bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
 	time.Sleep(10 * time.Millisecond)
 
 	// Verify data is cached
@@ -1475,11 +1475,12 @@ func TestBucketCachePutSlicesAtCellBoundaries(t *testing.T) {
 
 	// a 1s step gives 60s cells, so 30s to 150s spans three of them
 	query := &mockQuery{fingerprint: "sliced-query", startMs: 30000, endMs: 150000}
-	bc.Put(ctx, orgID, query, step, &qbtypes.Result{
+	result := &qbtypes.Result{
 		Type:  qbtypes.RequestTypeTimeSeries,
 		Value: createTestTimeSeries("A", 30000, 150000, 1000),
 		Stats: qbtypes.ExecStats{RowsScanned: 120, BytesScanned: 1200, DurationMS: 12},
-	})
+	}
+	bc.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: 30000, To: 150000}: result.Stats})
 
 	type bucketShape struct {
 		cell, startMs, endMs, rowsScanned uint64
@@ -1539,7 +1540,7 @@ func TestBucketCachePutAbsorbsWritesWithinACell(t *testing.T) {
 			name:                  "OverlappingWrites_BecomeOne",
 			firstWindow:           qbtypes.TimeRange{From: 1000, To: 4000},
 			secondWindow:          qbtypes.TimeRange{From: 3000, To: 6000},
-			expectedBuckets:       []bucketShape{{1000, 6000, 6}},
+			expectedBuckets:       []bucketShape{{1000, 6000, 5}},
 			expectedMissingRanges: []*qbtypes.TimeRange{},
 			expectedValueCount:    5,
 		},
@@ -1570,12 +1571,13 @@ func TestBucketCachePutAbsorbsWritesWithinACell(t *testing.T) {
 
 			for _, window := range []qbtypes.TimeRange{testCase.firstWindow, testCase.secondWindow} {
 				query := &mockQuery{fingerprint: "absorb-query", startMs: window.From, endMs: window.To}
-				bc.Put(ctx, orgID, query, step, &qbtypes.Result{
+				result := &qbtypes.Result{
 					Type:  qbtypes.RequestTypeTimeSeries,
 					Value: createTestTimeSeries("A", window.From, window.To, 1000),
-					// one row per second, so a bucket's stats read as the seconds it was scanned for
+					// one row per second, so a bucket's stats read as the seconds of data it holds
 					Stats: qbtypes.ExecStats{RowsScanned: (window.To - window.From) / 1000},
-				})
+				}
+				bc.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{window: result.Stats})
 			}
 
 			query := &mockQuery{
@@ -1609,7 +1611,7 @@ func TestBucketCachePutLeavesAnUnchangedCellAlone(t *testing.T) {
 		Type:     qbtypes.RequestTypeTimeSeries,
 		Value:    createTestTimeSeries("A", 1000, 10000, 1000),
 		Warnings: []string{"from the wide write"},
-	})
+	}, nil)
 
 	// a write already covered by the cell's bucket must not rewrite the entry,
 	// which shows as its warnings never arriving
@@ -1618,7 +1620,7 @@ func TestBucketCachePutLeavesAnUnchangedCellAlone(t *testing.T) {
 		Type:     qbtypes.RequestTypeTimeSeries,
 		Value:    createTestTimeSeries("A", 3000, 5000, 1000),
 		Warnings: []string{"from the inside write"},
-	})
+	}, nil)
 
 	var cellEntry qbtypes.CachedData
 	require.NoError(t, bc.cache.Get(ctx, orgID, bc.generateCacheKey(wide, 0), &cellEntry))

pkg/querier/postprocess.go

@@ -146,6 +146,8 @@ func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, res
 			if err != nil {
 				return nil, err
 			}
+			funcs := []qbtypes.Function{{Name: qbtypes.FunctionNameFillZero}}
+			funcs = q.prepareFillZeroArgsWithStep(funcs, req, stepInterval)
 			// empty time series if it doesn't exist
 			tsData, ok := typedResults[name].Value.(*qbtypes.TimeSeriesData)
 			if !ok {
@@ -168,9 +170,7 @@ func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, res
 				}
 			}
 
-			startMs := querybuilder.ToNanoSecs(req.Start) / 1_000_000
-			endMs := querybuilder.ToNanoSecs(req.End) / 1_000_000
-			tsData.SetZeroFill(int64(startMs), int64(endMs), stepInterval)
+			typedResults[name] = q.applyFunctions(typedResults[name], funcs)
 		}
 	}
 

pkg/querier/querier.go

@@ -823,7 +823,7 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 				return nil, err
 			}
 			// Store in cache for future use
-			q.bucketCache.Put(ctx, orgID, query, step, result)
+			q.bucketCache.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: startMs, To: endMs}: result.Stats})
 			return result, nil
 		}
 	}
@@ -879,16 +879,19 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 			if err != nil {
 				return nil, err
 			}
-			q.bucketCache.Put(ctx, orgID, query, step, result)
+			startMs, endMs := query.Window()
+			q.bucketCache.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: startMs, To: endMs}: result.Stats})
 			return result, nil
 		}
 	}
 
 	// Calculate total stats and filter out nil results
 	validResults := make([]*qbtypes.Result, 0, len(freshResults))
-	for _, result := range freshResults {
+	execStatsForQueriedTimeRanges := make(map[qbtypes.TimeRange]qbtypes.ExecStats, len(freshResults))
+	for i, result := range freshResults {
 		if result != nil {
 			validResults = append(validResults, result)
+			execStatsForQueriedTimeRanges[*missingRanges[i]] = result.Stats
 			totalStats.RowsScanned += result.Stats.RowsScanned
 			totalStats.BytesScanned += result.Stats.BytesScanned
 			totalStats.DurationMS += result.Stats.DurationMS
@@ -903,7 +906,7 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
 	mergedResult.Stats.DurationMS += totalStats.DurationMS
 
 	// Store merged result in cache
-	q.bucketCache.Put(ctx, orgID, query, step, mergedResult)
+	q.bucketCache.Put(ctx, orgID, query, step, mergedResult, execStatsForQueriedTimeRanges)
 
 	return mergedResult, nil
 }

pkg/types/querybuildertypes/querybuildertypesv5/cached.go

@@ -43,6 +43,23 @@ func (c *CachedBucket) Touches(other *CachedBucket) bool {
 	return c.StartMs <= other.EndMs && other.StartMs <= c.EndMs
 }
 
+// ProrateStatsToWindow returns the share of the stats for the part of the
+// bucket inside [startMs, endMs).
+func (c *CachedBucket) ProrateStatsToWindow(startMs, endMs uint64) ExecStats {
+	return c.Stats.prorate(overlapMs(c.StartMs, c.EndMs, startMs, endMs), c.EndMs-c.StartMs)
+}
+
+// ProrateStatsToUnqueried returns the share of the stats for the part of the
+// bucket no queried range covers.
+func (c *CachedBucket) ProrateStatsToUnqueried(execStatsForQueriedTimeRanges map[TimeRange]ExecStats) ExecStats {
+	widthMs := c.EndMs - c.StartMs
+	queriedMs := uint64(0)
+	for queried := range execStatsForQueriedTimeRanges {
+		queriedMs += overlapMs(c.StartMs, c.EndMs, queried.From, queried.To)
+	}
+	return c.Stats.prorate(widthMs-min(queriedMs, widthMs), widthMs)
+}
+
 // CachedData represents the full cached data for a query.
 type CachedData struct {
 	Buckets  []*CachedBucket `json:"buckets"`
@@ -88,7 +105,7 @@ func (c *CachedData) Cost() int64 {
 	return size
 }
 
-func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64) ([]*CachedBucket, error) {
+func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64, execStatsForQueriedTimeRanges map[TimeRange]ExecStats) ([]*CachedBucket, error) {
 	tsData, ok := r.Value.(*TimeSeriesData)
 	if !ok || tsData == nil {
 		return nil, errors.NewInternalf(errors.CodeInternal, "result of type %s cannot be bucketed", r.Type.StringValue())
@@ -111,18 +128,39 @@ func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64) (
 		if err != nil {
 			return nil, err
 		}
+		// A slice costs what was queried inside it; the rest of the result came
+		// out of the cache and is costed there already
+		sliceStats := ExecStats{StepIntervals: maps.Clone(r.Stats.StepIntervals)}
+		for queried, stats := range execStatsForQueriedTimeRanges {
+			queriedMs := overlapMs(from, to, queried.From, queried.To)
+			if queriedMs == 0 {
+				continue
+			}
+			share := stats.prorate(queriedMs, queried.To-queried.From)
+			sliceStats.RowsScanned += share.RowsScanned
+			sliceStats.BytesScanned += share.BytesScanned
+			sliceStats.DurationMS += share.DurationMS
+		}
 		cellBuckets = append(cellBuckets, &CachedBucket{
 			StartMs: from,
 			EndMs:   to,
 			Type:    r.Type,
 			Value:   value,
-			Stats:   r.Stats.prorate(to-from, endMs-startMs),
+			Stats:   sliceStats,
 		})
 		from = to
 	}
 	return cellBuckets, nil
 }
 
+// overlapMs is the length of [aFromMs, aToMs) ∩ [bFromMs, bToMs).
+func overlapMs(aFromMs, aToMs, bFromMs, bToMs uint64) uint64 {
+	if fromMs, toMs := max(aFromMs, bFromMs), min(aToMs, bToMs); fromMs < toMs {
+		return toMs - fromMs
+	}
+	return 0
+}
+
 // prorate returns the share of the stats that partMs out of wholeMs of the
 // scanned window accounts for.
 func (e ExecStats) prorate(partMs, wholeMs uint64) ExecStats {

pkg/types/querybuildertypes/querybuildertypesv5/resp.go

@@ -136,9 +136,6 @@ func (q *QueryRangeResponse) PrepareJSONSchema(schema *jsonschema.Schema) error
 type TimeSeriesData struct {
 	QueryName    string               `json:"queryName" description:"The name of the query that produced this result."`
 	Aggregations []*AggregationBucket `json:"aggregations" description:"One bucket per aggregation of the query. The order is not guaranteed; match buckets to aggregations by index."`
-
-	// zeroFill is applied only when encoding; in-memory series stay sparse.
-	zeroFill *zeroFill
 }
 
 // FilterToTimeRange copies the aggregation and series structure keeping only
@@ -565,6 +562,17 @@ func sanitizeValue(v any) any {
 	}
 }
 
+func (q QueryRangeResponse) MarshalJSON() ([]byte, error) {
+	type Alias QueryRangeResponse
+	return json.Marshal(&struct {
+		*Alias
+		Data any `json:"data"`
+	}{
+		Alias: (*Alias)(&q),
+		Data:  sanitizeValue(q.Data),
+	})
+}
+
 func (s ScalarData) MarshalJSON() ([]byte, error) {
 	type Alias ScalarData
 	sanitizedData := make([][]any, len(s.Data))
@@ -607,6 +615,36 @@ func (r RawRow) MarshalJSON() ([]byte, error) {
 	})
 }
 
+func (t TimeSeriesValue) MarshalJSON() ([]byte, error) {
+	type Alias TimeSeriesValue
+
+	var sanitizedValues any
+	if t.Values != nil {
+		sanitizedValues = sanitizeValue(t.Values)
+		// If original was empty slice, ensure we return empty slice not nil
+		if len(t.Values) == 0 {
+			sanitizedValues = []any{}
+		}
+	}
+
+	// a heatmap point's counts are spread across Values, so there is no one
+	// number Value could carry
+	var sanitizedValue any
+	if t.Values == nil {
+		sanitizedValue = sanitizeValue(t.Value)
+	}
+
+	return json.Marshal(&struct {
+		*Alias
+		Value  any `json:"value,omitempty"`
+		Values any `json:"values,omitempty"`
+	}{
+		Alias:  (*Alias)(&t),
+		Value:  sanitizedValue,
+		Values: sanitizedValues,
+	})
+}
+
 func (r RawData) MarshalJSON() ([]byte, error) {
 	type Alias RawData
 	return json.Marshal((*Alias)(&r))

pkg/types/querybuildertypes/querybuildertypesv5/resp_json.go

@@ -1,233 +0,0 @@
-package querybuildertypesv5
-
-import (
-	"math"
-
-	jsonv2 "github.com/go-json-experiment/json"
-	"github.com/go-json-experiment/json/jsontext"
-	jsonv1 "github.com/go-json-experiment/json/v1"
-)
-
-// SetZeroFill makes every series render one point per step over [startMs, endMs], writing 0 for
-// a step without a value, instead of storing those points. Points off the step grid are dropped.
-func (t *TimeSeriesData) SetZeroFill(startMs, endMs, stepMs int64) {
-	if stepMs <= 0 {
-		return
-	}
-	t.zeroFill = &zeroFill{start: startMs, end: endMs, step: stepMs}
-}
-
-func (q QueryRangeResponse) MarshalJSON() ([]byte, error) {
-	return jsonv2.Marshal(q, jsonv1.DefaultOptionsV1())
-}
-
-// MarshalJSONTo lets a time series response stream to the writer instead of being held in
-// memory whole.
-func (q QueryRangeResponse) MarshalJSONTo(enc *jsontext.Encoder) error {
-	type alias QueryRangeResponse
-	return jsonv2.MarshalEncode(enc, (*alias)(&q))
-}
-
-func (t TimeSeriesData) MarshalJSON() ([]byte, error) {
-	return jsonv2.Marshal(t, jsonv1.DefaultOptionsV1())
-}
-
-func (t TimeSeriesData) MarshalJSONTo(enc *jsontext.Encoder) error {
-	writer := tokenWriter{enc: enc}
-	writer.token(jsontext.BeginObject)
-	writer.token(jsontext.String("queryName"))
-	writer.token(jsontext.String(t.QueryName))
-	writer.token(jsontext.String("aggregations"))
-	if t.Aggregations == nil {
-		writer.token(jsontext.Null)
-	} else {
-		writer.token(jsontext.BeginArray)
-		for _, bucket := range t.Aggregations {
-			if writer.err != nil {
-				return writer.err
-			}
-			writer.aggregationBucket(bucket, t.zeroFill)
-		}
-		writer.token(jsontext.EndArray)
-	}
-	writer.token(jsontext.EndObject)
-	return writer.err
-}
-
-func (t TimeSeriesValue) MarshalJSON() ([]byte, error) {
-	return jsonv2.Marshal(t, jsonv1.DefaultOptionsV1())
-}
-
-func (t TimeSeriesValue) MarshalJSONTo(enc *jsontext.Encoder) error {
-	writer := tokenWriter{enc: enc}
-	writer.token(jsontext.BeginObject)
-	writer.token(jsontext.String("timestamp"))
-	writer.token(jsontext.Int(t.Timestamp))
-	if t.Partial {
-		writer.token(jsontext.String("partial"))
-		writer.token(jsontext.True)
-	}
-	// a heatmap point's counts are spread across Values, so there is no one
-	// number Value could carry
-	if t.Values == nil {
-		writer.token(jsontext.String("value"))
-		writer.token(sanitizedFloatToken(t.Value))
-	} else {
-		writer.token(jsontext.String("values"))
-		writer.token(jsontext.BeginArray)
-		for _, v := range t.Values {
-			writer.token(sanitizedFloatToken(v))
-		}
-		writer.token(jsontext.EndArray)
-	}
-	writer.token(jsontext.EndObject)
-	return writer.err
-}
-
-type zeroFill struct {
-	start int64
-	end   int64
-	step  int64
-}
-
-// tokenWriter keeps the first error so a sequence of writes needs a single check.
-type tokenWriter struct {
-	enc *jsontext.Encoder
-	err error
-}
-
-func (writer *tokenWriter) token(t jsontext.Token) {
-	if writer.err == nil {
-		writer.err = writer.enc.WriteToken(t)
-	}
-}
-
-func (writer *tokenWriter) value(v any) {
-	if writer.err == nil {
-		writer.err = jsonv2.MarshalEncode(writer.enc, v)
-	}
-}
-
-func (writer *tokenWriter) aggregationBucket(a *AggregationBucket, fill *zeroFill) {
-	if a == nil {
-		writer.token(jsontext.Null)
-		return
-	}
-
-	writer.token(jsontext.BeginObject)
-	writer.token(jsontext.String("index"))
-	writer.token(jsontext.Int(int64(a.Index)))
-	writer.token(jsontext.String("alias"))
-	writer.token(jsontext.String(a.Alias))
-	writer.token(jsontext.String("meta"))
-	writer.value(a.Meta)
-	writer.token(jsontext.String("series"))
-	writer.seriesList(a.Series, fill)
-	writer.optionalSeriesList("predictedSeries", a.PredictedSeries)
-	writer.optionalSeriesList("upperBoundSeries", a.UpperBoundSeries)
-	writer.optionalSeriesList("lowerBoundSeries", a.LowerBoundSeries)
-	writer.optionalSeriesList("anomalyScores", a.AnomalyScores)
-	writer.token(jsontext.EndObject)
-}
-
-func (writer *tokenWriter) optionalSeriesList(name string, series []*TimeSeries) {
-	if len(series) == 0 {
-		return
-	}
-	writer.token(jsontext.String(name))
-	writer.seriesList(series, nil)
-}
-
-func (writer *tokenWriter) seriesList(series []*TimeSeries, fill *zeroFill) {
-	if series == nil {
-		writer.token(jsontext.Null)
-		return
-	}
-
-	writer.token(jsontext.BeginArray)
-	for _, s := range series {
-		if writer.err != nil {
-			return
-		}
-		writer.series(s, fill)
-	}
-	writer.token(jsontext.EndArray)
-}
-
-func (writer *tokenWriter) series(s *TimeSeries, fill *zeroFill) {
-	if writer.err != nil {
-		return
-	}
-	if s == nil {
-		writer.token(jsontext.Null)
-		return
-	}
-
-	writer.token(jsontext.BeginObject)
-	if len(s.Labels) > 0 {
-		writer.token(jsontext.String("labels"))
-		writer.value(s.Labels)
-	}
-	writer.token(jsontext.String("values"))
-	switch {
-	case fill != nil:
-		writer.zeroFilledValues(s.Values, fill)
-	case s.Values == nil:
-		writer.token(jsontext.Null)
-	default:
-		writer.token(jsontext.BeginArray)
-		for _, v := range s.Values {
-			if writer.err != nil {
-				return
-			}
-			writer.point(v)
-		}
-		writer.token(jsontext.EndArray)
-	}
-	writer.token(jsontext.EndObject)
-}
-
-func (writer *tokenWriter) zeroFilledValues(values []*TimeSeriesValue, fill *zeroFill) {
-	byTimestamp := make(map[int64]*TimeSeriesValue, len(values))
-	for _, v := range values {
-		byTimestamp[v.Timestamp] = v
-	}
-
-	writer.token(jsontext.BeginArray)
-	for ts := fill.start - fill.start%fill.step; ts <= fill.end && writer.err == nil; ts += fill.step {
-		if v, ok := byTimestamp[ts]; ok {
-			writer.point(v)
-			continue
-		}
-		writer.token(jsontext.BeginObject)
-		writer.token(jsontext.String("timestamp"))
-		writer.token(jsontext.Int(ts))
-		writer.token(jsontext.String("value"))
-		writer.token(jsontext.Int(0))
-		writer.token(jsontext.EndObject)
-	}
-	writer.token(jsontext.EndArray)
-}
-
-func (writer *tokenWriter) point(v *TimeSeriesValue) {
-	if v == nil {
-		writer.token(jsontext.Null)
-		return
-	}
-	if writer.err == nil {
-		writer.err = v.MarshalJSONTo(writer.enc)
-	}
-}
-
-// sanitizedFloatToken encodes f the way sanitizeValue renders it.
-func sanitizedFloatToken(f float64) jsontext.Token {
-	switch {
-	case math.IsNaN(f):
-		return jsontext.String("NaN")
-	case math.IsInf(f, 1):
-		return jsontext.String("Inf")
-	case math.IsInf(f, -1):
-		return jsontext.String("-Inf")
-	}
-	return jsontext.Float(roundToNonZeroDecimals(f, 3))
-}

pkg/types/querybuildertypes/querybuildertypesv5/resp_json_test.go

@@ -1,168 +0,0 @@
-package querybuildertypesv5
-
-import (
-	"encoding/json"
-	"testing"
-
-	jsonv2 "github.com/go-json-experiment/json"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestTimeSeriesData_ZeroFill(t *testing.T) {
-	testCases := []struct {
-		name     string
-		values   []*TimeSeriesValue
-		start    int64
-		end      int64
-		step     int64
-		expected string
-	}{
-		{
-			name:     "Gaps",
-			values:   []*TimeSeriesValue{{Timestamp: 1000, Value: 1}, {Timestamp: 4000, Value: 4}},
-			start:    1000,
-			end:      4000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":1},{"timestamp":2000,"value":0},{"timestamp":3000,"value":0},{"timestamp":4000,"value":4}]`,
-		},
-		{
-			name:     "Empty",
-			values:   []*TimeSeriesValue{},
-			start:    1000,
-			end:      3000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":0},{"timestamp":3000,"value":0}]`,
-		},
-		{
-			name:     "Nil",
-			values:   nil,
-			start:    1000,
-			end:      2000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":0}]`,
-		},
-		{
-			name:     "UnalignedBounds",
-			values:   []*TimeSeriesValue{{Timestamp: 60000, Value: 1}, {Timestamp: 240000, Value: 4}},
-			start:    50000,
-			end:      250000,
-			step:     60000,
-			expected: `[{"timestamp":0,"value":0},{"timestamp":60000,"value":1},{"timestamp":120000,"value":0},{"timestamp":180000,"value":0},{"timestamp":240000,"value":4}]`,
-		},
-		{
-			name:     "OffGridAndOutOfRangeDropped",
-			values:   []*TimeSeriesValue{{Timestamp: 500, Value: 5}, {Timestamp: 1500, Value: 15}, {Timestamp: 2000, Value: 2}, {Timestamp: 3000, Value: 3}},
-			start:    1000,
-			end:      2000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":2}]`,
-		},
-		{
-			name:     "DuplicateTimestampKeepsLast",
-			values:   []*TimeSeriesValue{{Timestamp: 1000, Value: 1}, {Timestamp: 1000, Value: 9}},
-			start:    1000,
-			end:      1000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":9}]`,
-		},
-		{
-			name:     "PointFieldsKept",
-			values:   []*TimeSeriesValue{{Timestamp: 2000, Value: 2.12345, Partial: true}},
-			start:    1000,
-			end:      2000,
-			step:     1000,
-			expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"partial":true,"value":2.123}]`,
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			series := &TimeSeries{Values: testCase.values}
-			data := &TimeSeriesData{
-				QueryName: "A",
-				Aggregations: []*AggregationBucket{{
-					Series:          []*TimeSeries{series},
-					PredictedSeries: []*TimeSeries{{Values: []*TimeSeriesValue{{Timestamp: 3000, Value: 3}}}},
-				}},
-			}
-			data.SetZeroFill(testCase.start, testCase.end, testCase.step)
-
-			expected := `{"queryName":"A","aggregations":[{"index":0,"alias":"","meta":{},"series":[{"values":` + testCase.expected + `}],"predictedSeries":[{"values":[{"timestamp":3000,"value":3}]}]}]}`
-
-			marshalled, err := json.Marshal(data)
-			require.NoError(t, err)
-			assert.JSONEq(t, expected, string(marshalled))
-
-			response := QueryRangeResponse{Type: RequestTypeTimeSeries, Data: QueryData{Results: []any{data}}}
-			viaMarshalJSON, err := json.Marshal(response)
-			require.NoError(t, err)
-			viaMarshalJSONTo, err := jsonv2.Marshal(response)
-			require.NoError(t, err)
-			assert.Equal(t, string(viaMarshalJSON), string(viaMarshalJSONTo))
-
-			var decoded struct {
-				Data struct {
-					Results []json.RawMessage `json:"results"`
-				} `json:"data"`
-			}
-			require.NoError(t, json.Unmarshal(viaMarshalJSONTo, &decoded))
-			require.Len(t, decoded.Data.Results, 1)
-			assert.JSONEq(t, expected, string(decoded.Data.Results[0]))
-
-			assert.Equal(t, testCase.values, series.Values)
-		})
-	}
-}
-
-func TestTimeSeriesData_ZeroFillMatchesFuncFillZero(t *testing.T) {
-	values := []*TimeSeriesValue{{Timestamp: 500, Value: 5}, {Timestamp: 60000, Value: 1}, {Timestamp: 90000, Value: 9}, {Timestamp: 240000, Value: 4, Partial: true}}
-
-	filled := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: []*TimeSeries{funcFillZero(&TimeSeries{Values: values}, 50000, 250000, 60000)}}}}
-	deferred := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: []*TimeSeries{{Values: values}}}}}
-	deferred.SetZeroFill(50000, 250000, 60000)
-
-	expected, err := json.Marshal(filled)
-	require.NoError(t, err)
-	actual, err := json.Marshal(deferred)
-	require.NoError(t, err)
-	assert.Equal(t, string(expected), string(actual))
-}
-
-// failingWriter accepts its first write and fails every one after it.
-type failingWriter struct {
-	writes int
-}
-
-func (writer *failingWriter) Write(p []byte) (int, error) {
-	writer.writes++
-	if writer.writes == 1 {
-		return len(p), nil
-	}
-	return 0, assert.AnError
-}
-
-func TestQueryRangeResponse_MarshalStopsOnWriteError(t *testing.T) {
-	values := make([]*TimeSeriesValue, 16)
-	for i := range values {
-		values[i] = &TimeSeriesValue{Timestamp: int64(i) * 1000, Value: 1}
-	}
-	series := make([]*TimeSeries, 10000)
-	for i := range series {
-		series[i] = &TimeSeries{Values: values}
-	}
-	data := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: series}}}
-	data.SetZeroFill(0, 1000*1000, 1000)
-
-	response := QueryRangeResponse{Type: RequestTypeTimeSeries, Data: QueryData{Results: []any{data}}}
-
-	writer := &failingWriter{}
-	err := jsonv2.MarshalWrite(writer, response)
-	assert.ErrorIs(t, err, assert.AnError)
-	assert.Equal(t, 2, writer.writes)
-
-	allocs := testing.AllocsPerRun(1, func() {
-		_ = jsonv2.MarshalWrite(&failingWriter{}, response)
-	})
-	assert.Less(t, allocs, float64(len(series)))
-}

tests/integration/tests/queriermetrics/17_cache.py

@@ -442,3 +442,55 @@ def test_builder_refreshing_a_sliding_time_range(
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")
+
+
+def test_builder_refreshing_does_not_inflate_rows_scanned(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    metric_name = f"cache_rows_scanned_{uuid4().hex[:8]}"
+
+    # 90 minutes back so even the last refresh closes clear of the flux
+    # interval, which holds recent data out of the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=90)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+
+    query = [build_builder_query("A", metric_name, "max", "max")]
+
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=minute,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(70)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # a dashboard left open on a thirty minute range, re-running a minute later
+    # each time. rowsScanned is a cost indicator, so a refresh served mostly out
+    # of the cache cannot claim more than every read made so far: the first
+    # full read and then one new minute per refresh, measured with the cache off
+    uncached_rows_scanned_so_far = 0
+    for refresh in range(10):
+        refresh_start_ms = start_time_ms + refresh * MINUTE_MS
+        refresh_end_ms = refresh_start_ms + 30 * MINUTE_MS
+
+        uncached_start_ms = refresh_start_ms if refresh == 0 else refresh_end_ms - MINUTE_MS
+        uncached_response = make_query_request(signoz, token, uncached_start_ms, refresh_end_ms, query, no_cache=True)
+        assert uncached_response.status_code == HTTPStatus.OK, f"refresh {refresh}: uncached read failed: {uncached_response.text}"
+        uncached_rows_scanned_so_far += uncached_response.json()["data"]["meta"]["rowsScanned"]
+        assert uncached_rows_scanned_so_far > 0, f"refresh {refresh}: uncached read reported no rows scanned"
+
+        cached_response = make_query_request(signoz, token, refresh_start_ms, refresh_end_ms, query, no_cache=False)
+        assert cached_response.status_code == HTTPStatus.OK, f"refresh {refresh}: cached read failed: {cached_response.text}"
+        cached_rows_scanned = cached_response.json()["data"]["meta"]["rowsScanned"]
+        assert cached_rows_scanned <= uncached_rows_scanned_so_far, f"refresh {refresh}: cached read reported {cached_rows_scanned} rows scanned, more than the {uncached_rows_scanned_so_far} rows all uncached reads scanned so far"