Merge branch 'main' into nv/caching-edge-cases

conf/example.yaml

@@ -202,6 +202,7 @@ telemetrystore:
       max_bytes_to_read: 0
       max_result_rows: 0
       ignore_data_skipping_indices: ""
+      secondary_indices_enable_bulk_filtering: false
 
 ##################### Prometheus #####################
 prometheus:

docs/api/openapi.yml

@@ -11012,9 +11012,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:list
+        - ADMIN
       - tokenizer:
-        - cloud-integration:list
+        - ADMIN
       summary: List accounts
       tags:
       - cloudintegration
@@ -11069,9 +11069,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:create
+        - ADMIN
       - tokenizer:
-        - cloud-integration:create
+        - ADMIN
       summary: Create account
       tags:
       - cloudintegration
@@ -11114,9 +11114,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:delete
+        - ADMIN
       - tokenizer:
-        - cloud-integration:delete
+        - ADMIN
       summary: Disconnect account
       tags:
       - cloudintegration
@@ -11182,9 +11182,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:read
+        - ADMIN
       - tokenizer:
-        - cloud-integration:read
+        - ADMIN
       summary: Get account
       tags:
       - cloudintegration
@@ -11231,9 +11231,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:update
+        - ADMIN
       - tokenizer:
-        - cloud-integration:update
+        - ADMIN
       summary: Update account
       tags:
       - cloudintegration
@@ -11289,9 +11289,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration-service:list
+        - ADMIN
       - tokenizer:
-        - cloud-integration-service:list
+        - ADMIN
       summary: List account services metadata
       tags:
       - cloudintegration
@@ -11364,9 +11364,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration-service:read
+        - ADMIN
       - tokenizer:
-        - cloud-integration-service:read
+        - ADMIN
       summary: Get service for account
       tags:
       - cloudintegration
@@ -11418,9 +11418,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration-service:update
+        - ADMIN
       - tokenizer:
-        - cloud-integration-service:update
+        - ADMIN
       summary: Update service
       tags:
       - cloudintegration
@@ -11528,9 +11528,9 @@ paths:
           description: Internal Server Error
       security:
       - api_key:
-        - cloud-integration:create
+        - ADMIN
       - tokenizer:
-        - cloud-integration:create
+        - ADMIN
       summary: Get connection credentials
       tags:
       - cloudintegration
@@ -11580,8 +11580,10 @@ paths:
                 $ref: '#/components/schemas/RenderErrorResponse'
           description: Internal Server Error
       security:
-      - api_key: []
-      - tokenizer: []
+      - api_key:
+        - ADMIN
+      - tokenizer:
+        - ADMIN
       summary: List services metadata
       tags:
       - cloudintegration
@@ -11636,8 +11638,10 @@ paths:
                 $ref: '#/components/schemas/RenderErrorResponse'
           description: Internal Server Error
       security:
-      - api_key: []
-      - tokenizer: []
+      - api_key:
+        - ADMIN
+      - tokenizer:
+        - ADMIN
       summary: Get service
       tags:
       - cloudintegration

pkg/apiserver/signozapiserver/cloudintegration.go

@@ -5,15 +5,13 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/http/handler"
 	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
 	citypes "github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
 	"github.com/gorilla/mux"
 )
 
 func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/credentials", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.GetConnectionCredentials, authtypes.SigNozAdminRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.GetConnectionCredentials),
 		handler.OpenAPIDef{
 			ID:                  "GetConnectionCredentials",
 			Tags:                []string{"cloudintegration"},
@@ -26,20 +24,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbCreate)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbCreate, // get or create the credentials, so we use create verb here
-			Category: coretypes.ActionCategoryConfigurationChange,
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.CreateAccount, authtypes.SigNozAdminRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.CreateAccount),
 		handler.OpenAPIDef{
 			ID:                  "CreateAccount",
 			Tags:                []string{"cloudintegration"},
@@ -52,21 +44,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusCreated,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbCreate)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbCreate,
-			Category: coretypes.ActionCategoryConfigurationChange,
-			ID:       coretypes.ResponseJSONPath("data.id"),
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodPost).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.ListAccounts, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.ListAccounts),
 		handler.OpenAPIDef{
 			ID:                  "ListAccounts",
 			Tags:                []string{"cloudintegration"},
@@ -79,20 +64,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbList)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbList,
-			Category: coretypes.ActionCategoryDataAccess,
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.GetAccount, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.GetAccount),
 		handler.OpenAPIDef{
 			ID:                  "GetAccount",
 			Tags:                []string{"cloudintegration"},
@@ -105,21 +84,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{http.StatusBadRequest, http.StatusNotFound},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbRead)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbRead,
-			Category: coretypes.ActionCategoryDataAccess,
-			ID:       coretypes.PathParam("id"),
-			Selector: coretypes.IDSelector,
-		}),
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.UpdateAccount, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.UpdateAccount),
 		handler.OpenAPIDef{
 			ID:                  "UpdateAccount",
 			Tags:                []string{"cloudintegration"},
@@ -132,21 +104,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusNoContent,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbUpdate)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbUpdate,
-			Category: coretypes.ActionCategoryConfigurationChange,
-			ID:       coretypes.PathParam("id"),
-			Selector: coretypes.IDSelector,
-		}),
 	)).Methods(http.MethodPut).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.DisconnectAccount, authtypes.SigNozAdminRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.DisconnectAccount),
 		handler.OpenAPIDef{
 			ID:                  "DisconnectAccount",
 			Tags:                []string{"cloudintegration"},
@@ -159,21 +124,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusNoContent,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegration.Scope(coretypes.VerbDelete)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegration,
-			Verb:     coretypes.VerbDelete,
-			Category: coretypes.ActionCategoryConfigurationChange,
-			ID:       coretypes.PathParam("id"),
-			Selector: coretypes.IDSelector,
-		}),
 	)).Methods(http.MethodDelete).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/services", handler.New(
-		provider.authzMiddleware.OpenAccess(provider.cloudIntegrationHandler.ListServicesMetadata),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.ListServicesMetadata),
 		handler.OpenAPIDef{
 			ID:                  "ListServicesMetadata",
 			Tags:                []string{"cloudintegration"},
@@ -186,14 +144,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes(nil),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}/services", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.ListAccountServicesMetadata, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.ListAccountServicesMetadata),
 		handler.OpenAPIDef{
 			ID:                  "ListAccountServicesMetadata",
 			Tags:                []string{"cloudintegration"},
@@ -206,20 +164,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegrationService.Scope(coretypes.VerbList)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegrationService,
-			Verb:     coretypes.VerbList,
-			Category: coretypes.ActionCategoryDataAccess,
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/services/{service_id}", handler.New(
-		provider.authzMiddleware.OpenAccess(provider.cloudIntegrationHandler.GetService),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.GetService),
 		handler.OpenAPIDef{
 			ID:                  "GetService",
 			Tags:                []string{"cloudintegration"},
@@ -232,14 +184,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes(nil),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}/services/{service_id}", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.UpdateService, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.UpdateService),
 		handler.OpenAPIDef{
 			ID:                  "UpdateService",
 			Tags:                []string{"cloudintegration"},
@@ -252,21 +204,14 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusNoContent,
 			ErrorStatusCodes:    []int{},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegrationService.Scope(coretypes.VerbUpdate)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegrationService,
-			Verb:     coretypes.VerbUpdate,
-			Category: coretypes.ActionCategoryConfigurationChange,
-			ID:       coretypes.PathParam("service_id"),
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodPut).GetError(); err != nil {
 		return err
 	}
 
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}/services/{service_id}", handler.New(
-		provider.authzMiddleware.CheckResources(provider.cloudIntegrationHandler.GetAccountService, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName),
+		provider.authzMiddleware.AdminAccess(provider.cloudIntegrationHandler.GetAccountService),
 		handler.OpenAPIDef{
 			ID:                  "GetAccountService",
 			Tags:                []string{"cloudintegration"},
@@ -279,15 +224,8 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 			SuccessStatusCode:   http.StatusOK,
 			ErrorStatusCodes:    []int{http.StatusBadRequest, http.StatusNotFound},
 			Deprecated:          false,
-			SecuritySchemes:     newScopedSecuritySchemes([]string{coretypes.ResourceMetaResourceCloudIntegrationService.Scope(coretypes.VerbRead)}),
+			SecuritySchemes:     newSecuritySchemes(types.RoleAdmin),
 		},
-		handler.WithResourceDefs(handler.BasicResourceDef{
-			Resource: coretypes.ResourceMetaResourceCloudIntegrationService,
-			Verb:     coretypes.VerbRead,
-			Category: coretypes.ActionCategoryDataAccess,
-			ID:       coretypes.PathParam("service_id"),
-			Selector: coretypes.WildcardSelector,
-		}),
 	)).Methods(http.MethodGet).GetError(); err != nil {
 		return err
 	}
@@ -314,7 +252,6 @@ func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
 		return err
 	}
 
-	// TODO: figure out authz permission model for this endppoint without breaking existing deployed agents.
 	if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/check_in", handler.New(
 		provider.authzMiddleware.ViewAccess(provider.cloudIntegrationHandler.AgentCheckIn),
 		handler.OpenAPIDef{

pkg/modules/tracefunnel/clickhouse_queries.go

@@ -495,8 +495,8 @@ WITH
     toDateTime64(%[3]d/1e9, 9) AS start_ts,
     toDateTime64(%[4]d/1e9, 9) AS end_ts,
 
-    (%[5]s,%[6]s) AS step1,
-    (%[7]s,%[8]s) AS step2
+    ('%[5]s','%[6]s') AS step1,
+    ('%[7]s','%[8]s') AS step2
 
 SELECT
     trace_id,
@@ -527,10 +527,10 @@ LIMIT 5;
 		containsErrorT2,
 		startTs,
 		endTs,
-		clickhousesql.StringLiteral(serviceNameT1),
-		clickhousesql.StringLiteral(spanNameT1),
-		clickhousesql.StringLiteral(serviceNameT2),
-		clickhousesql.StringLiteral(spanNameT2),
+		serviceNameT1,
+		spanNameT1,
+		serviceNameT2,
+		spanNameT2,
 		clauseStep1,
 		clauseStep2,
 		t1TimeExpr,
@@ -571,8 +571,8 @@ WITH
     toDateTime64(%[3]d/1e9, 9) AS start_ts,
     toDateTime64(%[4]d/1e9, 9) AS end_ts,
 
-    (%[5]s,%[6]s) AS step1,
-    (%[7]s,%[8]s) AS step2
+    ('%[5]s','%[6]s') AS step1,
+    ('%[7]s','%[8]s') AS step2
 
 SELECT
     trace_id,
@@ -607,10 +607,10 @@ LIMIT 5;
 		containsErrorT2,
 		startTs,
 		endTs,
-		clickhousesql.StringLiteral(serviceNameT1),
-		clickhousesql.StringLiteral(spanNameT1),
-		clickhousesql.StringLiteral(serviceNameT2),
-		clickhousesql.StringLiteral(spanNameT2),
+		serviceNameT1,
+		spanNameT1,
+		serviceNameT2,
+		spanNameT2,
 		clauseStep1,
 		clauseStep2,
 		t1TimeExpr,

pkg/querier/bucket_cache.go

@@ -55,6 +55,9 @@ func (bc *bucketCache) GetMissRanges(
 	// Get query window
 	startMs, endMs := q.Window()
 
+	stepMs := uint64(step.Milliseconds())
+	startOffsetMs := calculateStartOffset(q, startMs, stepMs)
+
 	bc.logger.DebugContext(ctx, "getting miss ranges", slog.String("fingerprint", q.Fingerprint()), slog.Uint64("start", startMs), slog.Uint64("end", endMs))
 
 	// Generate cache key
@@ -74,11 +77,8 @@ func (bc *bucketCache) GetMissRanges(
 		return nil, missing
 	}
 
-	// Extract step interval if this is a builder query
-	stepMs := uint64(step.Milliseconds())
-
 	// Find missing ranges with step alignment
-	missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs)
+	missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs, startOffsetMs)
 	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
@@ -95,8 +95,7 @@ func (bc *bucketCache) GetMissRanges(
 	// Merge buckets into a single result
 	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, data.Warnings)
 
-	// Filter the merged result to only include values within the requested time range
-	mergedResult = bc.filterResultToTimeRange(mergedResult, startMs, endMs)
+	mergedResult = bc.filterResultToTimeRange(mergedResult, q, startMs, endMs, stepMs)
 
 	return mergedResult, missing
 }
@@ -106,6 +105,9 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 	// Get query window
 	startMs, endMs := q.Window()
 
+	stepMs := uint64(step.Milliseconds())
+	startOffsetMs := calculateStartOffset(q, startMs, stepMs)
+
 	// Calculate the flux boundary - data after this point should not be cached
 	currentMs := uint64(time.Now().UnixMilli())
 	fluxBoundary := currentMs - uint64(bc.fluxInterval.Milliseconds())
@@ -146,19 +148,14 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 
 	// Adjust start and end times to only cache complete intervals
 	cachableStartMs := startMs
-	stepMs := uint64(step.Milliseconds())
 
 	// If we have a step interval, adjust boundaries to only cache complete intervals
 	if stepMs > 0 {
 		// If start is not aligned, round up to next step boundary (first complete interval)
-		if startMs%stepMs != 0 {
-			cachableStartMs = ((startMs / stepMs) + 1) * stepMs
-		}
+		cachableStartMs = alignUpToStep(startMs, stepMs, startOffsetMs)
 
 		// If end is not aligned, round down to previous step boundary (last complete interval)
-		if cachableEndMs%stepMs != 0 {
-			cachableEndMs = (cachableEndMs / stepMs) * stepMs
-		}
+		cachableEndMs = alignDownToStep(cachableEndMs, stepMs, startOffsetMs)
 
 		// If after adjustment we have no complete intervals, don't cache
 		if cachableStartMs >= cachableEndMs {
@@ -206,8 +203,9 @@ func (bc *bucketCache) generateCacheKey(q qbtypes.Query) string {
 	return fmt.Sprintf("v5:query:%s", fingerprint)
 }
 
-// findMissingRangesWithStep identifies time ranges not covered by cached buckets with step alignment.
-func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64) []*qbtypes.TimeRange {
+// findMissingRangesWithStep identifies time ranges not covered by cached buckets
+// with step alignment. Boundaries are whole steps from startOffsetMs.
+func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64, startOffsetMs uint64) []*qbtypes.TimeRange {
 	// When step is 0 or window is too small to be cached, use simple algorithm
 	if stepMs == 0 || (startMs+stepMs) > endMs {
 		return bc.findMissingRangesBasic(buckets, startMs, endMs)
@@ -220,8 +218,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		currentMs := startMs
 
 		// Check if start is not aligned - add partial window
-		if startMs%stepMs != 0 {
-			nextAggStart := startMs - (startMs % stepMs) + stepMs
+		if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
 			missing = append(missing, &qbtypes.TimeRange{
 				From: startMs,
 				To:   min(nextAggStart, endMs),
@@ -267,8 +264,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 	currentMs := startMs
 
 	// Check if start is not aligned - add partial window
-	if startMs%stepMs != 0 {
-		nextAggStart := startMs - (startMs % stepMs) + stepMs
+	if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
 		missing = append(missing, &qbtypes.TimeRange{
 			From: startMs,
 			To:   min(nextAggStart, endMs),
@@ -287,11 +283,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		}
 
 		// Align bucket boundaries to step intervals
-		alignedBucketStart := bucket.StartMs
-		if bucket.StartMs%stepMs != 0 {
-			// Round up to next step boundary
-			alignedBucketStart = bucket.StartMs - (bucket.StartMs % stepMs) + stepMs
-		}
+		alignedBucketStart := alignUpToStep(bucket.StartMs, stepMs, startOffsetMs)
 
 		// Add gap before this bucket if needed
 		if currentMs < alignedBucketStart && currentMs < endMs {
@@ -304,9 +296,12 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		// Update current position to the end of this bucket
 		// But ensure it's aligned to step boundary
 		bucketEnd := min(bucket.EndMs, endMs)
-		if bucketEnd%stepMs != 0 && bucketEnd < endMs {
+		// The step the window ends inside reaches past it, so that stretch is
+		// missing however far the bucket runs.
+		bucketEnd = min(bucketEnd, alignDownToStep(endMs, stepMs, startOffsetMs))
+		if bucketEnd < endMs {
 			// Round down to step boundary
-			bucketEnd = bucketEnd - (bucketEnd % stepMs)
+			bucketEnd = alignDownToStep(bucketEnd, stepMs, startOffsetMs)
 		}
 		currentMs = max(currentMs, bucketEnd)
 	}
@@ -323,6 +318,42 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 	return missing
 }
 
+// calculateStartOffset returns how far into a step a query's values sit. Only
+// promql reports at the window start and every step after it; the rest report
+// on absolute step boundaries.
+func calculateStartOffset(q qbtypes.Query, startMs, stepMs uint64) uint64 {
+	if _, isPromQL := q.(*promqlQuery); !isPromQL || stepMs == 0 {
+		return 0
+	}
+	return startMs % stepMs
+}
+
+// With a 5m step and no offset the times seen by a query are 10:00, 10:05, 10:10. So 10:07
+// is at an offset of 2m, and 10:05 is at 0.
+//
+// With a 1m step and a 30s offset the times seen are 10:00:30, 10:01:30, 10:02:30. So 10:01:00
+// is at an offset of 30s.
+func calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	if stepMs == 0 {
+		return 0
+	}
+	return ((timestampMs % stepMs) + stepMs - startOffsetMs%stepMs) % stepMs
+}
+
+// alignUpToStep returns the first time seen by a query at or after timestampMs.
+func alignUpToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	offset := calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
+	if offset == 0 {
+		return timestampMs
+	}
+	return timestampMs - offset + stepMs
+}
+
+// alignDownToStep returns the last time seen by a query at or before timestampMs.
+func alignDownToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	return timestampMs - calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
+}
+
 // findMissingRangesBasic is the simple algorithm without step alignment.
 func (bc *bucketCache) findMissingRangesBasic(buckets []*qbtypes.CachedBucket, startMs, endMs uint64) []*qbtypes.TimeRange {
 	// Check if already sorted before sorting
@@ -791,12 +822,26 @@ func max(a, b uint64) uint64 {
 	return b
 }
 
-// filterResultToTimeRange filters the result to only include values within the requested time range.
-func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs, endMs uint64) *qbtypes.Result {
+// filterResultToTimeRange narrows the cached result to the requested window, both
+// the values in it and the heatmap axis under them.
+func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes.Query, startMs, endMs, stepMs uint64) *qbtypes.Result {
 	if result == nil || result.Value == nil {
 		return result
 	}
 
+	_, isPromQL := q.(*promqlQuery)
+	maxTimestampMs := endMs
+	// A promql value at T is the query evaluated at T, so T == endMs is inside the
+	// requested range. For every other query type the value at T aggregates
+	// [T, T+stepMs), which the requested range contains only when T <= endMs-stepMs.
+	if !isPromQL {
+		if stepMs > 0 {
+			maxTimestampMs = endMs - stepMs
+		} else {
+			maxTimestampMs = endMs - 1
+		}
+	}
+
 	switch result.Type {
 	case qbtypes.RequestTypeTimeSeries, qbtypes.RequestTypeHeatmap:
 		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
@@ -821,7 +866,7 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs,
 					// Filter values to only include those within the requested time range
 					for _, value := range series.Values {
 						timestampMs := uint64(value.Timestamp)
-						if timestampMs >= startMs && timestampMs < endMs {
+						if timestampMs >= startMs && timestampMs <= maxTimestampMs {
 							filteredSeries.Values = append(filteredSeries.Values, value)
 						}
 					}
@@ -836,6 +881,8 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs,
 				}
 			}
 
+			bc.trimHeatmapAxisToTheWindow(q, filteredData)
+
 			// Create a new result with the filtered data
 			return &qbtypes.Result{
 				Type:     result.Type,
@@ -849,3 +896,20 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs,
 	// For non-time series data, return as is
 	return result
 }
+
+// a cached range covers more than the window now being asked for, so its axis
+// carries buckets only the dropped columns reached. Left there, they show as
+// empty rows the same window never has when the cache did not answer it.
+func (bc *bucketCache) trimHeatmapAxisToTheWindow(q qbtypes.Query, tsData *qbtypes.TimeSeriesData) {
+	// promql and clickhouse name their own buckets, and an empty one of theirs
+	// still belongs on the axis
+	switch q.(type) {
+	case *builderQuery[qbtypes.MetricAggregation], *builderQuery[qbtypes.LogAggregation], *builderQuery[qbtypes.TraceAggregation]:
+	default:
+		return
+	}
+
+	for _, aggBucket := range tsData.Aggregations {
+		aggBucket.TrimAxisToCountedBuckets()
+	}
+}

pkg/querier/bucket_cache_bench_test.go

@@ -201,7 +201,7 @@ func BenchmarkBucketCache_FindMissingRangesWithStep(b *testing.B) {
 			b.ReportAllocs()
 
 			for i := 0; i < b.N; i++ {
-				missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs)
+				missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs, 0)
 				_ = missing
 			}
 		})
@@ -327,7 +327,7 @@ func BenchmarkBucketCache_FilterResultToTimeRange(b *testing.B) {
 			b.ReportAllocs()
 
 			for i := 0; i < b.N; i++ {
-				filtered := bc.filterResultToTimeRange(result, startMs, endMs)
+				filtered := bc.filterResultToTimeRange(result, &promqlQuery{}, startMs, endMs, 0)
 				_ = filtered
 			}
 		})

pkg/querier/bucket_cache_test.go

@@ -3,6 +3,7 @@ package querier
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"testing"
 	"time"
 
@@ -529,7 +530,7 @@ func TestBucketCache_FindMissingRanges_EdgeCases(t *testing.T) {
 	}
 
 	// Query range that spans all buckets
-	missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500)
+	missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500, 0)
 
 	// Expected missing ranges: 500-1000, 2000-2500, 4000-5000, 6000-6500
 	assert.Len(t, missing, 4)
@@ -1069,8 +1070,11 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
 	// Get cached data - should be filtered to requested range
 	cached, missing := bc.GetMissRanges(ctx, orgID, query2, qbtypes.Step{Duration: 1000 * time.Millisecond})
 
-	// Should have no missing ranges
-	assert.Len(t, missing, 0)
+	// The value at 3000 stands for the whole step to 4000, which reaches past the
+	// window, so it is left to be recomputed as a partial rather than served.
+	require.Len(t, missing, 1)
+	assert.Equal(t, uint64(3000), missing[0].From)
+	assert.Equal(t, uint64(3500), missing[0].To)
 	assert.NotNil(t, cached)
 
 	// Verify the cached result only contains values within the requested range
@@ -1080,29 +1084,77 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
 	require.Len(t, tsData.Aggregations[0].Series, 1)
 
 	series := tsData.Aggregations[0].Series[0]
-	assert.Len(t, series.Values, 2) // Only values at 2000 and 3000 should be included
+	require.Len(t, series.Values, 1)
 
 	// Verify the exact values
 	assert.Equal(t, int64(2000), series.Values[0].Timestamp)
 	assert.Equal(t, float64(20), series.Values[0].Value)
-	assert.Equal(t, int64(3000), series.Values[1].Timestamp)
-	assert.Equal(t, float64(30), series.Values[1].Value)
 
 	// Value at 1000 should not be included (before requested range)
 	// Value at 4000 should not be included (after requested range)
 }
 
+// A promql value is the query evaluated at a single moment rather than over a
+// span, so the one at the window's end belongs to it and has to survive caching.
+func TestBucketCache_PromQLKeepsTheValueAtTheWindowEnd(t *testing.T) {
+	bc := createTestBucketCache(t)
+	ctx := context.Background()
+	orgID := valuer.UUID{}
+	step := qbtypes.Step{Duration: time.Minute}
+
+	query := &promqlQuery{
+		logger:      slog.Default(),
+		query:       qbtypes.PromQuery{Query: "up", Step: step},
+		tr:          qbtypes.TimeRange{From: 600_000, To: 780_000},
+		requestType: qbtypes.RequestTypeTimeSeries,
+	}
+
+	bc.Put(ctx, orgID, query, step, &qbtypes.Result{
+		Type: qbtypes.RequestTypeTimeSeries,
+		Value: &qbtypes.TimeSeriesData{
+			QueryName: "A",
+			Aggregations: []*qbtypes.AggregationBucket{{
+				Series: []*qbtypes.TimeSeries{{
+					Values: []*qbtypes.TimeSeriesValue{
+						{Timestamp: 600_000, Value: 1},
+						{Timestamp: 660_000, Value: 2},
+						{Timestamp: 720_000, Value: 3},
+						{Timestamp: 780_000, Value: 4},
+					},
+				}},
+			}},
+		},
+	})
+	time.Sleep(10 * time.Millisecond)
+
+	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
+	assert.Empty(t, missing)
+	require.NotNil(t, cached)
+
+	tsData, ok := cached.Value.(*qbtypes.TimeSeriesData)
+	require.True(t, ok)
+	require.Len(t, tsData.Aggregations, 1)
+	require.Len(t, tsData.Aggregations[0].Series, 1)
+
+	timestamps := []int64{}
+	for _, value := range tsData.Aggregations[0].Series[0].Values {
+		timestamps = append(timestamps, value.Timestamp)
+	}
+	assert.Equal(t, []int64{600_000, 660_000, 720_000, 780_000}, timestamps)
+}
+
 func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 	bc := createTestBucketCache(t)
 
 	tests := []struct {
-		name         string
-		buckets      []*qbtypes.CachedBucket
-		startMs      uint64
-		endMs        uint64
-		stepMs       uint64
-		expectedMiss []*qbtypes.TimeRange
-		description  string
+		name          string
+		buckets       []*qbtypes.CachedBucket
+		startMs       uint64
+		endMs         uint64
+		stepMs        uint64
+		startOffsetMs uint64
+		expectedMiss  []*qbtypes.TimeRange
+		description   string
 	}{
 		{
 			name:    "start_not_aligned_to_step",
@@ -1152,6 +1204,32 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 			},
 			description: "Window smaller than step should use basic algorithm",
 		},
+		{
+			name:          "start_aligned_to_its_own_offset",
+			buckets:       []*qbtypes.CachedBucket{},
+			startMs:       1500,
+			endMs:         5000,
+			stepMs:        1000,
+			startOffsetMs: 500,
+			expectedMiss: []*qbtypes.TimeRange{
+				{From: 1500, To: 5000},
+			},
+			description: "A query reporting every 1000ms from 1500 needs no partial window at its own start",
+		},
+		{
+			name: "gap_lands_on_the_offset",
+			buckets: []*qbtypes.CachedBucket{
+				{StartMs: 1500, EndMs: 3500},
+			},
+			startMs:       1500,
+			endMs:         5500,
+			stepMs:        1000,
+			startOffsetMs: 500,
+			expectedMiss: []*qbtypes.TimeRange{
+				{From: 3500, To: 5500},
+			},
+			description: "The refetched range starts where the cached one ends, on an instant the query reports at",
+		},
 		{
 			name:    "zero_step_uses_basic_algorithm",
 			buckets: []*qbtypes.CachedBucket{},
@@ -1168,7 +1246,7 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// Mock current time for flux boundary tests
-			result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs)
+			result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs, tt.startOffsetMs)
 
 			// Compare lengths first
 			assert.Len(t, result, len(tt.expectedMiss), tt.description)

pkg/querier/promql_heatmap.go

@@ -22,7 +22,7 @@ import (
 const promHistogramBucketLabel = "le"
 
 // cumulativeColumn maps a bucket's upper bound to the cumulative count at it.
-// Differencing turns it into the per-band counts a heatmapColumn holds.
+// Differencing turns it into the per-bucket counts a heatmapColumn holds.
 type cumulativeColumn map[float64]float64
 
 // promHeatmapGroup assembles one group across the several matrix series its `le`
@@ -34,8 +34,8 @@ type promHeatmapGroup struct {
 }
 
 // foldMatrixAsHeatmap folds a matrix of one cumulative series per (group, `le`)
-// into one series per group whose points hold a count per band.
-func foldMatrixAsHeatmap(matrix promql.Matrix, queryWindow *qbv5.TimeRange, stepMs uint64, queryName string) (*qbv5.TimeSeriesData, error) {
+// into one series per group whose points hold a count per bucket.
+func foldMatrixAsHeatmap(matrix promql.Matrix, queryName string) (*qbv5.TimeSeriesData, error) {
 	groups, groupOrder := collectCumulativeGroups(matrix)
 
 	// An empty matrix is only ever the window having no data, but series that
@@ -53,11 +53,12 @@ func foldMatrixAsHeatmap(matrix promql.Matrix, queryWindow *qbv5.TimeRange, step
 		}
 	}
 
-	return accumulator.foldSeries(queryWindow, stepMs, queryName)
+	// a promql data point can never be partial, hence nil and 0 are sent here
+	return accumulator.foldSeries(nil, 0, queryName)
 }
 
 // collectCumulativeGroups reads the matrix into one group per label set. A series
-// without `le` has no band to sit in, so an expression that dropped the label
+// without `le` has no bucket to sit in, so an expression that dropped the label
 // draws nothing.
 func collectCumulativeGroups(matrix promql.Matrix) (groups map[string]*promHeatmapGroup, groupOrder []string) {
 	groups = map[string]*promHeatmapGroup{}

pkg/querier/promql_heatmap_test.go

@@ -16,7 +16,7 @@ import (
 
 // The cache key is the fingerprint alone, so two request types over one
 // expression must not produce the same one — a time series payload served to a
-// heatmap request has no axis and reads back as a single collapsed band.
+// heatmap request has no axis and reads back as a single collapsed bucket.
 func TestFingerprintSeparatesHeatmapFromTimeSeries(t *testing.T) {
 	fingerprintFor := func(requestType qbv5.RequestType) string {
 		q := &promqlQuery{
@@ -50,7 +50,7 @@ func TestFoldMatrixAsHeatmapClampsADecreasingCumulativeCount(t *testing.T) {
 		},
 	}
 
-	data, err := foldMatrixAsHeatmap(matrix, &qbv5.TimeRange{From: 1710000000000, To: 1710000060000}, uint64(time.Minute.Milliseconds()), "A")
+	data, err := foldMatrixAsHeatmap(matrix, "A")
 	require.NoError(t, err)
 	require.Len(t, data.Aggregations, 1)
 
@@ -76,7 +76,7 @@ func TestFoldMatrixAsHeatmapWidensTheBandOverAMissingUpperBound(t *testing.T) {
 		},
 	}
 
-	data, err := foldMatrixAsHeatmap(matrix, &qbv5.TimeRange{From: 1710000000000, To: 1710000060000}, uint64(time.Minute.Milliseconds()), "A")
+	data, err := foldMatrixAsHeatmap(matrix, "A")
 	require.NoError(t, err)
 	require.Len(t, data.Aggregations, 1)
 

pkg/querier/promql_query.go

@@ -175,6 +175,12 @@ func (q *promqlQuery) Fingerprint() string {
 		q.query.Step.String(),
 	}
 
+	// Two windows a fraction of a step apart describe different instants, so
+	// they must not share an entry.
+	if stepMs := uint64(q.query.Step.Milliseconds()); stepMs > 0 && q.tr.From%stepMs != 0 {
+		parts = append(parts, fmt.Sprintf("offset=%d", q.tr.From%stepMs))
+	}
+
 	return strings.Join(parts, "&")
 }
 
@@ -485,7 +491,7 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
 }
 
 func (q *promqlQuery) toResultForHeatmap(matrix promql.Matrix, warnings []string, began time.Time, statsMu *sync.Mutex, rowsScanned, bytesScanned *uint64) (*qbv5.Result, error) {
-	tsData, err := foldMatrixAsHeatmap(matrix, &q.tr, uint64(q.query.Step.Milliseconds()), q.query.Name)
+	tsData, err := foldMatrixAsHeatmap(matrix, q.query.Name)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/promql_query_test.go

@@ -461,6 +461,37 @@ func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
 	assert.Empty(t, q.Fingerprint())
 }
 
+// promql reports at the window start and every step after it, so a window
+// starting later inside the step describes instants the earlier one never does.
+func TestFingerprintSeparatesWindowsInsideAStep(t *testing.T) {
+	minuteStep := qbv5.Step{Duration: time.Minute}
+
+	onTheMinute := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 600_000, To: 1_200_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	halfAStepLater := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 630_000, To: 1_230_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	aWholeMinuteLater := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 900_000, To: 1_500_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	require.NotEmpty(t, onTheMinute)
+	assert.NotEqual(t, onTheMinute, halfAStepLater, "windows half a step apart share no instants")
+	assert.Equal(t, onTheMinute, aWholeMinuteLater, "windows whole steps apart report at the same instants")
+}
+
 func TestToResultDropsNonFiniteValues(t *testing.T) {
 	tests := []struct {
 		description        string

pkg/querybuilder/key_resolution.go

@@ -25,9 +25,8 @@ const (
 // ResolveLogicalFields picks which logical fields a filter term builds conditions
 // for. With 0 or 1 field it returns the input unchanged and no warning. When a
 // name is ambiguous (several logical fields — a family is one field and never
-// ambiguous with itself) it returns a warning; a resource + other-context mix
-// (attribute, body, scope, …) defaults to the resource fields (the common
-// intent), noted in the warning.
+// ambiguous with itself) it returns a warning; a resource+attribute mix defaults
+// to the resource fields (the common intent), noted in the warning.
 func ResolveLogicalFields(field *telemetrytypes.TelemetryFieldKey, logicalFields []*telemetrytypes.LogicalField) ([]*telemetrytypes.LogicalField, string) {
 	if len(logicalFields) <= 1 {
 		return logicalFields, ""
@@ -40,17 +39,18 @@ func ResolveLogicalFields(field *telemetrytypes.TelemetryFieldKey, logicalFields
 		logicalFields,
 	)
 
-	hasResource, hasOther := false, false
+	hasResource, hasAttribute := false, false
 	for _, item := range logicalFields {
-		if item.FieldContext == telemetrytypes.FieldContextResource {
+		switch item.FieldContext {
+		case telemetrytypes.FieldContextResource:
 			hasResource = true
-		} else {
-			hasOther = true
+		case telemetrytypes.FieldContextAttribute:
+			hasAttribute = true
 		}
 	}
 
-	// with resource and any other context, default to resource only
-	if hasResource && hasOther {
+	// when there is both resource and attribute context, default to resource only
+	if hasResource && hasAttribute {
 		filtered := make([]*telemetrytypes.LogicalField, 0, len(logicalFields))
 		for _, item := range logicalFields {
 			if item.FieldContext == telemetrytypes.FieldContextResource {
@@ -58,8 +58,8 @@ func ResolveLogicalFields(field *telemetrytypes.TelemetryFieldKey, logicalFields
 			}
 		}
 		logicalFields = filtered
-		warning += " " + "Using `resource` context by default. To query another context explicitly, " +
-			fmt.Sprintf("use the fully qualified name (e.g., 'attribute.%s' or 'body.%s')", field.Name, field.Name)
+		warning += " " + "Using `resource` context by default. To query attributes explicitly, " +
+			fmt.Sprintf("use the fully qualified name (e.g., 'attribute.%s')", field.Name)
 	}
 
 	return logicalFields, warning

pkg/querybuilder/logical_fields_test.go

@@ -175,42 +175,6 @@ func TestResolveLogicalFieldsKeepsFamilyThroughAmbiguity(t *testing.T) {
 	assert.Equal(t, []string{"deployment.environment.name", "deployment.environment"}, memberNames(resolved[0]))
 }
 
-// Resource wins over every other context, not just attribute: a bare key that
-// also lives in body or scope must collapse to resource alone, so the surviving
-// candidate does not AND against the resource fingerprint CTE.
-func TestResolveLogicalFieldsResourceWinsOverOtherContexts(t *testing.T) {
-	testCases := []struct {
-		name  string
-		other telemetrytypes.FieldContext
-	}{
-		{name: "ResourceOverBody", other: telemetrytypes.FieldContextBody},
-		{name: "ResourceOverScope", other: telemetrytypes.FieldContextScope},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			requested := &telemetrytypes.TelemetryFieldKey{Name: "service.name"}
-			fields := []*telemetrytypes.LogicalField{
-				telemetrytypes.SingleLogicalField("service.name", &telemetrytypes.TelemetryFieldKey{
-					Name:          "service.name",
-					FieldContext:  telemetrytypes.FieldContextResource,
-					FieldDataType: telemetrytypes.FieldDataTypeString,
-				}),
-				telemetrytypes.SingleLogicalField("service.name", &telemetrytypes.TelemetryFieldKey{
-					Name:          "service.name",
-					FieldContext:  testCase.other,
-					FieldDataType: telemetrytypes.FieldDataTypeString,
-				}),
-			}
-
-			resolved, warning := ResolveLogicalFields(requested, fields)
-			assert.NotEmpty(t, warning)
-			require.Len(t, resolved, 1)
-			assert.Equal(t, telemetrytypes.FieldContextResource, resolved[0].FieldContext)
-		})
-	}
-}
-
 // Members of a family with different data types never merge: the identity
 // (signal, context, data type) separates them into distinct logical fields.
 func TestMatchingLogicalFieldsNeverMergesAcrossDataTypes(t *testing.T) {

pkg/signoz/provider.go

@@ -254,7 +254,6 @@ func NewSQLMigrationProviderFactories(
 		sqlmigration.NewAddIngestionTuplesFactory(sqlstore),
 		sqlmigration.NewAddSubscriptionTuplesFactory(sqlstore),
 		sqlmigration.NewNormalizeQuickFilterFieldsFactory(sqlstore),
-		sqlmigration.NewAddCloudIntegrationTuplesFactory(sqlstore),
 	)
 }
 

pkg/sqlmigration/127_add_cloud_integration_tuples.go

@@ -1,175 +0,0 @@
-package sqlmigration
-
-import (
-	"context"
-	"database/sql"
-	"encoding/json"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/sqlstore"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/types/coretypes"
-	"github.com/oklog/ulid/v2"
-	"github.com/uptrace/bun"
-	"github.com/uptrace/bun/dialect"
-	"github.com/uptrace/bun/migrate"
-)
-
-type addCloudIntegrationTuples struct {
-	sqlstore sqlstore.SQLStore
-}
-
-func NewAddCloudIntegrationTuplesFactory(sqlstore sqlstore.SQLStore) factory.ProviderFactory[SQLMigration, Config] {
-	return factory.NewProviderFactory(factory.MustNewName("add_cloud_integration_tuples"), func(ctx context.Context, ps factory.ProviderSettings, c Config) (SQLMigration, error) {
-		return &addCloudIntegrationTuples{sqlstore: sqlstore}, nil
-	})
-}
-
-func (migration *addCloudIntegrationTuples) Register(migrations *migrate.Migrations) error {
-	return migrations.Register(migration.Up, migration.Down)
-}
-
-func (migration *addCloudIntegrationTuples) Up(ctx context.Context, db *bun.DB) error {
-	tx, err := db.BeginTx(ctx, nil)
-	if err != nil {
-		return err
-	}
-	defer func() { _ = tx.Rollback() }()
-
-	var storeID string
-	err = tx.QueryRowContext(ctx, `SELECT id FROM store WHERE name = ? LIMIT 1`, "signoz").Scan(&storeID)
-	if err != nil {
-		return err
-	}
-
-	var orgIDs []string
-	err = tx.NewSelect().
-		Table("organizations").
-		Column("id").
-		Scan(ctx, &orgIDs)
-	if err != nil && err != sql.ErrNoRows {
-		return err
-	}
-
-	isPG := migration.sqlstore.BunDB().Dialect().Name() == dialect.PG
-
-	// cloud-integration and cloud-integration-service moved from legacy role
-	// gates to CheckResources. Existing organizations need the same tuples that
-	// new organizations receive from the managed-role registry at bootstrap.
-	tuples := []migrationTuple{
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration", "create"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration", "read"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration", "update"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration", "delete"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration", "list"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration-service", "read"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration-service", "update"},
-		{authtypes.SigNozAdminRoleName, "metaresource", "cloud-integration-service", "list"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration", "read"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration", "update"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration", "list"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration-service", "read"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration-service", "update"},
-		{authtypes.SigNozEditorRoleName, "metaresource", "cloud-integration-service", "list"},
-		{authtypes.SigNozViewerRoleName, "metaresource", "cloud-integration", "read"},
-		{authtypes.SigNozViewerRoleName, "metaresource", "cloud-integration", "list"},
-		{authtypes.SigNozViewerRoleName, "metaresource", "cloud-integration-service", "read"},
-		{authtypes.SigNozViewerRoleName, "metaresource", "cloud-integration-service", "list"},
-	}
-
-	for _, orgID := range orgIDs {
-		for _, tuple := range tuples {
-			entropy := ulid.DefaultEntropy()
-			now := time.Now().UTC()
-			tupleID := ulid.MustNew(ulid.Timestamp(now), entropy).String()
-
-			objectID := "organization/" + orgID + "/" + tuple.objectName + "/*"
-			roleSubject := "organization/" + orgID + "/role/" + tuple.roleName
-
-			if isPG {
-				user := "role:" + roleSubject + "#assignee"
-				result, err := tx.ExecContext(ctx, `
-					INSERT INTO tuple (store, object_type, object_id, relation, _user, user_type, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, object_type, object_id, relation, _user) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, user, "userset", tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-				rowsAffected, err := result.RowsAffected()
-				if err != nil {
-					return err
-				}
-				if rowsAffected == 0 {
-					continue
-				}
-				_, err = tx.ExecContext(ctx, `
-					INSERT INTO changelog (store, object_type, object_id, relation, _user, operation, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, ulid, object_type) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, user, 0, tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-			} else {
-				result, err := tx.ExecContext(ctx, `
-					INSERT INTO tuple (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation, user_type, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, "role", roleSubject, "assignee", "userset", tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-				rowsAffected, err := result.RowsAffected()
-				if err != nil {
-					return err
-				}
-				if rowsAffected == 0 {
-					continue
-				}
-				_, err = tx.ExecContext(ctx, `
-					INSERT INTO changelog (store, object_type, object_id, relation, user_object_type, user_object_id, user_relation, operation, ulid, inserted_at)
-					VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-					ON CONFLICT (store, ulid, object_type) DO NOTHING`,
-					storeID, tuple.objectType, objectID, tuple.relation, "role", roleSubject, "assignee", 0, tupleID, now,
-				)
-				if err != nil {
-					return err
-				}
-			}
-		}
-	}
-
-	managedRoleGroups := make(map[string]string, len(coretypes.ManagedRoleToTransactions))
-	for roleName, transactions := range coretypes.ManagedRoleToTransactions {
-		data, err := json.Marshal(authtypes.NewTransactionGroupsFromTransactions(transactions))
-		if err != nil {
-			return err
-		}
-		managedRoleGroups[roleName] = string(data)
-	}
-
-	for _, orgID := range orgIDs {
-		for roleName, data := range managedRoleGroups {
-			if _, err := tx.NewUpdate().
-				Model(new(roles)).
-				Set("transaction_groups = ?", data).
-				Where("org_id = ?", orgID).
-				Where("type = ?", authtypes.RoleTypeManaged.StringValue()).
-				Where("name = ?", roleName).
-				Exec(ctx); err != nil {
-				return err
-			}
-		}
-	}
-
-	return tx.Commit()
-}
-
-func (migration *addCloudIntegrationTuples) Down(context.Context, *bun.DB) error {
-	return nil
-}

pkg/statementbuilder/logsstatementbuilder/resource_conflict_test.go

@@ -1,90 +0,0 @@
-package logsstatementbuilder
-
-import (
-	"context"
-	"testing"
-
-	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
-	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
-	"github.com/SigNoz/signoz/pkg/statementbuilder"
-	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
-	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/stretchr/testify/require"
-)
-
-// A key present in both resource and body contexts must filter on resource only.
-// The resource condition builds the fingerprint CTE, so a surviving body condition
-// would AND against it and match almost nothing (engineering-pod#6086).
-func TestStatementBuilderResourceBodyConflict(t *testing.T) {
-	store := telemetrytypestest.NewMockMetadataStore()
-	store.SetStaticFields(logstelemetryschema.IntrinsicFields)
-	store.SetKey(&telemetrytypes.TelemetryFieldKey{
-		Name:          "service.name",
-		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextResource,
-		FieldDataType: telemetrytypes.FieldDataTypeString,
-	})
-	bodyKey := &telemetrytypes.TelemetryFieldKey{
-		Name:          "service.name",
-		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextBody,
-		FieldDataType: telemetrytypes.FieldDataTypeString,
-	}
-	require.NoError(t, bodyKey.SetJSONAccessPlan(telemetrytypes.JSONColumnMetadata{
-		BaseColumn:     logstelemetryschema.LogsV2BodyV2Column,
-		PromotedColumn: logstelemetryschema.LogsV2BodyPromotedColumn,
-	}, map[string][]telemetrytypes.FieldDataType{"service.name": {telemetrytypes.FieldDataTypeString}}))
-	store.SetKey(bodyKey)
-
-	fl := flaggertest.WithUseJSONBody(t, true)
-	storage := logstelemetryschema.NewStorage()
-	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
-	statementBuilder := NewLogQueryStatementBuilder(
-		instrumentationtest.New().ToProviderSettings(),
-		store,
-		storage,
-		aggExprRewriter,
-		logstelemetryschema.DefaultFullTextColumn,
-		fl,
-		nil,
-		statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
-	)
-
-	testCases := []struct {
-		name        string
-		requestType qbtypes.RequestType
-		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
-		expected    qbtypes.Statement
-	}{
-		{
-			name:        "AmbiguousKeyFiltersResourceOnly",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Filter: &qbtypes.Filter{Expression: "service.name = 'webapp'"},
-				Limit:  10,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{"webapp", "%service.name%", "%service.name\":\"webapp%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-				Warnings: []string{
-					"Key `service.name` is ambiguous, found 2 different combinations of field context / data type: [name=service.name,context=resource,datatype=string name=service.name,context=body,datatype=string]. Using `resource` context by default. To query another context explicitly, use the fully qualified name (e.g., 'attribute.service.name' or 'body.service.name')",
-				},
-			},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
-			require.NoError(t, err)
-			require.Equal(t, testCase.expected.Query, q.Query)
-			require.Equal(t, testCase.expected.Args, q.Args)
-			require.Equal(t, testCase.expected.Warnings, q.Warnings)
-		})
-	}
-}

pkg/telemetrystore/config.go

@@ -46,6 +46,7 @@ type QuerySettings struct {
 	MaxBytesToRead                      int    `mapstructure:"max_bytes_to_read"`
 	MaxResultRows                       int    `mapstructure:"max_result_rows"`
 	IgnoreDataSkippingIndices           string `mapstructure:"ignore_data_skipping_indices"`
+	SecondaryIndicesEnableBulkFiltering bool   `mapstructure:"secondary_indices_enable_bulk_filtering"`
 }
 
 func NewConfigFactory() factory.ConfigFactory {

pkg/telemetrystore/telemetrystorehook/settings.go

@@ -72,6 +72,10 @@ func (h *provider) BeforeQuery(ctx context.Context, _ *telemetrystore.QueryEvent
 		settings["result_overflow_mode"] = ctx.Value("result_overflow_mode")
 	}
 
+	// TODO(srikanthccv): enable it when the "Cannot read all data" issue is fixed
+	// https://github.com/ClickHouse/ClickHouse/issues/82283
+	settings["secondary_indices_enable_bulk_filtering"] = false
+
 	ctx = clickhouse.Context(ctx, clickhouse.WithSettings(settings))
 	return ctx
 }

pkg/types/alertmanagertypes/channelconfig.go

@@ -1014,7 +1014,7 @@ func rejectHTTPBasicAuthBeyondPassword(channelName string, httpConfig *commoncfg
 
 	basicAuth := httpConfig.BasicAuth
 	if *basicAuth != (commoncfg.BasicAuth{Username: basicAuth.Username, Password: basicAuth.Password}) {
-		return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.basic_auth with fields other than username and password, which is not supported", channelName)
+		return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.basic_auth, which is not supported", channelName)
 	}
 
 	return nil
@@ -1026,8 +1026,8 @@ func rejectHTTPAuthorizationBeyondBearer(channelName string, httpConfig *commonc
 	}
 
 	authorization := httpConfig.Authorization
-	if !strings.EqualFold(authorization.Type, bearerAuthorizationType) || *authorization != (commoncfg.Authorization{Type: authorization.Type, Credentials: authorization.Credentials}) {
-		return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.authorization with fields other than a bearer token, which is not supported", channelName)
+	if *authorization != (commoncfg.Authorization{Type: bearerAuthorizationType, Credentials: authorization.Credentials}) {
+		return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.authorization, which is not supported", channelName)
 	}
 
 	return nil

pkg/types/alertmanagertypes/channelconvertors_test.go

@@ -542,42 +542,3 @@ func TestChannelToPostableChannelRejectsUnrepresentableChannels(t *testing.T) {
 		})
 	}
 }
-
-// The HTTP auth scheme is case-insensitive (RFC 7235) and Alertmanager sends
-// the stored spelling verbatim, so a hand-written receiver may carry any casing.
-func TestChannelToPostableChannelReadsWebhookBearerSchemeCaseInsensitively(t *testing.T) {
-	sendResolved := config.DefaultWebhookConfig.VSendResolved
-
-	testCases := []struct {
-		name                string
-		storedChannelData   string
-		expectedWebhookSpec *ChannelWebhookConfig
-	}{
-		{
-			name:                "CanonicalBearer",
-			storedChannelData:   `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://a","http_config":{"authorization":{"type":"Bearer","credentials":"tok"},"follow_redirects":true,"enable_http2":true}}]}`,
-			expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://a", BearerToken: "tok"},
-		},
-		{
-			name:                "LowercaseBearer",
-			storedChannelData:   `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://b","http_config":{"authorization":{"type":"bearer","credentials":"lower"},"follow_redirects":true,"enable_http2":true}}]}`,
-			expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://b", BearerToken: "lower"},
-		},
-		{
-			name:                "UppercaseBearer",
-			storedChannelData:   `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://c","http_config":{"authorization":{"type":"BEARER","credentials":"upper"},"follow_redirects":true,"enable_http2":true}}]}`,
-			expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://c", BearerToken: "upper"},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			channel := Channel{DisplayName: "hook", Data: testCase.storedChannelData}
-
-			postable, err := channel.toPostableNotificationChannel()
-			require.NoError(t, err)
-			assert.Equal(t, ChannelKindWebhook, postable.Config.Kind)
-			assert.Equal(t, testCase.expectedWebhookSpec, postable.Config.Spec)
-		})
-	}
-}

pkg/types/coretypes/registry_managed_role.go

@@ -35,15 +35,17 @@ var ManagedRoleToTransactions = map[string][]Transaction{
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindAuthDomain}, WildCardSelectorString)},
 		{Verb: VerbAttach, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindAuthDomain}, WildCardSelectorString)},
 		{Verb: VerbDetach, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindAuthDomain}, WildCardSelectorString)},
-		// cloud-integration — admin can fully manage accounts
+		// cloud-integration — admin only
 		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
 		{Verb: VerbUpdate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
 		{Verb: VerbDelete, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
 		{Verb: VerbCreate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		// cloud-integration-service — admin can read and update account services
+		// cloud-integration-service — admin only
 		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
 		{Verb: VerbUpdate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
+		{Verb: VerbDelete, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
+		{Verb: VerbCreate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
 		// integration — viewer/editor/admin (install/uninstall via ViewAccess)
 		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindIntegration}, WildCardSelectorString)},
@@ -214,14 +216,6 @@ var ManagedRoleToTransactions = map[string][]Transaction{
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindTracesField}, WildCardSelectorString)},
 	},
 	SigNozEditorRoleName: {
-		// cloud-integration — editor can read and update existing accounts
-		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		{Verb: VerbUpdate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		// cloud-integration-service — editor can read and update account services
-		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
-		{Verb: VerbUpdate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
-		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
 		// dashboard — full CRUD
 		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindDashboard}, WildCardSelectorString)},
 		{Verb: VerbUpdate, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindDashboard}, WildCardSelectorString)},
@@ -314,12 +308,6 @@ var ManagedRoleToTransactions = map[string][]Transaction{
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindTracesField}, WildCardSelectorString)},
 	},
 	SigNozViewerRoleName: {
-		// cloud-integration — viewer can read accounts
-		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegration}, WildCardSelectorString)},
-		// cloud-integration-service — viewer can read account services
-		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
-		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindCloudIntegrationService}, WildCardSelectorString)},
 		// dashboard — read only
 		{Verb: VerbRead, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindDashboard}, WildCardSelectorString)},
 		{Verb: VerbList, Object: *MustNewObject(ResourceRef{Type: TypeMetaResource, Kind: KindDashboard}, WildCardSelectorString)},

pkg/types/coretypes/registry_resource.go

@@ -52,8 +52,8 @@ var (
 	ResourceMetaResourceApdexSetting                     = NewResourceMetaResource(KindApdexSetting)
 	ResourceMetaResourceAuthDomain                       = NewResourceMetaResource(KindAuthDomain)
 	ResourceMetaResourceSession                          = NewResourceMetaResource(KindSession)
-	ResourceMetaResourceCloudIntegration                 = NewResourceMetaResource(KindCloudIntegration, VerbCreate, VerbList, VerbRead, VerbUpdate, VerbDelete)
-	ResourceMetaResourceCloudIntegrationService          = NewResourceMetaResource(KindCloudIntegrationService, VerbList, VerbRead, VerbUpdate)
+	ResourceMetaResourceCloudIntegration                 = NewResourceMetaResource(KindCloudIntegration)
+	ResourceMetaResourceCloudIntegrationService          = NewResourceMetaResource(KindCloudIntegrationService)
 	ResourceMetaResourceIntegration                      = NewResourceMetaResource(KindIntegration)
 	ResourceMetaResourceDashboard                        = NewResourceMetaResource(KindDashboard, VerbCreate, VerbList, VerbRead, VerbUpdate, VerbDelete)
 	ResourceMetaResourcePublicDashboard                  = NewResourceMetaResource(KindPublicDashboard)

pkg/types/querybuildertypes/querybuildertypesv5/resp.go

@@ -189,6 +189,50 @@ func (a *AggregationBucket) ReindexValuesToNewUpperBounds(onto []float64) {
 	a.Meta.Buckets = onto
 }
 
+// TrimAxisToCountedBuckets drops the buckets at either end of Meta.Buckets that hold
+// no counts, since an axis runs from the lowest value in the window to the highest.
+// Not for a query that chose its own buckets: an empty `le` is still one it reported.
+func (a *AggregationBucket) TrimAxisToCountedBuckets() {
+	if a == nil || len(a.Meta.Buckets) == 0 {
+		return
+	}
+
+	lowestCounted, highestCounted := len(a.Meta.Buckets), -1
+	for _, series := range a.Series {
+		for _, point := range series.Values {
+			for slot := 0; slot < len(a.Meta.Buckets) && slot < len(point.Values); slot++ {
+				if point.Values[slot] != 0 {
+					lowestCounted = min(lowestCounted, slot)
+					highestCounted = max(highestCounted, slot)
+				}
+			}
+		}
+	}
+	if highestCounted < 0 {
+		return
+	}
+
+	if lowestCounted == 0 && highestCounted == len(a.Meta.Buckets)-1 {
+		return
+	}
+
+	trimmed := a.Meta.Buckets[lowestCounted : highestCounted+1]
+	for _, series := range a.Series {
+		for _, point := range series.Values {
+			if len(point.Values) == 0 {
+				continue
+			}
+			counts := make([]float64, len(trimmed)+1)
+			for slot, count := range point.Values {
+				counts[min(max(slot-lowestCounted, 0), len(trimmed))] += count
+			}
+			point.Values = counts
+		}
+	}
+
+	a.Meta.Buckets = trimmed
+}
+
 type AggregationMeta struct {
 	Unit string `json:"unit,omitempty"`
 	// Buckets holds ascending upper bounds shared by every series in the

tests/integration/tests/promqlconformance/03_nonfinite_cache.py

@@ -58,8 +58,7 @@ def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
     assert set(first["active_job"].values()) == {25.0}, sorted(set(first["active_job"].values()))
     assert len(first["active_job"]) == expected_points, f"expected {expected_points} points, got {len(first['active_job'])}"
 
-    # The cached read excludes end_ms, the one legitimate difference.
+    # Both reads must agree exactly, including the point promql reports at end_ms.
     assert set(second) == set(first), sorted(second)
     for job_name, points in first.items():
-        expected = {ts: value for ts, value in points.items() if ts < end_ms}
-        assert second[job_name] == expected, f"{job_name}: got {len(second[job_name])} of {len(expected)} points"
+        assert second[job_name] == points, f"{job_name}: got {len(second[job_name])} of {len(points)} points"

tests/integration/tests/querier_json_body/08_resource_body_conflict.py

@@ -1,88 +0,0 @@
-import json
-from collections.abc import Callable
-from datetime import UTC, datetime, timedelta
-from http import HTTPStatus
-
-from fixtures import types
-from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.logs import Logs
-from fixtures.querier import (
-    build_raw_query,
-    get_rows,
-    make_query_request,
-)
-
-
-def test_resource_body_conflict(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_logs: Callable[[list[Logs]], None],
-    export_json_types: Callable[[list[Logs]], None],
-) -> None:
-    now = datetime.now(tz=UTC)
-    start_ms = int((now - timedelta(seconds=10)).timestamp() * 1000)
-    end_ms = int(now.timestamp() * 1000)
-
-    # python's body carries service.name, making the bare key ambiguous across
-    # resource and body; java's body omits it, so ANDing body in would drop it.
-    logs_list = [
-        Logs(
-            timestamp=now - timedelta(seconds=2),
-            resources={"service.name": "java"},
-            body_v2=json.dumps({"msg": "hello"}),
-            body_promoted="",
-        ),
-        Logs(
-            timestamp=now - timedelta(seconds=1),
-            resources={"service.name": "python"},
-            body_v2=json.dumps({"service.name": "python"}),
-            body_promoted="",
-        ),
-    ]
-    export_json_types(logs_list)
-    insert_logs(logs_list)
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    cases = [
-        {
-            "name": "bare_key_resolves_to_resource",
-            "filter": "service.name = 'java'",
-            "expected_service_names": ["java"],
-            "expect_resource_warning": True,
-        },
-        {
-            "name": "qualified_body_key_targets_body",
-            "filter": "body.service.name = 'python'",
-            "expected_service_names": ["python"],
-            "expect_resource_warning": False,
-        },
-    ]
-
-    for case in cases:
-        response = make_query_request(
-            signoz,
-            token,
-            start_ms,
-            end_ms,
-            request_type="raw",
-            queries=[
-                build_raw_query(
-                    name="A",
-                    signal="logs",
-                    filter_expression=case["filter"],
-                    limit=100,
-                    step_interval=60,
-                )
-            ],
-        )
-
-        assert response.status_code == HTTPStatus.OK, f"{case['name']}: {response.text}"
-        rows = get_rows(response)
-        assert [row["data"]["resources_string"].get("service.name") for row in rows] == case["expected_service_names"], f"{case['name']}: {response.json()}"
-
-        warning = response.json()["data"].get("warning")
-        if case["expect_resource_warning"]:
-            assert warning is not None and "Using `resource` context by default" in warning["warnings"][0]["message"], f"{case['name']}: {warning}"
-        else:
-            assert warning is None, f"{case['name']}: {warning}"

tests/integration/tests/queriercommon/02_qb_warnings.py

@@ -64,8 +64,8 @@ def test_resource_default_warning(
         "Key `service.name` is ambiguous, found 2 different combinations of "
         "field context / data type: [name=service.name,context=resource,datatype=string "
         "name=service.name,context=attribute,datatype=string]. Using `resource` context "
-        "by default. To query another context explicitly, use the fully qualified name "
-        "(e.g., 'attribute.service.name' or 'body.service.name')"
+        "by default. To query attributes explicitly, use the fully qualified name "
+        "(e.g., 'attribute.service.name')"
     )
     assert warning["warnings"] == [
         {"message": expected_service_name_warning},
@@ -237,8 +237,8 @@ def test_deduped_warnings_for_single_query(
         "Key `service.name` is ambiguous, found 2 different combinations of "
         "field context / data type: [name=service.name,context=resource,datatype=string "
         "name=service.name,context=attribute,datatype=string]. Using `resource` context "
-        "by default. To query another context explicitly, use the fully qualified name "
-        "(e.g., 'attribute.service.name' or 'body.service.name')"
+        "by default. To query attributes explicitly, use the fully qualified name "
+        "(e.g., 'attribute.service.name')"
     )
     expected_status_code_warning = "Key `http.status_code` is ambiguous, found 2 different combinations of field context / data type: [name=http.status_code,context=attribute,datatype=number name=http.status_code,context=attribute,datatype=string]."
     assert warning["warnings"] == [
@@ -328,8 +328,8 @@ def test_deduped_warnings_for_multiple_queries(
         "Key `service.name` is ambiguous, found 2 different combinations of "
         "field context / data type: [name=service.name,context=resource,datatype=string "
         "name=service.name,context=attribute,datatype=string]. Using `resource` context "
-        "by default. To query another context explicitly, use the fully qualified name "
-        "(e.g., 'attribute.service.name' or 'body.service.name')"
+        "by default. To query attributes explicitly, use the fully qualified name "
+        "(e.g., 'attribute.service.name')"
     )
     expected_status_code_warning = "Key `http.status_code` is ambiguous, found 2 different combinations of field context / data type: [name=http.status_code,context=attribute,datatype=number name=http.status_code,context=attribute,datatype=string]."
     assert warning["warnings"] == [

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

@@ -0,0 +1,325 @@
+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 (
+    assert_results_equal,
+    build_builder_query,
+    get_series_values,
+    make_query_request,
+)
+
+MINUTE_MS = 60_000
+
+
+def test_builder_shortening_the_time_range_at_the_end(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    # the cache outlives the run, so a fixed name would serve the previous run's
+    # points back to this one
+    metric_name = f"cache_end_shortened_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a multiple of the 5m step makes the base query span
+    # two whole steps, so both its points are complete
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms_base_query = start_time_ms + 10 * MINUTE_MS
+    end_time_ms_shortened_query = start_time_ms + 7 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
+
+    # the 5m step splits the ten minutes into two points, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. The second changes partway through,
+    # 256 until minute 7 and then 4096, so ending the range at minute 7 has to
+    # reach a different value than ending it at minute 10
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(16, 16, 16, 16, 16, 256, 256, 4096, 4096, 4096)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+    points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["value"], point.get("partial", False)) for point in points]
+    assert returned_points == [(16, False), (4096, False)]
+
+    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened end")
+
+    # the shortened end reaches only minutes 5-6 of the second point, so it comes
+    # back as 256 and partial, where the cached one spans all five minutes
+    for label, response in (("from cache", from_cache), ("uncached", uncached)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["value"], point.get("partial", False)) for point in points]
+        assert returned_points == [(16, False), (256, True)], label
+
+
+def test_builder_shortening_the_time_range_at_the_start(
+    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_start_shortened_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a multiple of the 5m step makes the base query span
+    # two whole steps, so both its points are complete
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms_base_query = int(start_time.timestamp() * 1000)
+    start_time_ms_shortened_query = start_time_ms_base_query + 3 * MINUTE_MS
+    end_time_ms = start_time_ms_base_query + 10 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
+
+    # the 5m step splits the ten minutes into two points, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. Only minute 0 holds 65536, so a first
+    # point reaching it says the whole step was read even though the shortened
+    # range opens at minute 3
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(65536, 16, 16, 16, 16, 4096, 4096, 4096, 4096, 4096)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+    points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["value"], point.get("partial", False)) for point in points]
+    assert returned_points == [(65536, False), (4096, False)]
+
+    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened start")
+
+    # starting inside the first point's step flags that point partial without
+    # clipping its value, which still covers the whole step and so reaches the
+    # 65536 at minute 0
+    for label, response in (("from cache", from_cache), ("uncached", uncached)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["value"], point.get("partial", False)) for point in points]
+        assert returned_points == [(65536, True), (4096, False)], label
+
+
+def test_promql_running_the_same_query_twice(
+    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_repeat_total_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms = start_time_ms + 2 * MINUTE_MS
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"sum(increase({metric_name}[2m]))", "step": 60}}]
+
+    # the counter opens a minute before the query so its first point has something
+    # to increase over, and starts far above its own rise across the range, below
+    # which increase clips its back-extrapolation at the counter's zero point. It
+    # rises by a different amount each minute, so every point is its own number
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(1000, 1010, 1030, 1060, 1100)[minute + 1],
+                temporality="Cumulative",
+                type_="Sum",
+                is_monotonic=True,
+            )
+            for minute in range(-1, 4)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    assert first.status_code == HTTPStatus.OK, first.text
+
+    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    assert second.status_code == HTTPStatus.OK, second.text
+
+    assert_results_equal(first.json(), second.json(), "A", "the same query twice")
+
+    # promql reports a point at the instant the range closes, and the second run,
+    # answered out of what the first one cached, has to keep it
+    for run, response in (("first", first), ("second", second)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"]) for point in points]
+        ## at each timestamp t, promql looks at points in (t-2minutes, t].
+        assert returned_points == [
+            (start_time_ms, 20),  # t = 0, points taken 1000, 1010. hence diff over 1m is 10, extrapolated to 20.
+            (start_time_ms + MINUTE_MS, 40),  # t = 1m, points taken 1010, 1030. hence diff over 1m is 20, extrapolated to 40.
+            (end_time_ms, 60),  # t = 2m, points taken 1030, 1060. hence diff over 1m is 30, extrapolated to 60.
+        ], f"{run} run"
+
+
+def test_promql_shifting_the_time_range(
+    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_shift_gauge_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a whole minute is what makes the first query aligned
+    # to its 1m step, and the unaligned one half a step off it
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    aligned_start_time_ms = int(start_time.timestamp() * 1000)
+    aligned_end_time_ms = aligned_start_time_ms + 3 * MINUTE_MS
+    unaligned_start_time_ms = aligned_start_time_ms + MINUTE_MS // 2
+    unaligned_end_time_ms = aligned_end_time_ms + MINUTE_MS // 2
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric_name}[2m])", "step": 60}}]
+
+    # a sample every 30s, rising by 100 each time. The two queries report 30s
+    # apart, so they land on different samples and share no value between them
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(seconds=30 * half_minute),
+                value=100 * (half_minute + 4),
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for half_minute in range(-3, 8)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False)
+    assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
+
+    # what the cache now holds, and what the unaligned query must not be served
+    points = sorted(get_series_values(aligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["timestamp"], point["value"]) for point in points]
+    ## at each timestamp t, promql takes the highest sample in (t-2minutes, t],
+    ## which is the one at t itself since the gauge only rises.
+    assert returned_points == [
+        (aligned_start_time_ms, 400),  # t = 0
+        (aligned_start_time_ms + MINUTE_MS, 600),  # t = 1m
+        (aligned_start_time_ms + 2 * MINUTE_MS, 800),  # t = 2m
+        (aligned_end_time_ms, 1000),  # t = 3m
+    ]
+
+    unaligned_and_uncached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=True)
+    assert unaligned_and_uncached.status_code == HTTPStatus.OK, unaligned_and_uncached.text
+
+    # promql reports at the range start plus whole steps, so these points sit 30s
+    # off the cached ones. The first run stores them, the second reads them back
+    for run in ("first", "second"):
+        unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False)
+        assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
+        assert_results_equal(unaligned_and_cached.json(), unaligned_and_uncached.json(), "A", f"unaligned query, {run} run")
+
+        points = sorted(get_series_values(unaligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"]) for point in points]
+        ## every point falls on a sample the aligned run never reported, so being
+        ## served the cached run's answer shows up in the values and not only the
+        ## timestamps.
+        assert returned_points == [
+            (unaligned_start_time_ms, 500),  # t = 30s
+            (unaligned_start_time_ms + MINUTE_MS, 700),  # t = 1m30s
+            (unaligned_start_time_ms + 2 * MINUTE_MS, 900),  # t = 2m30s
+            (unaligned_end_time_ms, 1100),  # t = 3m30s
+        ], f"unaligned query, {run} run"
+
+
+def test_builder_refreshing_a_sliding_time_range(
+    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_sliding_{uuid4().hex[:8]}"
+
+    # 90 minutes back so even the twentieth 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")]
+
+    # the 1m step gives one point per seeded minute, and a value no other minute
+    # carries, so a point stitched in from the wrong range reads as the wrong minute
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=1000 + minute,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(80)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # a dashboard left open on a one hour range, re-running a minute later each time
+    for refresh in range(20):
+        refresh_start_ms = start_time_ms + refresh * MINUTE_MS
+        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False)
+        assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+        # each refresh is stitched out of overlapping cached ranges, so this catches
+        # a point served twice, dropped, or carried over from an earlier refresh
+        points = sorted(get_series_values(from_cache.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"], point.get("partial", False)) for point in points]
+        expected_points = [(start_time_ms + minute * MINUTE_MS, 1000 + minute, False) for minute in range(refresh, refresh + 60)]
+        assert returned_points == expected_points, f"refresh {refresh} did not return the minutes it covers"
+
+    last_refresh_start_ms = start_time_ms + 19 * MINUTE_MS
+    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")

tests/integration/tests/queriermetrics/18_heatmap_cache.py

@@ -0,0 +1,546 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+from uuid import uuid4
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import Metrics
+from fixtures.querier import (
+    RequestType,
+    assert_identical_query_response,
+    build_builder_query,
+    build_linear_bucket_options,
+    get_heatmap_buckets,
+    get_heatmap_columns,
+    make_query_request,
+)
+
+MINUTE_MS = 60_000
+
+
+@pytest.mark.parametrize(
+    "first_minute, expected_buckets",
+    [
+        pytest.param(0, [100, 200], id="the_lower_half"),
+        pytest.param(5, [800, 900], id="the_upper_half"),
+    ],
+)
+def test_builder_narrowing_to_half_the_range(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+    first_minute: int,
+    expected_buckets: list[int],
+) -> None:
+    metric_name = f"heatmap_cache_narrowed_{uuid4().hex[:8]}"
+
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms = start_time_ms + 10 * MINUTE_MS
+
+    # 100 wide buckets, and the first five minutes sit seven buckets under the
+    # last five, so the axis over all ten covers a stretch neither half reaches
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=150 if minute < 5 else 850,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = [build_builder_query("A", metric_name, "max", "max", bucket_options=build_linear_bucket_options(1000, 10))]
+
+    # the whole range first, which is what puts its axis in the cache
+    whole_range = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert whole_range.status_code == HTTPStatus.OK, whole_range.text
+    assert get_heatmap_buckets(whole_range.json(), "A") == pytest.approx([100, 200, 300, 400, 500, 600, 700, 800, 900])
+    assert [column["values"] for column in get_heatmap_columns(whole_range.json(), "A")] == [
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+    ]
+
+    half_start_ms = start_time_ms + first_minute * MINUTE_MS
+    half_end_ms = half_start_ms + 5 * MINUTE_MS
+    from_cache = make_query_request(signoz, token, half_start_ms, half_end_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, half_start_ms, half_end_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    for source, response in (("uncached", uncached), ("from cache", from_cache)):
+        assert get_heatmap_buckets(response.json(), "A") == pytest.approx(expected_buckets), source
+        assert [column["values"] for column in get_heatmap_columns(response.json(), "A")] == [
+            [0, 1, 0],
+            [0, 1, 0],
+            [0, 1, 0],
+            [0, 1, 0],
+            [0, 1, 0],
+        ], source
+
+    assert_identical_query_response(from_cache, uncached)
+
+
+def test_builder_narrowing_a_histogram(
+    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"heatmap_cache_histogram_{uuid4().hex[:8]}_bucket"
+
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms = start_time_ms + 10 * MINUTE_MS
+
+    # the count each `le` reports every minute, cumulative across `le` as a
+    # histogram is. For the first five minutes the ten arrivals are all at or
+    # below 1, for the last five they are all between 4 and 8, and the buckets
+    # holding none of them report a count of 0 rather than going unreported
+    le_to_counts = {
+        "1": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
+        "2": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
+        "4": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
+        "8": [10, 10, 10, 10, 10, 10, 10, 10, 10, 10],
+        "+Inf": [10, 10, 10, 10, 10, 10, 10, 10, 10, 10],
+    }
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"le": le},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=count,
+                temporality="Delta",
+                type_="Histogram",
+            )
+            for le, counts in le_to_counts.items()
+            for minute, count in enumerate(counts)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    query = [build_builder_query("A", metric_name, "increase", "p50", temporality="delta", group_by=["le"])]
+
+    # the whole range first, which is what puts its axis in the cache
+    whole_range = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert whole_range.status_code == HTTPStatus.OK, whole_range.text
+    assert get_heatmap_buckets(whole_range.json(), "A") == [1, 2, 4, 8]
+    assert [column["values"] for column in get_heatmap_columns(whole_range.json(), "A")] == [
+        [10, 0, 0, 0, 0],
+        [10, 0, 0, 0, 0],
+        [10, 0, 0, 0, 0],
+        [10, 0, 0, 0, 0],
+        [10, 0, 0, 0, 0],
+        [0, 0, 0, 10, 0],
+        [0, 0, 0, 10, 0],
+        [0, 0, 0, 10, 0],
+        [0, 0, 0, 10, 0],
+        [0, 0, 0, 10, 0],
+    ]
+
+    # even though this shortened time range has no data below 4, all histogram
+    # buckets are still returned back
+    half_start_ms = start_time_ms + 5 * MINUTE_MS
+    from_cache = make_query_request(signoz, token, half_start_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, half_start_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    for source, response in (("uncached", uncached), ("from cache", from_cache)):
+        assert get_heatmap_buckets(response.json(), "A") == [1, 2, 4, 8], source
+        assert [column["values"] for column in get_heatmap_columns(response.json(), "A")] == [
+            [0, 0, 0, 10, 0],
+            [0, 0, 0, 10, 0],
+            [0, 0, 0, 10, 0],
+            [0, 0, 0, 10, 0],
+            [0, 0, 0, 10, 0],
+        ], source
+
+    assert_identical_query_response(from_cache, uncached)
+
+
+def test_builder_shortening_the_time_range_at_the_end(
+    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"heatmap_cache_end_shortened_{uuid4().hex[:8]}"
+
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms_base_query = start_time_ms + 10 * MINUTE_MS
+    end_time_ms_shortened_query = start_time_ms + 7 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300, bucket_options=build_linear_bucket_options(1000, 10))]
+
+    # the 5m step splits the ten minutes into two columns, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. The second changes partway through,
+    # 250 until minute 7 and then 850, which fall six buckets apart, so ending
+    # the range at minute 7 has to reach a different bucket than ending it at
+    # minute 10 and an axis that stops well below it
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(150, 150, 150, 150, 150, 250, 250, 850, 850, 850)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+
+    # 100 wide buckets, and the two maxes are 150 and 850, so the axis runs from
+    # the bottom of (100, 200] to the top of (800, 900]
+    assert get_heatmap_buckets(base_query.json(), "A") == pytest.approx([100, 200, 300, 400, 500, 600, 700, 800, 900])
+    base_columns = get_heatmap_columns(base_query.json(), "A")
+    assert [column["values"] for column in base_columns] == [
+        [0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
+        [0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
+    ]
+    assert [column.get("partial", False) for column in base_columns] == [False, False]
+
+    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    # the shortened end reaches only minutes 5-6 of the second column, whose max
+    # is 250 and which comes back partial. Nothing in this window passes 300, so
+    # the axis stops there rather than carrying the buckets above it
+    for label, response in (("uncached", uncached), ("from cache", from_cache)):
+        assert get_heatmap_buckets(response.json(), "A") == pytest.approx([100, 200, 300]), label
+        columns = get_heatmap_columns(response.json(), "A")
+        assert [column["values"] for column in columns] == [
+            [0, 1, 0, 0],
+            [0, 0, 1, 0],
+        ], label
+        assert [column.get("partial", False) for column in columns] == [False, True], label
+
+    assert_identical_query_response(from_cache, uncached)
+
+
+def test_builder_shortening_the_time_range_at_the_start(
+    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"heatmap_cache_start_shortened_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a multiple of the 5m step makes the base query span
+    # two whole steps, so both its columns are complete
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms_base_query = int(start_time.timestamp() * 1000)
+    start_time_ms_shortened_query = start_time_ms_base_query + 3 * MINUTE_MS
+    end_time_ms = start_time_ms_base_query + 10 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300, bucket_options=build_linear_bucket_options(1000, 10))]
+
+    # the 5m step splits the ten minutes into two columns, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. Only minute 0 reaches 950, so a
+    # first column counted in (900, 1000] says the whole step was read even
+    # though the shortened range opens at minute 3
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(950, 150, 150, 150, 150, 350, 350, 350, 350, 350)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+
+    # 100 wide buckets, and the two maxes are 950 and 350, so the axis runs from
+    # the bottom of (300, 400] to the top of (900, 1000]
+    assert get_heatmap_buckets(base_query.json(), "A") == pytest.approx([300, 400, 500, 600, 700, 800, 900, 1000])
+    base_columns = get_heatmap_columns(base_query.json(), "A")
+    assert [column["values"] for column in base_columns] == [
+        [0, 0, 0, 0, 0, 0, 0, 1, 0],
+        [0, 1, 0, 0, 0, 0, 0, 0, 0],
+    ]
+    assert [column.get("partial", False) for column in base_columns] == [False, False]
+
+    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    # starting inside the first column's step flags that column partial without
+    # clipping its counts, which still cover the whole step and so reach the 950
+    # at minute 0, leaving the axis where the base query drew it
+    for label, response in (("uncached", uncached), ("from cache", from_cache)):
+        assert get_heatmap_buckets(response.json(), "A") == pytest.approx([300, 400, 500, 600, 700, 800, 900, 1000]), label
+        columns = get_heatmap_columns(response.json(), "A")
+        assert [column["values"] for column in columns] == [
+            [0, 0, 0, 0, 0, 0, 0, 1, 0],
+            [0, 1, 0, 0, 0, 0, 0, 0, 0],
+        ], label
+        assert [column.get("partial", False) for column in columns] == [True, False], label
+
+    assert_identical_query_response(from_cache, uncached)
+
+
+def test_builder_refreshing_a_sliding_time_range(
+    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"heatmap_cache_sliding_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+
+    query = [build_builder_query("A", metric_name, "max", "max", bucket_options=build_linear_bucket_options(1000, 10))]
+
+    # the 1m step gives one column per seeded minute, and 100 wide buckets give
+    # every minute a bucket no other minute reaches, so a column stitched in from
+    # the wrong range is counted in the wrong bucket
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=100 * minute + 50,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(7)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # the window slides onto a bucket a minute higher each refresh, so an axis
+    # carried over from an earlier one is off by as many buckets
+    expected_buckets_by_refresh = [
+        [0, 100, 200, 300, 400],
+        [100, 200, 300, 400, 500],
+        [200, 300, 400, 500, 600],
+        [300, 400, 500, 600, 700],
+    ]
+
+    # whichever four minutes a refresh reads, each is in a bucket of its own and
+    # they arrive in order, so the counts run down the diagonal
+    expected_columns = [
+        [0, 1, 0, 0, 0, 0],
+        [0, 0, 1, 0, 0, 0],
+        [0, 0, 0, 1, 0, 0],
+        [0, 0, 0, 0, 1, 0],
+    ]
+
+    # a dashboard left open on a four minute range, re-running a minute later each
+    # time, so every refresh is stitched out of the ranges the ones before it cached
+    for refresh, expected_buckets in enumerate(expected_buckets_by_refresh):
+        refresh_start_ms = start_time_ms + refresh * MINUTE_MS
+        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 4 * MINUTE_MS, query, request_type=RequestType.HEATMAP, no_cache=False)
+        assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+        assert get_heatmap_buckets(from_cache.json(), "A") == pytest.approx(expected_buckets), f"refresh {refresh}"
+
+        # a column served twice, dropped, or carried over from an earlier refresh
+        # breaks the diagonal or the run of timestamps
+        columns = get_heatmap_columns(from_cache.json(), "A")
+        assert [column["timestamp"] for column in columns] == [
+            refresh_start_ms,
+            refresh_start_ms + MINUTE_MS,
+            refresh_start_ms + 2 * MINUTE_MS,
+            refresh_start_ms + 3 * MINUTE_MS,
+        ], f"refresh {refresh}"
+        assert [column["values"] for column in columns] == expected_columns, f"refresh {refresh}"
+
+    uncached = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 4 * MINUTE_MS, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_identical_query_response(from_cache, uncached)
+
+
+def test_promql_running_the_same_query_twice(
+    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"heatmap_cache_repeat_{uuid4().hex[:8]}_bucket"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms = start_time_ms + 2 * MINUTE_MS
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"sum by (le) (increase({metric_name}[2m]))", "step": 60}}]
+
+    # the cumulative count of each `le`, one entry per minute. The counters open
+    # a minute before the query so its first column has something to increase
+    # over, and start far above their own rise across the range, below which
+    # increase clips its back-extrapolation at a counter's zero point
+    le_to_counts = {
+        "1": [1000, 1005, 1010, 1020],
+        "2": [2000, 2010, 2025, 2040],
+        "4": [3000, 3015, 3040, 3070],
+        "+Inf": [4000, 4022, 4050, 4090],
+    }
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"__temporality__": "Cumulative", "service": "api", "le": le},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=count,
+                temporality="Cumulative",
+                type_="Histogram",
+            )
+            for le, counts in le_to_counts.items()
+            for minute, count in enumerate(counts, start=-1)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert first.status_code == HTTPStatus.OK, first.text
+
+    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert second.status_code == HTTPStatus.OK, second.text
+
+    # promql reports a column at the instant the range closes, and the second
+    # run, answered out of what the first one cached, has to keep it
+    for run, response in (("first", first), ("second", second)):
+        assert get_heatmap_buckets(response.json(), "A") == [1, 2, 4], run
+        ## what the query returns per `le` is cumulative across `le`, so each
+        ## count is its own minus the one below it, and `le=+Inf` has no finite
+        ## bound to sit on and lands in the trailing slot. increase over a 2m
+        ## window of minutely samples extrapolates one minute's rise to two.
+        assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(response.json(), "A")] == [
+            (start_time_ms, [10, 10, 10, 14]),  # t = 0, the minute brings 5, 10, 15 and 22 arrivals at or below each `le`
+            (start_time_ms + MINUTE_MS, [10, 20, 20, 6]),  # t = 1m, 5, 15, 25 and 28
+            (end_time_ms, [20, 10, 30, 20]),  # t = 2m, 10, 15, 30 and 40
+        ], f"{run} run"
+
+    assert_identical_query_response(first, second)
+
+
+def test_promql_shifting_the_time_range(
+    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"heatmap_cache_shift_{uuid4().hex[:8]}_bucket"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a whole minute is what makes the first query aligned
+    # to its 1m step, and the unaligned one half a step off it
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    aligned_start_time_ms = int(start_time.timestamp() * 1000)
+    aligned_end_time_ms = aligned_start_time_ms + 3 * MINUTE_MS
+    unaligned_start_time_ms = aligned_start_time_ms + MINUTE_MS // 2
+    unaligned_end_time_ms = aligned_end_time_ms + MINUTE_MS // 2
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"sum by (le) (max_over_time({metric_name}[2m]))", "step": 60}}]
+
+    # a sample every 30s, each `le` counting up by its own fixed amount every
+    # time. The two queries report 30s apart, so they land on different samples
+    # and share no count between them
+    le_to_arrivals_per_sample = {"1": 100, "2": 300, "4": 600, "+Inf": 1000}
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"__temporality__": "Cumulative", "service": "api", "le": le},
+                timestamp=start_time + timedelta(seconds=30 * half_minute),
+                value=arrivals_per_sample * (half_minute + 4),
+                temporality="Cumulative",
+                type_="Histogram",
+            )
+            for le, arrivals_per_sample in le_to_arrivals_per_sample.items()
+            for half_minute in range(-3, 8)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+    assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
+
+    # what the cache now holds, and what the unaligned query must not be served
+    assert get_heatmap_buckets(aligned_and_cached.json(), "A") == [1, 2, 4]
+    ## each column reads the counters at their latest sample at or before its
+    ## timestamp, and a bucket holds its own `le`'s count less the one below it.
+    assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(aligned_and_cached.json(), "A")] == [
+        (aligned_start_time_ms, [400, 800, 1200, 1600]),  # t = 0, the fourth sample
+        (aligned_start_time_ms + MINUTE_MS, [600, 1200, 1800, 2400]),  # t = 1m, the sixth
+        (aligned_start_time_ms + 2 * MINUTE_MS, [800, 1600, 2400, 3200]),  # t = 2m, the eighth
+        (aligned_end_time_ms, [1000, 2000, 3000, 4000]),  # t = 3m, the tenth
+    ]
+
+    ## every column falls on a sample the aligned run never reported, so being
+    ## served the cached run's answer shows up in the counts and not only the
+    ## timestamps.
+    unaligned_columns = [
+        (unaligned_start_time_ms, [500, 1000, 1500, 2000]),  # t = 30s, the fifth sample
+        (unaligned_start_time_ms + MINUTE_MS, [700, 1400, 2100, 2800]),  # t = 1m30s, the seventh
+        (unaligned_start_time_ms + 2 * MINUTE_MS, [900, 1800, 2700, 3600]),  # t = 2m30s, the ninth
+        (unaligned_end_time_ms, [1100, 2200, 3300, 4400]),  # t = 3m30s, the eleventh
+    ]
+
+    unaligned_and_uncached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
+    assert unaligned_and_uncached.status_code == HTTPStatus.OK, unaligned_and_uncached.text
+    assert get_heatmap_buckets(unaligned_and_uncached.json(), "A") == [1, 2, 4], "unaligned query, uncached"
+    assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(unaligned_and_uncached.json(), "A")] == unaligned_columns, "unaligned query, uncached"
+
+    # promql reports at the range start plus whole steps, so these columns sit
+    # 30s off the cached ones. The first run stores them, the second reads them back
+    for run in ("first", "second"):
+        unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
+        assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
+        assert get_heatmap_buckets(unaligned_and_cached.json(), "A") == [1, 2, 4], f"unaligned query, {run} run"
+        assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(unaligned_and_cached.json(), "A")] == unaligned_columns, f"unaligned query, {run} run"
+        assert_identical_query_response(unaligned_and_cached, unaligned_and_uncached)