Merge branch 'main' into fix/clickhouse-sql-mistakes-invalid-input

Assisted-by: Claude Opus 5.5

pkg/cache/cache.go

@@ -16,11 +16,6 @@ type Cache interface {
 	// Get gets the cacheble entity in the dest entity passed.
 	Get(ctx context.Context, orgID valuer.UUID, cacheKey string, dest cachetypes.Cacheable) error
 
-	// GetMany gets the cacheable entities of cacheKeys into dests, one per key in
-	// the same order. A key that is not found or could not be read sets its dest
-	// to nil. The returned error joins the read failures; a miss alone is not one.
-	GetMany(ctx context.Context, orgID valuer.UUID, cacheKeys []string, dests []cachetypes.Cacheable) error
-
 	// Delete deletes the cacheable entity from cache
 	Delete(ctx context.Context, orgID valuer.UUID, cacheKey string)
 

pkg/cache/memorycache/provider.go

@@ -196,28 +196,6 @@ func (provider *provider) Get(ctx context.Context, orgID valuer.UUID, cacheKey s
 	return errors.NewInternalf(errors.CodeInternal, "unrecognized: (value: \"%s\")", reflect.TypeOf(cachedData).String())
 }
 
-func (provider *provider) GetMany(ctx context.Context, orgID valuer.UUID, cacheKeys []string, dests []cachetypes.Cacheable) error {
-	if len(dests) != len(cacheKeys) {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "got %d destinations for %d keys", len(dests), len(cacheKeys))
-	}
-
-	var errs []error
-	for i, cacheKey := range cacheKeys {
-		err := provider.Get(ctx, orgID, cacheKey, dests[i])
-		if err == nil {
-			continue
-		}
-		dests[i] = nil
-		if !errors.Ast(err, errors.TypeNotFound) {
-			errs = append(errs, err)
-		}
-	}
-	if len(errs) == 0 {
-		return nil
-	}
-	return errors.Join(errs...)
-}
-
 func (provider *provider) Delete(ctx context.Context, orgID valuer.UUID, cacheKey string) {
 	_, span := provider.settings.Tracer().Start(ctx, "memory.delete", trace.WithAttributes(
 		attribute.String(semconv.AttributeDBSystem, "memory"),

pkg/cache/memorycache/provider_test.go

@@ -260,27 +260,3 @@ func TestCloneableConcurrentSetGet(t *testing.T) {
 		assert.NotSame(t, cachedCloneable, cloneables[i])
 	}
 }
-
-func TestGetMany(t *testing.T) {
-	cache, err := New(context.Background(), factorytest.NewSettings(), cache.Config{Provider: "memory", Memory: cache.Memory{
-		NumCounters: 10 * 1000,
-		MaxCost:     1 << 26,
-	}})
-	require.NoError(t, err)
-
-	orgID := valuer.GenerateUUID()
-	require.NoError(t, cache.Set(context.Background(), orgID, "first", &CloneableA{Key: "first", Value: 1}, 10*time.Second))
-	require.NoError(t, cache.Set(context.Background(), orgID, "third", &CloneableA{Key: "third", Value: 3}, 10*time.Second))
-
-	// bytes that are not JSON cannot be read into a CloneableA
-	memory := cache.(*provider)
-	memory.cc.SetWithTTL(strings.Join([]string{orgID.StringValue(), "fourth"}, "::"), []byte("not json"), 1, 10*time.Second)
-	memory.cc.Wait()
-
-	dests := []cachetypes.Cacheable{new(CloneableA), new(CloneableA), new(CloneableA), new(CloneableA)}
-	err = cache.GetMany(context.Background(), orgID, []string{"first", "second", "third", "fourth"}, dests)
-
-	// the unreadable key is reported and nilled, the readable ones are still served
-	assert.Error(t, err)
-	assert.Equal(t, []cachetypes.Cacheable{&CloneableA{Key: "first", Value: 1}, nil, &CloneableA{Key: "third", Value: 3}, nil}, dests)
-}

pkg/cache/rediscache/provider.go

@@ -66,49 +66,6 @@ func (c *provider) Get(ctx context.Context, orgID valuer.UUID, cacheKey string,
 	return nil
 }
 
-func (c *provider) GetMany(ctx context.Context, orgID valuer.UUID, cacheKeys []string, dests []cachetypes.Cacheable) error {
-	if len(dests) != len(cacheKeys) {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "got %d destinations for %d keys", len(dests), len(cacheKeys))
-	}
-	if len(cacheKeys) == 0 {
-		return nil
-	}
-
-	updatedCacheKeys := make([]string, len(cacheKeys))
-	for i, cacheKey := range cacheKeys {
-		updatedCacheKeys[i] = strings.Join([]string{orgID.StringValue(), cacheKey}, "::")
-	}
-
-	values, err := c.client.MGet(ctx, updatedCacheKeys...).Result()
-	if err != nil {
-		for i := range dests {
-			dests[i] = nil
-		}
-		return err
-	}
-
-	var errs []error
-	for i, value := range values {
-		// MGET answers a missing key with nil and a present one with its string
-		switch stored := value.(type) {
-		case nil:
-			dests[i] = nil
-		case string:
-			if err := dests[i].UnmarshalBinary([]byte(stored)); err != nil {
-				dests[i] = nil
-				errs = append(errs, err)
-			}
-		default:
-			dests[i] = nil
-			errs = append(errs, errors.NewInternalf(errors.CodeInternal, "unexpected value of type %T for key %s", value, cacheKeys[i]))
-		}
-	}
-	if len(errs) == 0 {
-		return nil
-	}
-	return errors.Join(errs...)
-}
-
 func (c *provider) Delete(ctx context.Context, orgID valuer.UUID, cacheKey string) {
 	c.DeleteMany(ctx, orgID, []string{cacheKey})
 }

pkg/cache/rediscache/provider_test.go

@@ -13,7 +13,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/go-redis/redismock/v9"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 type CacheableA struct {
@@ -59,28 +58,3 @@ func TestSet(t *testing.T) {
 	assert.NoError(t, cache.Set(context.Background(), orgID, "key", cacheable, 10*time.Second))
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
-
-func TestGetMany(t *testing.T) {
-	db, mock := redismock.NewClientMock()
-	providerSettings := instrumentationtest.New().ToProviderSettings()
-	cache := &provider{client: db, settings: factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/cache/rediscache")}
-
-	orgID := valuer.GenerateUUID()
-	stored, err := json.Marshal(&CacheableA{Key: "first", Value: 1, Expiry: time.Microsecond})
-	require.NoError(t, err)
-	// one round trip for all keys, with the missing one answered as nil and
-	// the third holding something that is not a CacheableA
-	mock.ExpectMGet(
-		strings.Join([]string{orgID.StringValue(), "first"}, "::"),
-		strings.Join([]string{orgID.StringValue(), "second"}, "::"),
-		strings.Join([]string{orgID.StringValue(), "third"}, "::"),
-	).SetVal([]any{string(stored), nil, "not json"})
-
-	dests := []cachetypes.Cacheable{new(CacheableA), new(CacheableA), new(CacheableA)}
-	err = cache.GetMany(context.Background(), orgID, []string{"first", "second", "third"}, dests)
-
-	// the unreadable key is reported and nilled, the readable one is still served
-	assert.Error(t, err)
-	assert.Equal(t, []cachetypes.Cacheable{&CacheableA{Key: "first", Value: 1, Expiry: time.Microsecond}, nil, nil}, dests)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}

pkg/querier/bucket_cache.go

@@ -1,7 +1,6 @@
 package querier
 
 import (
-	"cmp"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -12,16 +11,11 @@ import (
 	"github.com/SigNoz/signoz/pkg/cache"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/types/cachetypes"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
-// bucketCacheCellSteps is the grid cell width in steps; no cached bucket
-// crosses a cell boundary.
-const bucketCacheCellSteps = 60
-
 // BucketCache is the interface for bucket-based caching.
 type BucketCache interface {
 	// cached portion + list of gaps to fetch
@@ -64,59 +58,44 @@ func (bc *bucketCache) GetMissRanges(
 
 	bc.logger.DebugContext(ctx, "getting miss ranges", slog.String("fingerprint", q.Fingerprint()), slog.Uint64("start", startMs), slog.Uint64("end", endMs))
 
+	// Generate cache key
+	cacheKey := bc.generateCacheKey(q)
+
+	bc.logger.DebugContext(ctx, "cache key", slog.String("cache_key", cacheKey))
+
+	// Try to get cached data
+	var data qbtypes.CachedData
+	err := bc.cache.Get(ctx, orgID, cacheKey, &data)
+	if err != nil {
+		if !errors.Ast(err, errors.TypeNotFound) {
+			bc.logger.ErrorContext(ctx, "error getting cached data", errors.Attr(err))
+		}
+		// No cached data, need to fetch entire range
+		missing = []*qbtypes.TimeRange{{From: startMs, To: endMs}}
+		return nil, missing
+	}
+
 	// Extract step interval if this is a builder query
 	stepMs := uint64(step.Milliseconds())
 
-	// No cached data, need to fetch entire range
-	missing = []*qbtypes.TimeRange{{From: startMs, To: endMs}}
-
-	if endMs <= startMs {
-		return nil, missing
-	}
-	cellWidthMs := bucketCacheCellSteps * stepMs
-	firstCellIdx, lastCellIdx := deriveCellIdx(startMs, cellWidthMs), deriveCellIdx(endMs-1, cellWidthMs)
-
-	// Gather the buckets of every cell the window touches
-	cellKeys := make([]string, 0, lastCellIdx-firstCellIdx+1)
-	cellEntries := make([]cachetypes.Cacheable, 0, lastCellIdx-firstCellIdx+1)
-	for cellIdx := firstCellIdx; cellIdx <= lastCellIdx; cellIdx++ {
-		cellKeys = append(cellKeys, bc.generateCacheKey(q, cellIdx))
-		cellEntries = append(cellEntries, &qbtypes.CachedData{})
-	}
-
-	// A cell that could not be read comes back nil like a miss, so the rest
-	// are still worth serving
-	if err := bc.cache.GetMany(ctx, orgID, cellKeys, cellEntries); err != nil {
-		bc.logger.ErrorContext(ctx, "error getting cached data", errors.Attr(err))
-	}
-
-	cachedBuckets := []*qbtypes.CachedBucket{}
-	cachedWarnings := []string{}
-	for _, cellEntry := range cellEntries {
-		if cellEntry == nil {
-			continue
-		}
-		cellData := cellEntry.(*qbtypes.CachedData)
-		cachedBuckets = append(cachedBuckets, cellData.Buckets...)
-		cachedWarnings = append(cachedWarnings, cellData.Warnings...)
-	}
-	if len(cachedBuckets) == 0 {
-		return nil, missing
-	}
-
 	// Find missing ranges with step alignment
-	missing = bc.findMissingRangesWithStep(cachedBuckets, startMs, endMs, stepMs)
+	missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs)
 	bc.logger.DebugContext(ctx, "missing ranges", slog.Any("missing", missing), slog.Uint64("step", stepMs))
 
+	// If no cached data overlaps with requested range, return empty result
+	if len(data.Buckets) == 0 {
+		return nil, missing
+	}
+
 	// Extract relevant buckets and merge them
-	relevantBuckets := bc.filterRelevantBuckets(cachedBuckets, startMs, endMs)
+	relevantBuckets := bc.filterRelevantBuckets(data.Buckets, startMs, endMs)
 	if len(relevantBuckets) == 0 {
 		// nothing cached overlaps the window, so all of it has to be fetched.
 		return nil, []*qbtypes.TimeRange{{From: startMs, To: endMs}}
 	}
 
 	// Merge buckets into a single result
-	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, cachedWarnings)
+	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, data.Warnings)
 
 	// Filter the merged result to only include values within the requested time range
 	mergedResult = bc.filterResultToTimeRange(mergedResult, q, startMs, endMs, stepMs)
@@ -151,6 +130,15 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 			slog.Uint64("cachable_end", cachableEndMs))
 	}
 
+	// Generate cache key
+	cacheKey := bc.generateCacheKey(q)
+
+	// Get existing cached data
+	var existingData qbtypes.CachedData
+	if err := bc.cache.Get(ctx, orgID, cacheKey, &existingData); err != nil {
+		existingData = qbtypes.CachedData{}
+	}
+
 	// Trim the result to exclude data within flux interval
 	trimmedResult := bc.trimResultToFluxBoundary(fresh, cachableEndMs)
 	if trimmedResult == nil {
@@ -186,44 +174,38 @@ 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)
+	// Convert trimmed result to buckets with adjusted boundaries
+	freshBuckets := bc.resultToBuckets(ctx, trimmedResult, cachableStartMs, cachableEndMs)
 
-	// Each cell is its own cache entry, rewritten only when the fresh bucket
-	// changes what it holds, so an entry's TTL runs from its last new data
-	for _, freshBucket := range freshBuckets {
-		cellKey := bc.generateCacheKey(q, deriveCellIdx(freshBucket.StartMs, cellWidthMs))
+	// If no fresh buckets and no existing data, don't cache
+	if len(freshBuckets) == 0 && len(existingData.Buckets) == 0 {
+		return
+	}
 
-		var cellEntry qbtypes.CachedData
-		if err := bc.cache.Get(ctx, orgID, cellKey, &cellEntry); err != nil {
-			cellEntry = qbtypes.CachedData{}
-		}
+	// Merge with existing buckets
+	mergedBuckets := bc.mergeAndDeduplicateBuckets(existingData.Buckets, freshBuckets)
 
-		cellBuckets, changed := bc.mergeFreshBucketIntoCell(ctx, cellEntry.Buckets, freshBucket)
-		if !changed {
-			continue
-		}
-		cellEntry.Buckets = cellBuckets
-		cellEntry.Warnings = bc.deduplicateWarnings(append(cellEntry.Warnings, trimmedResult.Warnings...))
-		if err := bc.cache.Set(ctx, orgID, cellKey, &cellEntry, bc.cacheTTL); err != nil {
-			bc.logger.ErrorContext(ctx, "error setting cached data", errors.Attr(err))
-		}
+	// Update warnings
+	allWarnings := append(existingData.Warnings, trimmedResult.Warnings...)
+	uniqueWarnings := bc.deduplicateWarnings(allWarnings)
+
+	// Create updated cached data
+	updatedData := qbtypes.CachedData{
+		Buckets:  mergedBuckets,
+		Warnings: uniqueWarnings,
+	}
+
+	// Marshal and store in cache
+	if err := bc.cache.Set(ctx, orgID, cacheKey, &updatedData, bc.cacheTTL); err != nil {
+		bc.logger.ErrorContext(ctx, "error setting cached data", errors.Attr(err))
 	}
 }
 
-// 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)
-}
+// generateCacheKey creates a unique cache key based on query fingerprint.
+func (bc *bucketCache) generateCacheKey(q qbtypes.Query) string {
+	fingerprint := q.Fingerprint()
 
-// deriveCellIdx returns the index of the grid cell holding startMs; with no grid
-// every bucket shares cell 0.
-func deriveCellIdx(startMs, cellWidthMs uint64) uint64 {
-	if cellWidthMs == 0 {
-		return 0
-	}
-	return startMs / cellWidthMs
+	return fmt.Sprintf("v5:query:%s", fingerprint)
 }
 
 // findMissingRangesWithStep identifies time ranges not covered by cached buckets with step alignment.
@@ -664,7 +646,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 uint64) []*qbtypes.CachedBucket {
 	// Check if result is empty
 	isEmpty, isFiltered := bc.isEmptyResult(result)
 
@@ -675,74 +657,62 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
 		return nil
 	}
 
-	// Empty filtered results still become buckets, so the range is not re-queried
-	perCellBuckets, err := result.SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs)
+	// For now, create a single bucket for the entire range
+	// In the future, we could split large ranges into smaller buckets
+	valueBytes, err := json.Marshal(result.Value)
 	if err != nil {
-		bc.logger.ErrorContext(ctx, "failed to convert result to buckets", errors.Attr(err))
+		bc.logger.ErrorContext(ctx, "failed to marshal result value", errors.Attr(err))
 		return nil
 	}
-	return perCellBuckets
+
+	// Always create a bucket, even for empty filtered results
+	// This ensures we don't re-query for data that doesn't exist
+	return []*qbtypes.CachedBucket{
+		{
+			StartMs: startMs,
+			EndMs:   endMs,
+			Type:    result.Type,
+			Value:   valueBytes,
+			Stats:   result.Stats,
+		},
+	}
 }
 
-// mergeFreshBucketIntoCell folds a fresh bucket into the buckets of its cell
-// 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) {
-	keptBuckets := make([]*qbtypes.CachedBucket, 0, len(cellBuckets)+1)
-	touchingBuckets := []*qbtypes.CachedBucket{}
-	for _, cellBucket := range cellBuckets {
-		switch {
-		case cellBucket.Contains(freshBucket):
-			return cellBuckets, false
-		case cellBucket.Touches(freshBucket):
-			touchingBuckets = append(touchingBuckets, cellBucket)
-		default:
-			keptBuckets = append(keptBuckets, cellBucket)
-		}
-	}
-	keptBuckets = append(keptBuckets, bc.unionBuckets(ctx, freshBucket, touchingBuckets))
+// mergeAndDeduplicateBuckets combines and deduplicates bucket lists.
+func (bc *bucketCache) mergeAndDeduplicateBuckets(existing, fresh []*qbtypes.CachedBucket) []*qbtypes.CachedBucket {
+	// Create a map to deduplicate by time range
+	bucketMap := make(map[string]*qbtypes.CachedBucket)
 
-	slices.SortFunc(keptBuckets, func(a, b *qbtypes.CachedBucket) int {
-		return cmp.Compare(a.StartMs, b.StartMs)
+	// Add existing buckets
+	for _, bucket := range existing {
+		key := fmt.Sprintf("%d-%d", bucket.StartMs, bucket.EndMs)
+		bucketMap[key] = bucket
+	}
+
+	// Add/update with fresh buckets
+	for _, bucket := range fresh {
+		key := fmt.Sprintf("%d-%d", bucket.StartMs, bucket.EndMs)
+		bucketMap[key] = bucket
+	}
+
+	// Convert back to slice with pre-allocated capacity
+	result := make([]*qbtypes.CachedBucket, 0, len(bucketMap))
+	for _, bucket := range bucketMap {
+		result = append(result, bucket)
+	}
+
+	// Sort by start time
+	slices.SortFunc(result, func(a, b *qbtypes.CachedBucket) int {
+		if a.StartMs < b.StartMs {
+			return -1
+		}
+		if a.StartMs > b.StartMs {
+			return 1
+		}
+		return 0
 	})
-	return keptBuckets, true
-}
 
-// 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 {
-	bucketsToMerge := []*qbtypes.CachedBucket{freshBucket}
-	for _, touchingBucket := range touchingBuckets {
-		if !freshBucket.Contains(touchingBucket) {
-			bucketsToMerge = append(bucketsToMerge, touchingBucket)
-		}
-	}
-	if len(bucketsToMerge) == 1 {
-		return freshBucket
-	}
-
-	value, err := json.Marshal(bc.mergeTimeSeriesValues(ctx, bucketsToMerge))
-	if err != nil {
-		bc.logger.ErrorContext(ctx, "failed to marshal unioned bucket", errors.Attr(err))
-		return freshBucket
-	}
-	unionBucket := &qbtypes.CachedBucket{
-		StartMs: freshBucket.StartMs,
-		EndMs:   freshBucket.EndMs,
-		Type:    freshBucket.Type,
-		Value:   value,
-		Stats:   qbtypes.ExecStats{StepIntervals: freshBucket.Stats.StepIntervals},
-	}
-	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
+	return result
 }
 
 // deduplicateWarnings removes duplicate warnings.

pkg/querier/bucket_cache_test.go

@@ -2,7 +2,6 @@ package querier
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log/slog"
 	"testing"
@@ -1466,161 +1465,3 @@ func TestBucketCache_NoCache(t *testing.T) {
 	// The actual NoCache logic is implemented in querier.run(), not in bucket cache
 	// This test verifies that the cache works normally and NoCache bypasses it at a higher level
 }
-
-func TestBucketCachePutSlicesAtCellBoundaries(t *testing.T) {
-	bc := createTestBucketCache(t)
-	ctx := context.Background()
-	orgID := valuer.UUID{}
-	step := qbtypes.Step{Duration: time.Second}
-
-	// 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{
-		Type:  qbtypes.RequestTypeTimeSeries,
-		Value: createTestTimeSeries("A", 30000, 150000, 1000),
-		Stats: qbtypes.ExecStats{RowsScanned: 120, BytesScanned: 1200, DurationMS: 12},
-	})
-
-	type bucketShape struct {
-		cell, startMs, endMs, rowsScanned uint64
-		valueCount                        int
-	}
-	shapes := []bucketShape{}
-	for cell := uint64(0); cell < 3; cell++ {
-		var cellEntry qbtypes.CachedData
-		require.NoError(t, bc.cache.Get(ctx, orgID, bc.generateCacheKey(query, cell), &cellEntry))
-		for _, bucket := range cellEntry.Buckets {
-			var tsData qbtypes.TimeSeriesData
-			require.NoError(t, json.Unmarshal(bucket.Value, &tsData))
-			require.Len(t, tsData.Aggregations, 1)
-			require.Len(t, tsData.Aggregations[0].Series, 1)
-			shapes = append(shapes, bucketShape{cell, bucket.StartMs, bucket.EndMs, bucket.Stats.RowsScanned, len(tsData.Aggregations[0].Series[0].Values)})
-		}
-	}
-	assert.Equal(t, []bucketShape{
-		{0, 30000, 60000, 30, 30},
-		{1, 60000, 120000, 60, 60},
-		{2, 120000, 150000, 30, 30},
-	}, shapes)
-
-	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
-	assert.Equal(t, []*qbtypes.TimeRange{}, missing)
-	assert.Len(t, cached.Value.(*qbtypes.TimeSeriesData).Aggregations[0].Series[0].Values, 120)
-	assert.Equal(t, uint64(120), cached.Stats.RowsScanned)
-}
-
-func TestBucketCachePutAbsorbsWritesWithinACell(t *testing.T) {
-	type bucketShape struct{ startMs, endMs, rowsScanned uint64 }
-	testCases := []struct {
-		name                  string
-		firstWindow           qbtypes.TimeRange
-		secondWindow          qbtypes.TimeRange
-		expectedBuckets       []bucketShape
-		expectedMissingRanges []*qbtypes.TimeRange
-		expectedValueCount    int
-	}{
-		{
-			name:                  "DisjointWrites_StayApart",
-			firstWindow:           qbtypes.TimeRange{From: 1000, To: 3000},
-			secondWindow:          qbtypes.TimeRange{From: 5000, To: 8000},
-			expectedBuckets:       []bucketShape{{1000, 3000, 2}, {5000, 8000, 3}},
-			expectedMissingRanges: []*qbtypes.TimeRange{{From: 3000, To: 5000}},
-			expectedValueCount:    5,
-		},
-		{
-			name:                  "EdgeToEdgeWrites_BecomeOne",
-			firstWindow:           qbtypes.TimeRange{From: 1000, To: 3000},
-			secondWindow:          qbtypes.TimeRange{From: 3000, To: 5000},
-			expectedBuckets:       []bucketShape{{1000, 5000, 4}},
-			expectedMissingRanges: []*qbtypes.TimeRange{},
-			expectedValueCount:    4,
-		},
-		{
-			name:                  "OverlappingWrites_BecomeOne",
-			firstWindow:           qbtypes.TimeRange{From: 1000, To: 4000},
-			secondWindow:          qbtypes.TimeRange{From: 3000, To: 6000},
-			expectedBuckets:       []bucketShape{{1000, 6000, 6}},
-			expectedMissingRanges: []*qbtypes.TimeRange{},
-			expectedValueCount:    5,
-		},
-		{
-			name:                  "ContainedWrite_KeepsExistingBucket",
-			firstWindow:           qbtypes.TimeRange{From: 1000, To: 10000},
-			secondWindow:          qbtypes.TimeRange{From: 3000, To: 5000},
-			expectedBuckets:       []bucketShape{{1000, 10000, 9}},
-			expectedMissingRanges: []*qbtypes.TimeRange{},
-			expectedValueCount:    9,
-		},
-		{
-			name:                  "ContainingWrite_ReplacesExistingBucket",
-			firstWindow:           qbtypes.TimeRange{From: 3000, To: 5000},
-			secondWindow:          qbtypes.TimeRange{From: 1000, To: 10000},
-			expectedBuckets:       []bucketShape{{1000, 10000, 9}},
-			expectedMissingRanges: []*qbtypes.TimeRange{},
-			expectedValueCount:    9,
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			bc := createTestBucketCache(t)
-			ctx := context.Background()
-			orgID := valuer.UUID{}
-			step := qbtypes.Step{Duration: time.Second}
-
-			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{
-					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
-					Stats: qbtypes.ExecStats{RowsScanned: (window.To - window.From) / 1000},
-				})
-			}
-
-			query := &mockQuery{
-				fingerprint: "absorb-query",
-				startMs:     min(testCase.firstWindow.From, testCase.secondWindow.From),
-				endMs:       max(testCase.firstWindow.To, testCase.secondWindow.To),
-			}
-			var cellEntry qbtypes.CachedData
-			require.NoError(t, bc.cache.Get(ctx, orgID, bc.generateCacheKey(query, 0), &cellEntry))
-			shapes := make([]bucketShape, 0, len(cellEntry.Buckets))
-			for _, bucket := range cellEntry.Buckets {
-				shapes = append(shapes, bucketShape{bucket.StartMs, bucket.EndMs, bucket.Stats.RowsScanned})
-			}
-			assert.Equal(t, testCase.expectedBuckets, shapes)
-
-			cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
-			assert.Equal(t, testCase.expectedMissingRanges, missing)
-			assert.Len(t, cached.Value.(*qbtypes.TimeSeriesData).Aggregations[0].Series[0].Values, testCase.expectedValueCount)
-		})
-	}
-}
-
-func TestBucketCachePutLeavesAnUnchangedCellAlone(t *testing.T) {
-	bc := createTestBucketCache(t)
-	ctx := context.Background()
-	orgID := valuer.UUID{}
-	step := qbtypes.Step{Duration: time.Second}
-
-	wide := &mockQuery{fingerprint: "unchanged-query", startMs: 1000, endMs: 10000}
-	bc.Put(ctx, orgID, wide, step, &qbtypes.Result{
-		Type:     qbtypes.RequestTypeTimeSeries,
-		Value:    createTestTimeSeries("A", 1000, 10000, 1000),
-		Warnings: []string{"from the wide write"},
-	})
-
-	// a write already covered by the cell's bucket must not rewrite the entry,
-	// which shows as its warnings never arriving
-	inside := &mockQuery{fingerprint: "unchanged-query", startMs: 3000, endMs: 5000}
-	bc.Put(ctx, orgID, inside, step, &qbtypes.Result{
-		Type:     qbtypes.RequestTypeTimeSeries,
-		Value:    createTestTimeSeries("A", 3000, 5000, 1000),
-		Warnings: []string{"from the inside write"},
-	})
-
-	var cellEntry qbtypes.CachedData
-	require.NoError(t, bc.cache.Get(ctx, orgID, bc.generateCacheKey(wide, 0), &cellEntry))
-	assert.Equal(t, []string{"from the wide write"}, cellEntry.Warnings)
-}

pkg/telemetrystore/clickhousetelemetrystore/provider.go

@@ -20,6 +20,9 @@ var (
 	ErrCodeUnknownIdentifier = errors.MustNewCode("unknown_identifier")
 	ErrCodeIllegalArgument   = errors.MustNewCode("illegal_argument")
 
+	ErrCodeDistributedSubqueryDenied = errors.MustNewCode("distributed_subquery_denied")
+	ErrCodeDecimalOverflow           = errors.MustNewCode("decimal_overflow")
+
 	ErrCodeQueryCanceled   = errors.MustNewCode("query_canceled")
 	ErrCodeQueryTimeout    = errors.MustNewCode("query_timeout")
 	ErrCodeExecutionFailed = errors.MustNewCode("execution_failed")
@@ -72,6 +75,18 @@ var clickHouseExceptionWrappers = map[chproto.Error]func(cause error, ex *clickh
 	chproto.ErrTooLessArgumentsForFunction: func(cause error, ex *clickhouse.Exception) error {
 		return errors.WrapInvalidInputf(cause, ErrCodeIllegalArgument, "too few arguments to function: %s", ex.Message)
 	},
+	chproto.ErrBadGet: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapInvalidInputf(cause, ErrCodeIllegalArgument, "parameter of the wrong type: %s", ex.Message).
+			WithSuggestions("write numeric function parameters in the type the function expects, for example 1.0 rather than 1 where it takes a Float64")
+	},
+	chproto.ErrDistributedInJoinSubqueryDenied: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapInvalidInputf(cause, ErrCodeDistributedSubqueryDenied, "IN or JOIN subquery over a distributed table is not allowed: %s", ex.Message).
+			WithSuggestions("use GLOBAL IN or GLOBAL JOIN for the subquery")
+	},
+	chproto.ErrDecimalOverflow: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapInvalidInputf(cause, ErrCodeDecimalOverflow, "decimal overflow: %s", ex.Message).
+			WithSuggestions("cast the Decimal values to Float64, or to a wider Decimal such as Decimal128, before the arithmetic")
+	},
 }
 
 type provider struct {

pkg/types/querybuildertypes/querybuildertypesv5/cached.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"maps"
 
-	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/cachetypes"
 )
 
@@ -34,15 +33,6 @@ func (c *CachedBucket) Clone() *CachedBucket {
 	}
 }
 
-func (c *CachedBucket) Contains(other *CachedBucket) bool {
-	return c.StartMs <= other.StartMs && other.EndMs <= c.EndMs
-}
-
-// true if c and other overlap or sit edge to edge, false otherwise.
-func (c *CachedBucket) Touches(other *CachedBucket) bool {
-	return c.StartMs <= other.EndMs && other.StartMs <= c.EndMs
-}
-
 // CachedData represents the full cached data for a query.
 type CachedData struct {
 	Buckets  []*CachedBucket `json:"buckets"`
@@ -87,53 +77,3 @@ func (c *CachedData) Cost() int64 {
 	}
 	return size
 }
-
-func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64) ([]*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())
-	}
-
-	cellBuckets := make([]*CachedBucket, 0, 1)
-	for from := startMs; from < endMs; {
-		to := endMs
-		if cellWidthMs > 0 {
-			to = min(endMs, (from/cellWidthMs+1)*cellWidthMs)
-		}
-		// A promql value stamped exactly at the window end is an instant inside
-		// the window, so the last slice keeps it. A builder value never sits
-		// there, since the end is step-aligned and partial steps are dropped.
-		sliceEndMs := to
-		if to == endMs {
-			sliceEndMs = endMs + 1
-		}
-		value, err := json.Marshal(tsData.FilterToTimeRange(from, sliceEndMs))
-		if err != nil {
-			return nil, err
-		}
-		cellBuckets = append(cellBuckets, &CachedBucket{
-			StartMs: from,
-			EndMs:   to,
-			Type:    r.Type,
-			Value:   value,
-			Stats:   r.Stats.prorate(to-from, endMs-startMs),
-		})
-		from = to
-	}
-	return cellBuckets, nil
-}
-
-// 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 {
-	ratio := 1.0
-	if partMs < wholeMs {
-		ratio = float64(partMs) / float64(wholeMs)
-	}
-	return ExecStats{
-		RowsScanned:   uint64(float64(e.RowsScanned) * ratio),
-		BytesScanned:  uint64(float64(e.BytesScanned) * ratio),
-		DurationMS:    uint64(float64(e.DurationMS) * ratio),
-		StepIntervals: maps.Clone(e.StepIntervals),
-	}
-}

pkg/types/querybuildertypes/querybuildertypesv5/resp.go

@@ -137,28 +137,6 @@ type TimeSeriesData struct {
 	Aggregations []*AggregationBucket `json:"aggregations" description:"One bucket per aggregation of the query. The order is not guaranteed; match buckets to aggregations by index."`
 }
 
-// FilterToTimeRange copies the aggregation and series structure keeping only
-// the values with startMs <= timestamp < endMs.
-func (t *TimeSeriesData) FilterToTimeRange(startMs, endMs uint64) *TimeSeriesData {
-	filtered := &TimeSeriesData{
-		QueryName:    t.QueryName,
-		Aggregations: make([]*AggregationBucket, 0, len(t.Aggregations)),
-	}
-	for _, aggBucket := range t.Aggregations {
-		filteredBucket := &AggregationBucket{
-			Index:  aggBucket.Index,
-			Alias:  aggBucket.Alias,
-			Meta:   aggBucket.Meta,
-			Series: make([]*TimeSeries, 0, len(aggBucket.Series)),
-		}
-		for _, series := range aggBucket.Series {
-			filteredBucket.Series = append(filteredBucket.Series, series.filterToTimeRange(startMs, endMs))
-		}
-		filtered.Aggregations = append(filtered.Aggregations, filteredBucket)
-	}
-	return filtered
-}
-
 type AggregationBucket struct {
 	Index  int             `json:"index" description:"The zero-based position of the aggregation in the query."`
 	Alias  string          `json:"alias" description:"The alias of the aggregation, or the internal name __result_N, with N its index, when no alias was given."`
@@ -281,21 +259,6 @@ func (ts *TimeSeries) EvaluableValues() []*TimeSeriesValue {
 	return result
 }
 
-// filterToTimeRange copies the series keeping only the values with
-// startMs <= timestamp < endMs.
-func (ts *TimeSeries) filterToTimeRange(startMs, endMs uint64) *TimeSeries {
-	filtered := &TimeSeries{
-		Labels: ts.Labels,
-		Values: make([]*TimeSeriesValue, 0, len(ts.Values)),
-	}
-	for _, value := range ts.Values {
-		if timestamp := uint64(value.Timestamp); timestamp >= startMs && timestamp < endMs {
-			filtered.Values = append(filtered.Values, value)
-		}
-	}
-	return filtered
-}
-
 type Label struct {
 	Key   telemetrytypes.TelemetryFieldKey `json:"key" description:"The group-by field this label value belongs to."`
 	Value any                              `json:"value" description:"The label value as a string, number or boolean."`

pkg/types/querybuildertypes/querybuildertypesv5/resp_test.go

@@ -436,97 +436,3 @@ func TestRoundToNonZeroDecimals(t *testing.T) {
 	assert.Equal(t, math.Inf(1), roundToNonZeroDecimals(math.Inf(1), 3))
 	assert.Equal(t, math.Inf(-1), roundToNonZeroDecimals(math.Inf(-1), 3))
 }
-
-func TestTimeSeriesDataFilterToTimeRange(t *testing.T) {
-	testCases := []struct {
-		name             string
-		timeSeriesData   *TimeSeriesData
-		startMs          uint64
-		endMs            uint64
-		expectedFiltered *TimeSeriesData
-	}{
-		{
-			name: "HalfOpenRange_KeepsStartDropsEnd",
-			timeSeriesData: &TimeSeriesData{
-				QueryName: "A",
-				Aggregations: []*AggregationBucket{{
-					Index: 0,
-					Series: []*TimeSeries{{
-						Labels: []*Label{{Key: telemetrytypes.TelemetryFieldKey{Name: "host"}, Value: "node-1"}},
-						Values: []*TimeSeriesValue{
-							{Timestamp: 1000, Value: 1},
-							{Timestamp: 2000, Value: 2},
-							{Timestamp: 3000, Value: 3},
-							{Timestamp: 4000, Value: 4},
-						},
-					}},
-				}},
-			},
-			startMs: 2000,
-			endMs:   4000,
-			expectedFiltered: &TimeSeriesData{
-				QueryName: "A",
-				Aggregations: []*AggregationBucket{{
-					Index: 0,
-					Series: []*TimeSeries{{
-						Labels: []*Label{{Key: telemetrytypes.TelemetryFieldKey{Name: "host"}, Value: "node-1"}},
-						Values: []*TimeSeriesValue{
-							{Timestamp: 2000, Value: 2},
-							{Timestamp: 3000, Value: 3},
-						},
-					}},
-				}},
-			},
-		},
-		{
-			// heatmap counts are positional against Meta.Buckets, so the axis
-			// has to survive even when no value falls in the range
-			name: "NothingInRange_KeepsSeriesAndHeatmapAxis",
-			timeSeriesData: &TimeSeriesData{
-				QueryName: "A",
-				Aggregations: []*AggregationBucket{{
-					Index: 1,
-					Alias: "__result_1",
-					Meta:  AggregationMeta{Unit: "By", Buckets: []float64{1, 2, 4}},
-					Series: []*TimeSeries{{
-						Labels: []*Label{{Key: telemetrytypes.TelemetryFieldKey{Name: "host"}, Value: "node-1"}},
-						Values: []*TimeSeriesValue{{Timestamp: 1000, Values: []float64{1, 2, 3, 4}}},
-					}},
-				}},
-			},
-			startMs: 5000,
-			endMs:   6000,
-			expectedFiltered: &TimeSeriesData{
-				QueryName: "A",
-				Aggregations: []*AggregationBucket{{
-					Index: 1,
-					Alias: "__result_1",
-					Meta:  AggregationMeta{Unit: "By", Buckets: []float64{1, 2, 4}},
-					Series: []*TimeSeries{{
-						Labels: []*Label{{Key: telemetrytypes.TelemetryFieldKey{Name: "host"}, Value: "node-1"}},
-						Values: []*TimeSeriesValue{},
-					}},
-				}},
-			},
-		},
-		{
-			name: "NoSeries_KeepsAggregation",
-			timeSeriesData: &TimeSeriesData{
-				QueryName:    "A",
-				Aggregations: []*AggregationBucket{{Index: 0}},
-			},
-			startMs: 1000,
-			endMs:   2000,
-			expectedFiltered: &TimeSeriesData{
-				QueryName:    "A",
-				Aggregations: []*AggregationBucket{{Index: 0, Series: []*TimeSeries{}}},
-			},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			assert.Equal(t, testCase.expectedFiltered, testCase.timeSeriesData.FilterToTimeRange(testCase.startMs, testCase.endMs))
-		})
-	}
-}