fix(authdomain): validate SSO role attribute claim against existing roles

When the role mapping uses the IDP role attribute, the claimed role is
assigned only if it exists in the org (resolved case-insensitively, with
legacy ADMIN/EDITOR/VIEWER mapped to their managed names); otherwise the
mapping falls through to group mappings and the default role. This lets
custom roles be assigned via the attribute and avoids failing login on an
unknown claim.

Also normalize legacy role names case-insensitively and fold the
single-use managed-role lookup into NormalizeRoleName.

ee/authz/openfgaauthz/provider.go

@@ -370,7 +370,7 @@ func (provider *provider) Delete(ctx context.Context, orgID valuer.UUID, id valu
 	}
 
 	for _, cb := range provider.onBeforeRoleDelete {
-		if err := cb(ctx, orgID, id); err != nil {
+		if err := cb(ctx, orgID, id, role.Name); err != nil {
 			return err
 		}
 	}

frontend/src/pages/DashboardPageV2/DashboardContainer/store/slices/variableSelectionSlice.ts

@@ -48,8 +48,15 @@ export const createVariableSelectionSlice: StateCreator<
 	},
 });
 
+/**
+ * Stable empty map for dashboards with no stored selections. Returning an inline
+ * `{}` here would hand zustand's useSyncExternalStore a new reference every call,
+ * which it reads as a changed snapshot → infinite re-render loop.
+ */
+const EMPTY_SELECTION_MAP: VariableSelectionMap = {};
+
 /** Selector: the selection map for a dashboard (empty if none). */
 export const selectVariableValues =
 	(dashboardId: string) =>
 	(state: DashboardStore): VariableSelectionMap =>
-		state.variableValues[dashboardId] ?? {};
+		state.variableValues[dashboardId] ?? EMPTY_SELECTION_MAP;

pkg/authz/authz.go

@@ -92,7 +92,7 @@ type AuthZ interface {
 }
 
 // OnBeforeRoleDelete is a callback invoked before a role is deleted.
-type OnBeforeRoleDelete func(context.Context, valuer.UUID, valuer.UUID) error
+type OnBeforeRoleDelete func(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, roleName string) error
 
 type Handler interface {
 	Create(http.ResponseWriter, *http.Request)

pkg/flagger/registry.go

@@ -3,16 +3,17 @@ package flagger
 import "github.com/SigNoz/signoz/pkg/types/featuretypes"
 
 var (
-	FeatureUseSpanMetrics        = featuretypes.MustNewName("use_span_metrics")
-	FeatureKafkaSpanEval         = featuretypes.MustNewName("kafka_span_eval")
-	FeatureHideRootUser          = featuretypes.MustNewName("hide_root_user")
-	FeatureGetMetersFromZeus     = featuretypes.MustNewName("get_meters_from_zeus")
-	FeaturePutMetersInZeus       = featuretypes.MustNewName("put_meters_in_zeus")
-	FeatureUseMeterReporter      = featuretypes.MustNewName("use_meter_reporter")
-	FeatureUseJSONBody           = featuretypes.MustNewName("use_json_body")
-	FeatureUseFineGrainedAuthz   = featuretypes.MustNewName("use_fine_grained_authz")
-	FeatureUseDashboardV2        = featuretypes.MustNewName("use_dashboard_v2")
-	FeatureEnableAIObservability = featuretypes.MustNewName("enable_ai_observability")
+	FeatureUseSpanMetrics         = featuretypes.MustNewName("use_span_metrics")
+	FeatureKafkaSpanEval          = featuretypes.MustNewName("kafka_span_eval")
+	FeatureHideRootUser           = featuretypes.MustNewName("hide_root_user")
+	FeatureGetMetersFromZeus      = featuretypes.MustNewName("get_meters_from_zeus")
+	FeaturePutMetersInZeus        = featuretypes.MustNewName("put_meters_in_zeus")
+	FeatureUseMeterReporter       = featuretypes.MustNewName("use_meter_reporter")
+	FeatureUseJSONBody            = featuretypes.MustNewName("use_json_body")
+	FeatureUseFineGrainedAuthz    = featuretypes.MustNewName("use_fine_grained_authz")
+	FeatureUseDashboardV2         = featuretypes.MustNewName("use_dashboard_v2")
+	FeatureEnableAIObservability  = featuretypes.MustNewName("enable_ai_observability")
+	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 )
 
 func MustNewRegistry() featuretypes.Registry {
@@ -97,6 +98,14 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
+		&featuretypes.Feature{
+			Name:           FeatureEnableMetricsReduction,
+			Kind:           featuretypes.KindBoolean,
+			Stage:          featuretypes.StageExperimental,
+			Description:    "Controls whether metrics cardinality reduction (buffer/reduced tables) is read by the querier",
+			DefaultVariant: featuretypes.MustNewName("disabled"),
+			Variants:       featuretypes.NewBooleanVariants(),
+		},
 	)
 	if err != nil {
 		panic(err)

pkg/modules/authdomain/authdomain.go

@@ -44,3 +44,7 @@ type Handler interface {
 	Update(http.ResponseWriter, *http.Request)
 	Delete(http.ResponseWriter, *http.Request)
 }
+
+type Getter interface {
+	OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, roleName string) error
+}

pkg/modules/authdomain/implauthdomain/getter.go

@@ -0,0 +1,45 @@
+package implauthdomain
+
+import (
+	"context"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/modules/authdomain"
+	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+type getter struct {
+	store authtypes.AuthDomainStore
+}
+
+func NewGetter(store authtypes.AuthDomainStore) authdomain.Getter {
+	return &getter{store: store}
+}
+
+func (getter *getter) OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, roleName string) error {
+	domains, err := getter.store.ListByOrgID(ctx, orgID)
+	if err != nil {
+		return err
+	}
+
+	referencedBy := make([]string, 0)
+	for _, domain := range domains {
+		for _, mappedRole := range domain.AuthDomainConfig().RoleMapping.RoleNames() {
+			if mappedRole == roleName {
+				referencedBy = append(referencedBy, domain.StorableAuthDomain().Name)
+				break
+			}
+		}
+	}
+
+	if len(referencedBy) > 0 {
+		return errors.WithAdditionalf(
+			errors.New(errors.TypeInvalidInput, authtypes.ErrCodeRoleHasAuthDomainMappings, "role is referenced by an SSO role mapping, remove it before deleting"),
+			"referenced by auth domain(s): %s", strings.Join(referencedBy, ", "),
+		)
+	}
+
+	return nil
+}

pkg/modules/authdomain/implauthdomain/module.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/SigNoz/signoz/pkg/authn"
+	"github.com/SigNoz/signoz/pkg/authz"
 	"github.com/SigNoz/signoz/pkg/modules/authdomain"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -12,13 +13,18 @@ import (
 type module struct {
 	store  authtypes.AuthDomainStore
 	authNs map[authtypes.AuthNProvider]authn.AuthN
+	authz  authz.AuthZ
 }
 
-func NewModule(store authtypes.AuthDomainStore, authNs map[authtypes.AuthNProvider]authn.AuthN) authdomain.Module {
-	return &module{store: store, authNs: authNs}
+func NewModule(store authtypes.AuthDomainStore, authNs map[authtypes.AuthNProvider]authn.AuthN, authz authz.AuthZ) authdomain.Module {
+	return &module{store: store, authNs: authNs, authz: authz}
 }
 
 func (module *module) Create(ctx context.Context, domain *authtypes.AuthDomain) error {
+	if err := module.validateRoleMapping(ctx, domain); err != nil {
+		return err
+	}
+
 	return module.store.Create(ctx, domain)
 }
 
@@ -50,6 +56,10 @@ func (module *module) ListByOrgID(ctx context.Context, orgID valuer.UUID) ([]*au
 }
 
 func (module *module) Update(ctx context.Context, domain *authtypes.AuthDomain) error {
+	if err := module.validateRoleMapping(ctx, domain); err != nil {
+		return err
+	}
+
 	return module.store.Update(ctx, domain)
 }
 
@@ -74,3 +84,13 @@ func (module *module) Collect(ctx context.Context, orgID valuer.UUID) (map[strin
 
 	return stats, nil
 }
+
+func (module *module) validateRoleMapping(ctx context.Context, domain *authtypes.AuthDomain) error {
+	roleNames := domain.AuthDomainConfig().RoleMapping.RoleNames()
+	if len(roleNames) == 0 {
+		return nil
+	}
+
+	_, err := module.authz.ListByOrgIDAndNames(ctx, domain.StorableAuthDomain().OrgID, roleNames)
+	return err
+}

pkg/modules/inframonitoring/implinframonitoring/helpers.go

@@ -341,12 +341,12 @@ func alignedMetricWindow(startMs, endMs int64) (
 	}
 
 	tsAdjustedStartMs, _, distributedTSTable, localTSTable := telemetrymetrics.WhichTSTableToUse(
-		samplesAdjustedStartMs, flooredEndMs, nil,
+		samplesAdjustedStartMs, flooredEndMs, false, nil,
 	)
 
 	distributedSamplesTable, localSamplesTable := telemetrymetrics.WhichSamplesTableToUse(
 		samplesAdjustedStartMs, flooredEndMs,
-		metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil,
+		metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil,
 	)
 
 	return samplesAdjustedStartMs, flooredEndMs, tsAdjustedStartMs, distributedTSTable, localTSTable, distributedSamplesTable, localSamplesTable

pkg/modules/metricsexplorer/implmetricsexplorer/module.go

@@ -141,7 +141,7 @@ func (m *module) listMetrics(ctx context.Context, orgID valuer.UUID, params *met
 	sb.Select("DISTINCT metric_name")
 
 	if params.Start != nil && params.End != nil {
-		start, end, distributedTsTable, _ := telemetrymetrics.WhichTSTableToUse(uint64(*params.Start), uint64(*params.End), nil)
+		start, end, distributedTsTable, _ := telemetrymetrics.WhichTSTableToUse(uint64(*params.Start), uint64(*params.End), false, nil)
 		sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
 		sb.Where(sb.Between("unix_milli", start, end))
 	} else {
@@ -527,7 +527,7 @@ func (m *module) InspectMetrics(
 		return nil, err
 	}
 
-	tsStart, _, tsTable, _ := telemetrymetrics.WhichTSTableToUse(start, end, nil)
+	tsStart, _, tsTable, _ := telemetrymetrics.WhichTSTableToUse(start, end, false, nil)
 	tsSb := sqlbuilder.NewSelectBuilder()
 	tsSb.Select("fingerprint", "labels")
 	tsSb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, tsTable))
@@ -971,8 +971,8 @@ func (m *module) fetchMetricsStatsWithSamples(
 		}
 	}
 
-	start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), nil)
-	distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil)
+	start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
+	distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
 	countExp := telemetrymetrics.CountExpressionForSamplesTable(distributedSamplesTable)
 
 	// Timeseries counts per metric
@@ -1100,7 +1100,7 @@ func (m *module) computeTimeseriesTreemap(ctx context.Context, req *metricsexplo
 		}
 	}
 
-	start, end, distributedTsTable, _ := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), nil)
+	start, end, distributedTsTable, _ := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
 
 	totalTSBuilder := sqlbuilder.NewSelectBuilder()
 	totalTSBuilder.Select("uniq(fingerprint) AS total_time_series")
@@ -1176,8 +1176,8 @@ func (m *module) computeSamplesTreemap(ctx context.Context, req *metricsexplorer
 		}
 	}
 
-	start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), nil)
-	distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil)
+	start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
+	distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
 	countExp := telemetrymetrics.CountExpressionForSamplesTable(distributedSamplesTable)
 
 	candidateLimit := req.Limit + 50

pkg/modules/rawdataexport/implrawdataexport/handler.go

@@ -114,6 +114,10 @@ func validateAndApplyDefaultExportLimits(queries []qbtypes.QueryEnvelope) error
 			return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit cannot be more than %d", MaxExportRowCountLimit)
 		}
 		queries[idx].SetLimit(limit)
+
+		if queries[idx].GetOffset() < 0 {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset must be non-negative")
+		}
 	}
 	return nil
 }

pkg/modules/rawdataexport/implrawdataexport/module.go

@@ -70,12 +70,13 @@ func exportRawDataForSingleQuery(querier querier.Querier, ctx context.Context, o
 
 	queries := rangeRequest.CompositeQuery.Queries
 	rowCountLimit := queries[queryIndex].GetLimit()
+	startingOffset := queries[queryIndex].GetOffset()
 	rowCount := 0
 
 	for rowCount < rowCountLimit {
 		chunkSize := min(ChunkSize, rowCountLimit-rowCount)
 		queries[queryIndex].SetLimit(chunkSize)
-		queries[queryIndex].SetOffset(rowCount)
+		queries[queryIndex].SetOffset(startingOffset + rowCount)
 
 		response, err := querier.QueryRange(ctx, orgID, rangeRequest)
 		if err != nil {

pkg/modules/serviceaccount/implserviceaccount/getter.go

@@ -18,7 +18,7 @@ func NewGetter(store serviceaccounttypes.Store) serviceaccount.Getter {
 	return &getter{store: store}
 }
 
-func (getter *getter) OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) error {
+func (getter *getter) OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, _ string) error {
 	serviceAccounts, err := getter.store.GetServiceAccountsByOrgIDAndRoleID(ctx, orgID, roleID)
 	if err != nil {
 		return err

pkg/modules/serviceaccount/serviceaccount.go

@@ -13,7 +13,7 @@ import (
 
 type Getter interface {
 	// OnBeforeRoleDelete checks if any service accounts are assigned to the role and rejects deletion if so.
-	OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) error
+	OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, roleName string) error
 }
 
 type Module interface {

pkg/modules/session/implsession/module.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/authn"
+	"github.com/SigNoz/signoz/pkg/authz"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/modules/authdomain"
@@ -29,9 +30,10 @@ type module struct {
 	authDomain authdomain.Module
 	tokenizer  tokenizer.Tokenizer
 	orgGetter  organization.Getter
+	authz      authz.AuthZ
 }
 
-func NewModule(providerSettings factory.ProviderSettings, authNs map[authtypes.AuthNProvider]authn.AuthN, userSetter user.Setter, userGetter user.Getter, authDomain authdomain.Module, tokenizer tokenizer.Tokenizer, orgGetter organization.Getter) session.Module {
+func NewModule(providerSettings factory.ProviderSettings, authNs map[authtypes.AuthNProvider]authn.AuthN, userSetter user.Setter, userGetter user.Getter, authDomain authdomain.Module, tokenizer tokenizer.Tokenizer, orgGetter organization.Getter, authz authz.AuthZ) session.Module {
 	return &module{
 		settings:   factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/modules/session/implsession"),
 		authNs:     authNs,
@@ -40,6 +42,7 @@ func NewModule(providerSettings factory.ProviderSettings, authNs map[authtypes.A
 		authDomain: authDomain,
 		tokenizer:  tokenizer,
 		orgGetter:  orgGetter,
+		authz:      authz,
 	}
 }
 
@@ -143,15 +146,23 @@ func (module *module) CreateCallbackAuthNSession(ctx context.Context, authNProvi
 	}
 
 	roleMapping := authDomain.AuthDomainConfig().RoleMapping
-	role := roleMapping.NewRoleFromCallbackIdentity(callbackIdentity)
-	signozManagedRole := authtypes.MustGetSigNozManagedRoleFromExistingRole(role)
+
+	roleAttributeExists := false
+	if roleMapping != nil && roleMapping.UseRoleAttribute && callbackIdentity.Role != "" {
+		_, err := module.authz.GetByOrgIDAndName(ctx, callbackIdentity.OrgID, authtypes.NormalizeRoleName(callbackIdentity.Role))
+		if err == nil {
+			roleAttributeExists = true
+		}
+	}
+
+	roleNames := roleMapping.NewRolesFromCallbackIdentity(callbackIdentity, roleAttributeExists)
 
 	newUser, err := types.NewUser(callbackIdentity.Name, callbackIdentity.Email, callbackIdentity.OrgID, types.UserStatusActive)
 	if err != nil {
 		return "", err
 	}
 
-	newUser, err = module.userSetter.GetOrCreateUser(ctx, newUser, user.WithRoleNames([]string{signozManagedRole}))
+	newUser, err = module.userSetter.GetOrCreateUser(ctx, newUser, user.WithRoleNames(roleNames))
 	if err != nil {
 		return "", err
 	}

pkg/modules/user/impluser/getter.go

@@ -239,7 +239,7 @@ func (module *getter) VerifyResetPasswordToken(ctx context.Context, token string
 	return nil
 }
 
-func (module *getter) OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) error {
+func (module *getter) OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, _ string) error {
 	users, err := module.GetUsersByOrgIDAndRoleID(ctx, orgID, roleID)
 	if err != nil {
 		return err

pkg/modules/user/user.go

@@ -96,7 +96,7 @@ type Getter interface {
 	GetUsersByOrgIDAndRoleID(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) ([]*types.User, error)
 
 	// OnBeforeRoleDelete checks if any users are assigned to the role and rejects deletion if so.
-	OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) error
+	OnBeforeRoleDelete(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID, roleName string) error
 
 	// VerifyResetPasswordToken checks if a reset password token exists and is not expired.
 	VerifyResetPasswordToken(ctx context.Context, token string) error

pkg/querier/builder_query.go

@@ -91,13 +91,22 @@ func (q *builderQuery[T]) Fingerprint() string {
 				if a.ComparisonSpaceAggregationParam != nil {
 					spaceAggParamStr = a.ComparisonSpaceAggregationParam.StringValue()
 				}
-				aggParts = append(aggParts, fmt.Sprintf("%s:%s:%s:%s:%s",
+				part := fmt.Sprintf("%s:%s:%s:%s:%s",
 					a.MetricName,
 					a.Temporality.StringValue(),
 					a.TimeAggregation.StringValue(),
 					a.SpaceAggregation.StringValue(),
 					spaceAggParamStr,
-				))
+				)
+				if a.Reduced {
+					oneDay := uint64(24 * time.Hour.Milliseconds())
+					route := "reduced"
+					if q.toMS-q.fromMS < oneDay && q.fromMS >= uint64(time.Now().UnixMilli())-oneDay {
+						route = "buffer"
+					}
+					part += ":" + route
+				}
+				aggParts = append(aggParts, part)
 			}
 		}
 		parts = append(parts, fmt.Sprintf("aggs=[%s]", strings.Join(aggParts, ",")))

pkg/querier/querier.go

@@ -111,7 +111,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 	// We need to set if it is unspecified or adjust it if value is not within recommended range
 	intervalWarnings := q.adjustStepInterval(req.CompositeQuery.Queries, req.Start, req.End)
 
-	missingMetricQueries, metricWarnings, err := q.resolveMetricMetadata(ctx, req.CompositeQuery.Queries, req.Start, req.End)
+	missingMetricQueries, metricWarnings, err := q.resolveMetricMetadata(ctx, orgID, req.CompositeQuery.Queries, req.Start, req.End)
 	if err != nil {
 		return nil, err
 	}
@@ -320,7 +320,7 @@ func (q *querier) populateQBEvent(event *qbtypes.QBEvent, queries []qbtypes.Quer
 //     resolved: never-seen metrics and dormant metrics (seen but no data in
 //     the query window).
 //   - err: Internal when a metadata fetch fails.
-func (q *querier) resolveMetricMetadata(ctx context.Context, queries []qbtypes.QueryEnvelope, start, end uint64) (missingMetricQueries []string, metricWarnings []string, err error) {
+func (q *querier) resolveMetricMetadata(ctx context.Context, orgID valuer.UUID, queries []qbtypes.QueryEnvelope, start, end uint64) (missingMetricQueries []string, metricWarnings []string, err error) {
 	metricNames := make([]string, 0)
 	for idx := range queries {
 		if queries[idx].Type != qbtypes.QueryTypeBuilder {
@@ -341,7 +341,7 @@ func (q *querier) resolveMetricMetadata(ctx context.Context, queries []qbtypes.Q
 		return nil, nil, nil
 	}
 
-	metricTemporality, metricTypes, err := q.metadataStore.FetchTemporalityAndTypeMulti(ctx, start, end, metricNames...)
+	metricTemporality, metricTypes, reducedMetricsSet, err := q.metadataStore.FetchTemporalityAndTypeMulti(ctx, orgID, start, end, metricNames...)
 	if err != nil {
 		q.logger.WarnContext(ctx, "failed to fetch metric temporality", errors.Attr(err), slog.Any("metrics", metricNames))
 		return nil, nil, errors.NewInternalf(errors.CodeInternal, "failed to fetch metrics temporality")
@@ -378,6 +378,9 @@ func (q *querier) resolveMetricMetadata(ctx context.Context, queries []qbtypes.Q
 			if err := spec.Aggregations[i].ValidateForType(); err != nil {
 				return nil, nil, err
 			}
+			if reducedMetricsSet[spec.Aggregations[i].MetricName] {
+				spec.Aggregations[i].Reduced = true
+			}
 			presentAggregations = append(presentAggregations, spec.Aggregations[i])
 		}
 		if len(presentAggregations) == 0 {

pkg/signoz/module.go

@@ -128,6 +128,7 @@ func NewModules(
 	}
 	userSetter := impluser.NewSetter(impluser.NewStore(sqlstore, providerSettings), tokenizer, emailing, providerSettings, orgSetter, authz, analytics, config.User, userRoleStore, userGetter, onDeleteUser)
 	ruleStore := sqlrulestore.NewRuleStore(sqlstore, queryParser, providerSettings)
+	authDomainModule := implauthdomain.NewModule(implauthdomain.NewStore(sqlstore), authNs, authz)
 
 	return Modules{
 		OrgGetter:        orgGetter,
@@ -142,8 +143,8 @@ func NewModules(
 		QuickFilter:      quickfilter,
 		TraceFunnel:      impltracefunnel.NewModule(impltracefunnel.NewStore(sqlstore)),
 		RawDataExport:    implrawdataexport.NewModule(querier),
-		AuthDomain:       implauthdomain.NewModule(implauthdomain.NewStore(sqlstore), authNs),
-		Session:          implsession.NewModule(providerSettings, authNs, userSetter, userGetter, implauthdomain.NewModule(implauthdomain.NewStore(sqlstore), authNs), tokenizer, orgGetter),
+		AuthDomain:       authDomainModule,
+		Session:          implsession.NewModule(providerSettings, authNs, userSetter, userGetter, authDomainModule, tokenizer, orgGetter, authz),
 		SpanPercentile:   implspanpercentile.NewModule(querier, providerSettings),
 		Services:         implservices.NewModule(querier, telemetryStore),
 		MetricsExplorer:  implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, ruleStore, dashboard, providerSettings, config.MetricsExplorer),

pkg/signoz/provider.go

@@ -215,6 +215,7 @@ func NewSQLMigrationProviderFactories(
 		sqlmigration.NewRecreateUserDashboardPreferenceFactory(sqlstore, sqlschema),
 		sqlmigration.NewMigrateRecurrenceBoundsFactory(sqlstore),
 		sqlmigration.NewAddDashboardViewFactory(sqlstore, sqlschema),
+		sqlmigration.NewMigrateSSORoleMappingNamesFactory(sqlstore),
 	)
 }
 

pkg/signoz/signoz.go

@@ -24,6 +24,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/instrumentation"
 	"github.com/SigNoz/signoz/pkg/licensing"
 	"github.com/SigNoz/signoz/pkg/meterreporter"
+	"github.com/SigNoz/signoz/pkg/modules/authdomain/implauthdomain"
 	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
@@ -349,10 +350,13 @@ func New(
 	// Initialize service account getter
 	serviceAccountGetter := implserviceaccount.NewGetter(implserviceaccount.NewStore(sqlstore))
 
+	authDomainGetter := implauthdomain.NewGetter(implauthdomain.NewStore(sqlstore))
+
 	// Build pre-delete callbacks from modules
 	onBeforeRoleDelete := []authz.OnBeforeRoleDelete{
 		userGetter.OnBeforeRoleDelete,
 		serviceAccountGetter.OnBeforeRoleDelete,
+		authDomainGetter.OnBeforeRoleDelete,
 	}
 
 	// Initialize authz

pkg/sqlmigration/096_migrate_sso_role_mapping_names.go

@@ -0,0 +1,127 @@
+package sqlmigration
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/migrate"
+)
+
+type migrateSSORoleMappingNames struct {
+	sqlstore sqlstore.SQLStore
+	logger   *slog.Logger
+}
+
+type authDomainRow struct {
+	bun.BaseModel `bun:"table:auth_domain"`
+
+	ID   string `bun:"id"`
+	Data string `bun:"data"`
+}
+
+var legacyRoleToManagedRoleName = map[string]string{
+	"ADMIN":  "signoz-admin",
+	"EDITOR": "signoz-editor",
+	"VIEWER": "signoz-viewer",
+}
+
+type ssoRoleMapping struct {
+	DefaultRole      string            `json:"defaultRole"`
+	GroupMappings    map[string]string `json:"groupMappings"`
+	UseRoleAttribute bool              `json:"useRoleAttribute"`
+}
+
+func NewMigrateSSORoleMappingNamesFactory(sqlstore sqlstore.SQLStore) factory.ProviderFactory[SQLMigration, Config] {
+	return factory.NewProviderFactory(
+		factory.MustNewName("migrate_sso_role_mapping_names"),
+		func(ctx context.Context, ps factory.ProviderSettings, c Config) (SQLMigration, error) {
+			return &migrateSSORoleMappingNames{sqlstore: sqlstore, logger: ps.Logger}, nil
+		},
+	)
+}
+
+func (migration *migrateSSORoleMappingNames) Register(migrations *migrate.Migrations) error {
+	return migrations.Register(migration.Up, migration.Down)
+}
+
+func (migration *migrateSSORoleMappingNames) Up(ctx context.Context, db *bun.DB) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	rows := make([]*authDomainRow, 0)
+	if err := tx.NewSelect().Model(&rows).Scan(ctx); err != nil {
+		return err
+	}
+
+	for _, row := range rows {
+		config := make(map[string]json.RawMessage)
+		if err := json.Unmarshal([]byte(row.Data), &config); err != nil {
+			migration.logger.WarnContext(ctx, "skipping auth domain with unreadable data", slog.String("auth_domain_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		roleMappingRaw, ok := config["roleMapping"]
+		if !ok || string(roleMappingRaw) == "null" {
+			continue
+		}
+
+		var roleMapping ssoRoleMapping
+		if err := json.Unmarshal(roleMappingRaw, &roleMapping); err != nil {
+			migration.logger.WarnContext(ctx, "skipping auth domain with unreadable role mapping", slog.String("auth_domain_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		changed := false
+		if managed, ok := legacyRoleToManagedRoleName[strings.ToUpper(roleMapping.DefaultRole)]; ok {
+			roleMapping.DefaultRole = managed
+			changed = true
+		}
+		for group, role := range roleMapping.GroupMappings {
+			if managed, ok := legacyRoleToManagedRoleName[strings.ToUpper(role)]; ok {
+				roleMapping.GroupMappings[group] = managed
+				changed = true
+			}
+		}
+
+		if !changed {
+			continue
+		}
+
+		newRoleMapping, err := json.Marshal(roleMapping)
+		if err != nil {
+			return err
+		}
+		config["roleMapping"] = newRoleMapping
+
+		newData, err := json.Marshal(config)
+		if err != nil {
+			return err
+		}
+
+		if _, err := tx.NewUpdate().
+			Model((*authDomainRow)(nil)).
+			Set("data = ?", string(newData)).
+			Where("id = ?", row.ID).
+			Exec(ctx); err != nil {
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (migration *migrateSSORoleMappingNames) Down(context.Context, *bun.DB) error {
+	return nil
+}

pkg/telemetrymetadata/metadata.go

@@ -2136,12 +2136,12 @@ func (t *telemetryMetaStore) GetAllValues(ctx context.Context, fieldValueSelecto
 	return values, complete, nil
 }
 
-func (t *telemetryMetaStore) FetchTemporality(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error) {
+func (t *telemetryMetaStore) FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error) {
 	if metricName == "" {
 		return metrictypes.Unknown, errors.Newf(errors.TypeInternal, errors.CodeInternal, "metric name cannot be empty")
 	}
 
-	temporalityMap, err := t.FetchTemporalityMulti(ctx, queryTimeRangeStartTs, queryTimeRangeEndTs, metricName)
+	temporalityMap, err := t.FetchTemporalityMulti(ctx, orgID, queryTimeRangeStartTs, queryTimeRangeEndTs, metricName)
 	if err != nil {
 		return metrictypes.Unknown, err
 	}
@@ -2154,25 +2154,27 @@ func (t *telemetryMetaStore) FetchTemporality(ctx context.Context, queryTimeRang
 	return temporality, nil
 }
 
-func (t *telemetryMetaStore) FetchTemporalityMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error) {
-	temporalities, _, err := t.FetchTemporalityAndTypeMulti(ctx, queryTimeRangeStartTs, queryTimeRangeEndTs, metricNames...)
+func (t *telemetryMetaStore) FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error) {
+	temporalities, _, _, err := t.FetchTemporalityAndTypeMulti(ctx, orgID, queryTimeRangeStartTs, queryTimeRangeEndTs, metricNames...)
 	return temporalities, err
 }
 
-func (t *telemetryMetaStore) FetchTemporalityAndTypeMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {
+func (t *telemetryMetaStore) FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
 	if len(metricNames) == 0 {
-		return make(map[string]metrictypes.Temporality), make(map[string]metrictypes.Type), nil
+		return make(map[string]metrictypes.Temporality), make(map[string]metrictypes.Type), make(map[string]bool), nil
 	}
 
+	reductionEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID))
+
 	temporalities := make(map[string]metrictypes.Temporality)
 	types := make(map[string]metrictypes.Type)
-	metricsTemporality, metricTypes, err := t.fetchMetricsTemporalityAndType(ctx, queryTimeRangeStartTs, queryTimeRangeEndTs, metricNames...)
+	metricsTemporality, metricTypes, reduced, err := t.fetchMetricsTemporalityAndType(ctx, queryTimeRangeStartTs, queryTimeRangeEndTs, reductionEnabled, metricNames...)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 	meterMetricsTemporality, meterMetricsTypes, err := t.fetchMeterSourceMetricsTemporalityAndType(ctx, metricNames...)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
 	// For metrics not found in the database, set to Unknown
@@ -2197,10 +2199,10 @@ func (t *telemetryMetaStore) FetchTemporalityAndTypeMulti(ctx context.Context, q
 		}
 	}
 
-	return temporalities, types, nil
+	return temporalities, types, reduced, nil
 }
 
-func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, error) {
+func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, reductionEnabled bool, metricNames ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
 		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalMetrics.StringValue(),
 		instrumentationtypes.CodeNamespace:    "metadata",
@@ -2208,48 +2210,58 @@ func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context,
 	})
 	temporalities := make(map[string][]metrictypes.Temporality)
 	types := make(map[string]metrictypes.Type)
+	reduced := make(map[string]bool)
 
-	adjustedStartTs, adjustedEndTs, tsTableName, _ := telemetrymetrics.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, nil)
+	adjustedStartTs, adjustedEndTs, tsTableName, _ := telemetrymetrics.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, false, nil)
 
-	// Build query to fetch temporality for all metrics
-	// We use attr_string_value where attr_name = '__temporality__'
-	// Note: The columns are mixed in the current data - temporality column contains metric_name
-	// and metric_name column contains temporality value, so we use the correct mapping
-	sb := sqlbuilder.Select(
-		"metric_name",
-		"temporality",
-		"any(type) AS type",
-		"any(is_monotonic) as is_monotonic",
-	).
-		From(t.metricsDBName + "." + tsTableName)
+	cols := []string{"metric_name", "temporality", "any(type) AS type", "any(is_monotonic) as is_monotonic"}
 
-	// Filter by metric names (in the temporality column due to data mix-up)
+	// When reduction is enabled, fold the reduced-catalog presence check into the
+	// same query so a metric's reduced status comes back in one round trip.
+	var reducedArgs []any
+	if reductionEnabled {
+		rs := sqlbuilder.NewSelectBuilder()
+		rs.Select("metric_name")
+		rs.From(t.metricsDBName + "." + telemetrymetrics.TimeseriesV4ReducedTableName)
+		rs.Where(rs.In("metric_name", metricNames), rs.GTE("unix_milli", adjustedStartTs), rs.LT("unix_milli", adjustedEndTs))
+		rs.GroupBy("metric_name")
+		rsQuery, rsArgs := rs.BuildWithFlavor(sqlbuilder.ClickHouse)
+		cols = append(cols, fmt.Sprintf("metric_name GLOBAL IN (%s) AS reduced", rsQuery))
+		reducedArgs = rsArgs
+	}
+
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select(cols...)
+	sb.From(t.metricsDBName + "." + tsTableName)
 	sb.Where(
 		sb.In("metric_name", metricNames),
 		sb.GTE("unix_milli", adjustedStartTs),
 		sb.LT("unix_milli", adjustedEndTs),
 	)
-
 	sb.GroupBy("metric_name", "temporality")
 
-	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, reducedArgs...)
 
 	t.logger.DebugContext(ctx, "fetching metric temporality", slog.String("query", query), slog.Any("args", args))
 
 	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
-		return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric temporality")
+		return nil, nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric temporality")
 	}
 	defer rows.Close()
 
-	// Process results
 	for rows.Next() {
 		var metricName string
 		var temporality metrictypes.Temporality
 		var metricType metrictypes.Type
 		var isMonotonic bool
-		if err := rows.Scan(&metricName, &temporality, &metricType, &isMonotonic); err != nil {
-			return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
+		var isReduced uint8
+		dest := []any{&metricName, &temporality, &metricType, &isMonotonic}
+		if reductionEnabled {
+			dest = append(dest, &isReduced)
+		}
+		if err := rows.Scan(dest...); err != nil {
+			return nil, nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
 		}
 		if temporality != metrictypes.Unknown {
 			temporalities[metricName] = append(temporalities[metricName], temporality)
@@ -2258,12 +2270,15 @@ func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context,
 			metricType = metrictypes.GaugeType
 		}
 		types[metricName] = metricType
+		if isReduced != 0 {
+			reduced[metricName] = true
+		}
 	}
 	if err := rows.Err(); err != nil {
-		return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
+		return nil, nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
 	}
 
-	return temporalities, types, nil
+	return temporalities, types, reduced, nil
 }
 
 func (t *telemetryMetaStore) fetchMeterSourceMetricsTemporalityAndType(ctx context.Context, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {

pkg/telemetrymetrics/reduced_test.go

@@ -0,0 +1,157 @@
+package telemetrymetrics
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	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/stretchr/testify/require"
+)
+
+func reducedQuery(metric string, ty metrictypes.Type, temp metrictypes.Temporality, ta metrictypes.TimeAggregation, sa metrictypes.SpaceAggregation) qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation] {
+	return qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+		Signal:       telemetrytypes.SignalMetrics,
+		StepInterval: qbtypes.Step{Duration: 5 * time.Minute},
+		Aggregations: []qbtypes.MetricAggregation{{
+			MetricName:       metric,
+			Type:             ty,
+			Temporality:      temp,
+			TimeAggregation:  ta,
+			SpaceAggregation: sa,
+			Reduced:          true,
+		}},
+	}
+}
+
+func TestReducedStatementBuilder(t *testing.T) {
+	cases := []struct {
+		name     string
+		query    qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]
+		expected qbtypes.Statement
+	}{
+		{
+			name:  "gauge_sum_latest",
+			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationLatest, metrictypes.SpaceAggregationSum),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, anyLast(last) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, argMax(value, unix_milli) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "gauge_avg_avg",
+			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "gauge_min_min",
+			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMin, metrictypes.SpaceAggregationMin),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(min) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`min`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "gauge_max_max",
+			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMax, metrictypes.SpaceAggregationMax),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`max`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "counter_sum_rate",
+			query: reducedQuery("test.metric.sum", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationSum),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) / 300 AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric.sum", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric.sum", uint64(1746999600000), uint64(1747172760000), 0, "test.metric.sum", uint64(1746999600000), uint64(1747172760000), "test.metric.sum", uint64(1746999600000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "counter_avg_increase",
+			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationIncrease, metrictypes.SpaceAggregationAvg),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0, "test.metric", uint64(1746999600000), uint64(1747172760000), "test.metric", uint64(1746999600000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "counter_min_omitted",
+			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationMin),
+			expected: qbtypes.Statement{
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
+			},
+		},
+		{
+			name:  "counter_max_omitted",
+			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationMax),
+			expected: qbtypes.Statement{
+				Query: "WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric", uint64(1746999600000), uint64(1747172760000), 0},
+			},
+		},
+		{
+			name:  "histogram_p99",
+			query: reducedQuery("test.metric.bucket", metrictypes.HistogramType, metrictypes.Cumulative, metrictypes.TimeAggregationUnspecified, metrictypes.SpaceAggregationPercentile99),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, `le`, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `le` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, `le`, sum(value) / 300 AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts, `le`), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric.bucket", uint64(1746921600000), uint64(1747172760000), "cumulative", false, "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), 0, "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), "test.metric.bucket", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+		{
+			name:  "summary_avg",
+			query: reducedQuery("test.metric", metrictypes.SummaryType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
+			expected: qbtypes.Statement{
+				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				Args:  []any{"test.metric", uint64(1746921600000), uint64(1747172760000), "unspecified", false, "test.metric", uint64(1746999900000), uint64(1747172760000), 0, "test.metric", uint64(1746999900000), uint64(1747172760000), "test.metric", uint64(1746999900000), uint64(1747172760000), false},
+			},
+		},
+	}
+
+	fm := NewFieldMapper()
+	cb := NewConditionBuilder(fm)
+	fl, err := flagger.New(context.Background(), instrumentationtest.New().ToProviderSettings(), flagger.Config{}, flagger.MustNewRegistry())
+	require.NoError(t, err)
+	sb := NewMetricQueryStatementBuilder(instrumentationtest.New().ToProviderSettings(), telemetrytypestest.NewMockMetadataStore(), fm, cb, fl)
+
+	const start, end = uint64(1747000000000), uint64(1747172800000)
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got, err := sb.Build(context.Background(), start, end, qbtypes.RequestTypeTimeSeries, c.query, nil)
+			require.NoError(t, err)
+			require.Equal(t, c.expected.Query, got.Query)
+			require.Equal(t, c.expected.Args, got.Args)
+		})
+	}
+
+	t.Run("buffer_recent_window", func(t *testing.T) {
+		now := time.Now().UnixMilli()
+		q := reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationLatest, metrictypes.SpaceAggregationSum)
+		got, err := sb.Build(context.Background(), uint64(now-2*time.Hour.Milliseconds()), uint64(now), qbtypes.RequestTypeTimeSeries, q, nil)
+		require.NoError(t, err)
+		require.Contains(t, got.Query, "signoz_metrics.distributed_samples_v4_buffer")
+		require.Contains(t, got.Query, "signoz_metrics.time_series_v4_buffer")
+		require.Contains(t, got.Query, "is_reduced")
+		require.NotContains(t, got.Query, "UNION ALL")
+	})
+
+	t.Run("not_reduced", func(t *testing.T) {
+		q := reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationLatest, metrictypes.SpaceAggregationSum)
+		q.Aggregations[0].Reduced = false
+		got, err := sb.Build(context.Background(), start, end, qbtypes.RequestTypeTimeSeries, q, nil)
+		require.NoError(t, err)
+		require.NotContains(t, got.Query, "UNION ALL")
+		require.NotContains(t, got.Query, "reduced")
+		require.NotContains(t, got.Query, "buffer")
+	})
+}

pkg/telemetrymetrics/statement_builder.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/flagger"
@@ -177,19 +178,30 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 		query.Aggregations[0].SpaceAggregation = metrictypes.SpaceAggregationSum
 	}
 
+	agg := query.Aggregations[0]
+
+	// A reduced metric reads the raw buffer for recent short windows, and
+	// samples_v4/agg (unioned with the reduced tables) otherwise. The buffer is
+	// shaped exactly like samples_v4 / time_series_v4, so once the table names are
+	// chosen the rest of the pipeline is unchanged.
+	useBuffer := agg.Reduced &&
+		end-start < oneDayInMilliseconds &&
+		start >= uint64(time.Now().UnixMilli())-oneDayInMilliseconds
+
+	samplesTable, _ := WhichSamplesTableToUse(start, end, agg.Type, agg.TimeAggregation, useBuffer, agg.TableHints)
+	tsStart, tsEnd, _, tsTable := WhichTSTableToUse(start, end, useBuffer, agg.TableHints)
+
 	var timeSeriesCTE string
 	var timeSeriesCTEArgs []any
 	var err error
 
-	// time_series_cte
-	// this is applicable for all the queries
-	if timeSeriesCTE, timeSeriesCTEArgs, err = b.buildTimeSeriesCTE(ctx, start, end, query, keys, variables); err != nil {
+	if timeSeriesCTE, timeSeriesCTEArgs, err = b.buildTimeSeriesCTE(ctx, tsStart, tsEnd, query, keys, variables, tsTable); err != nil {
 		return nil, err
 	}
 
 	if qbtypes.CanShortCircuitDelta(query.Aggregations[0]) {
 		// spatial_aggregation_cte directly for certain delta queries
-		if frag, args, err := b.buildTemporalAggDeltaFastPath(start, end, query, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
+		if frag, args, err := b.buildTemporalAggDeltaFastPath(start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
 			return nil, err
 		} else if frag != "" {
 			cteFragments = append(cteFragments, frag)
@@ -197,7 +209,7 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 		}
 	} else {
 		// temporal_aggregation_cte
-		if frag, args, err := b.buildTemporalAggregationCTE(ctx, start, end, query, keys, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
+		if frag, args, err := b.buildTemporalAggregationCTE(ctx, start, end, query, keys, samplesTable, timeSeriesCTE, timeSeriesCTEArgs); err != nil {
 			return nil, err
 		} else if frag != "" {
 			cteFragments = append(cteFragments, frag)
@@ -211,18 +223,188 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
 		}
 	}
 
+	var reducedFragments []string
+	var reducedArgs [][]any
+	if agg.Reduced && !useBuffer {
+		var tsCTE string
+		var tsArgs []any
+		if tsCTE, tsArgs, err = b.buildReducedTimeSeriesCTE(ctx, start, end, query, keys, variables); err != nil {
+			return nil, err
+		}
+		if temporalFrag, temporalArgs, ok := b.buildReducedTemporalAggregationCTE(start, end, query, tsCTE, tsArgs); ok {
+			spatialFrag, spatialArgs := b.buildReducedSpatialAggregationCTE(query)
+			reducedFragments = []string{temporalFrag, spatialFrag}
+			reducedArgs = [][]any{temporalArgs, spatialArgs}
+		}
+	}
+
 	// reset the query to the original state
 	query.Aggregations[0].SpaceAggregation = origSpaceAgg
 	query.Aggregations[0].TimeAggregation = origTimeAgg
 	query.GroupBy = origGroupBy
 
-	// final SELECT
-	return b.BuildFinalSelect(cteFragments, cteArgs, query)
+	mainStmt, err := b.BuildFinalSelect(cteFragments, cteArgs, query)
+	if err != nil {
+		return nil, err
+	}
+	if reducedFragments == nil {
+		return mainStmt, nil
+	}
+	reducedStmt, err := b.BuildFinalSelect(reducedFragments, reducedArgs, query)
+	if err != nil {
+		return nil, err
+	}
+	return unionStatements(mainStmt, reducedStmt, query)
+}
+
+func unionStatements(main, reduced *qbtypes.Statement, query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]) (*qbtypes.Statement, error) {
+	orderBy := "ts"
+	for _, g := range query.GroupBy {
+		orderBy = fmt.Sprintf("`%s`, ", g.Name) + orderBy
+	}
+	q := fmt.Sprintf("SELECT * FROM (%s) UNION ALL SELECT * FROM (%s) ORDER BY %s", main.Query, reduced.Query, orderBy)
+	args := append(append([]any{}, main.Args...), reduced.Args...)
+	warnings := append(append([]string{}, main.Warnings...), reduced.Warnings...)
+	return &qbtypes.Statement{Query: q, Args: args, Warnings: warnings}, nil
+}
+
+func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
+	ctx context.Context,
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	keys map[string][]*telemetrytypes.TelemetryFieldKey,
+	variables map[string]qbtypes.VariableItem,
+) (string, []any, error) {
+	sb := sqlbuilder.NewSelectBuilder()
+
+	var preparedWhereClause querybuilder.PreparedWhereClause
+	var err error
+	if query.Filter != nil && query.Filter.Expression != "" {
+		preparedWhereClause, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
+			Context:          ctx,
+			Logger:           b.logger,
+			FieldMapper:      b.fm,
+			ConditionBuilder: b.cb,
+			FieldKeys:        keys,
+			FullTextColumn:   &telemetrytypes.TelemetryFieldKey{Name: "labels"},
+			Variables:        variables,
+			StartNs:          start,
+			EndNs:            end,
+		})
+		if err != nil {
+			return "", nil, err
+		}
+	}
+
+	sb.From(fmt.Sprintf("%s.%s", DBName, TimeseriesV4ReducedTableName))
+	sb.Select("fingerprint")
+	for _, g := range query.GroupBy {
+		col, err := b.fm.ColumnExpressionFor(ctx, start, end, &g.TelemetryFieldKey, keys)
+		if err != nil {
+			return "", nil, err
+		}
+		sb.SelectMore(col)
+	}
+	sb.Where(
+		sb.In("metric_name", query.Aggregations[0].MetricName),
+		sb.GTE("unix_milli", start),
+		sb.LTE("unix_milli", end),
+		sb.EQ("__normalized", false),
+	)
+
+	if !preparedWhereClause.IsEmpty() {
+		sb.AddWhereClause(preparedWhereClause.WhereClause)
+	}
+	sb.GroupBy("fingerprint")
+	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return fmt.Sprintf("(%s) AS filtered_time_series", q), args, nil
+}
+
+func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	timeSeriesCTE string,
+	timeSeriesCTEArgs []any,
+) (string, []any, bool) {
+	agg := query.Aggregations[0]
+	stepSec := int64(query.StepInterval.Seconds())
+
+	value, weight, ok := ReducedValueColumn(agg.Type, agg.SpaceAggregation)
+	if !ok {
+		return "", nil, false
+	}
+
+	// dedup recomputed buckets: latest computed_at wins per (series, 60s bucket)
+	dedup := sqlbuilder.NewSelectBuilder()
+	dedup.Select("reduced_fingerprint AS fingerprint", "unix_milli")
+	dedup.SelectMore(fmt.Sprintf("argMax(%s, computed_at) AS value", value))
+	if weight != "" {
+		dedup.SelectMore(fmt.Sprintf("argMax(%s, computed_at) AS weight", weight))
+	}
+	dedup.From(fmt.Sprintf("%s.%s", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	dedup.Where(
+		dedup.In("metric_name", agg.MetricName),
+		dedup.GTE("unix_milli", start),
+		dedup.LT("unix_milli", end),
+	)
+	dedup.GroupBy("reduced_fingerprint", "unix_milli")
+	dedupQuery, dedupArgs := dedup.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("fingerprint")
+	sb.SelectMore(fmt.Sprintf("toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts", stepSec))
+	for _, g := range query.GroupBy {
+		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
+	}
+	sb.SelectMore(fmt.Sprintf("%s AS per_series_value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec)))
+	if weight != "" {
+		// count_series is a series count, not additive over time, so the avg
+		// denominator is reduced with avg
+		sb.SelectMore("avg(weight) AS per_series_weight")
+	}
+	sb.From(fmt.Sprintf("(%s) AS points", dedupQuery))
+	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
+	sb.GroupBy("fingerprint", "ts")
+	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+
+	initArgs := append(append([]any{}, dedupArgs...), timeSeriesCTEArgs...)
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, initArgs...)
+	return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, true
+}
+
+func (b *MetricQueryStatementBuilder) buildReducedSpatialAggregationCTE(
+	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+) (string, []any) {
+	spatial := "sum(per_series_value)"
+	switch query.Aggregations[0].SpaceAggregation {
+	case metrictypes.SpaceAggregationAvg:
+		spatial = "sum(per_series_value) / sum(per_series_weight)"
+	case metrictypes.SpaceAggregationMin:
+		spatial = "min(per_series_value)"
+	case metrictypes.SpaceAggregationMax:
+		spatial = "max(per_series_value)"
+	}
+
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("ts")
+	for _, g := range query.GroupBy {
+		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
+	}
+	sb.SelectMore(spatial + " AS value")
+	sb.From("__temporal_aggregation_cte")
+	sb.GroupBy("ts")
+	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args
 }
 
 func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
 ) (string, []any, error) {
@@ -239,8 +421,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
 	}
 
 	aggCol, err := AggregationColumnForSamplesTable(
-		start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality,
-		query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints,
+		samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
 	)
 	if err != nil {
 		return "", nil, err
@@ -257,8 +438,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
 
 	sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))
 
-	tbl, _ := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
-	sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
+	sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
 	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
 	sb.Where(
 		sb.In("metric_name", query.Aggregations[0].MetricName),
@@ -278,6 +458,7 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
 	keys map[string][]*telemetrytypes.TelemetryFieldKey,
 	variables map[string]qbtypes.VariableItem,
+	tsTable string,
 ) (string, []any, error) {
 	sb := sqlbuilder.NewSelectBuilder()
 
@@ -301,8 +482,7 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
 		}
 	}
 
-	start, end, _, tbl := WhichTSTableToUse(start, end, query.Aggregations[0].TableHints)
-	sb.From(fmt.Sprintf("%s.%s", DBName, tbl))
+	sb.From(fmt.Sprintf("%s.%s", DBName, tsTable))
 
 	sb.Select("fingerprint")
 	for _, g := range query.GroupBy {
@@ -328,6 +508,12 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
 		sb.EQ("__normalized", false),
 	)
 
+	// the buffer holds both raw rows and the reduced catalog rows; the raw read
+	// only wants the original series
+	if tsTable == TimeseriesV4BufferLocalTableName {
+		sb.Where(sb.EQ("is_reduced", false))
+	}
+
 	if !preparedWhereClause.IsEmpty() {
 		sb.AddWhereClause(preparedWhereClause.WhereClause)
 	}
@@ -344,21 +530,23 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggregationCTE(
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
 	_ map[string][]*telemetrytypes.TelemetryFieldKey,
+	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
 ) (string, []any, error) {
 	if query.Aggregations[0].Temporality == metrictypes.Delta {
-		return b.buildTemporalAggDelta(ctx, start, end, query, timeSeriesCTE, timeSeriesCTEArgs)
+		return b.buildTemporalAggDelta(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
 	} else if query.Aggregations[0].Temporality != metrictypes.Multiple {
-		return b.buildTemporalAggCumulativeOrUnspecified(ctx, start, end, query, timeSeriesCTE, timeSeriesCTEArgs)
+		return b.buildTemporalAggCumulativeOrUnspecified(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
 	}
-	return b.buildTemporalAggForMultipleTemporalities(ctx, start, end, query, timeSeriesCTE, timeSeriesCTEArgs)
+	return b.buildTemporalAggForMultipleTemporalities(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
 }
 
 func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
 	_ context.Context,
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
 ) (string, []any, error) {
@@ -375,7 +563,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
 		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
 
-	aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
+	aggCol, err := AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
 	if err != nil {
 		return "", nil, err
 	}
@@ -386,8 +574,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
 
 	sb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
 
-	tbl, _ := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
-	sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
+	sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
 	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
 	sb.Where(
 		sb.In("metric_name", query.Aggregations[0].MetricName),
@@ -406,6 +593,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
 	_ context.Context,
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
 ) (string, []any, error) {
@@ -421,14 +609,13 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
 		baseSb.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
 
-	aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
+	aggCol, err := AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
 	if err != nil {
 		return "", nil, err
 	}
 	baseSb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
 
-	tbl, _ := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
-	baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
+	baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
 	baseSb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
 	baseSb.Where(
 		baseSb.In("metric_name", query.Aggregations[0].MetricName),
@@ -472,6 +659,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
 	_ context.Context,
 	start, end uint64,
 	query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
+	samplesTable string,
 	timeSeriesCTE string,
 	timeSeriesCTEArgs []any,
 ) (string, []any, error) {
@@ -486,11 +674,11 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
 		sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
 
-	aggForDeltaTemporality, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, metrictypes.Delta, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
+	aggForDeltaTemporality, err := AggregationColumnForSamplesTable(samplesTable, metrictypes.Delta, query.Aggregations[0].TimeAggregation)
 	if err != nil {
 		return "", nil, err
 	}
-	aggForCumulativeTemporality, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, metrictypes.Cumulative, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
+	aggForCumulativeTemporality, err := AggregationColumnForSamplesTable(samplesTable, metrictypes.Cumulative, query.Aggregations[0].TimeAggregation)
 	if err != nil {
 		return "", nil, err
 	}
@@ -518,8 +706,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
 		sb.SelectMore(expr)
 	}
 
-	tbl, _ := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
-	sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
+	sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
 	sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
 	sb.Where(
 		sb.In("metric_name", query.Aggregations[0].MetricName),

pkg/telemetrymetrics/tables.go

@@ -30,6 +30,17 @@ const (
 	TimeseriesV41weekLocalTableName  = "time_series_v4_1week"
 	AttributesMetadataTableName      = "distributed_metadata"
 	AttributesMetadataLocalTableName = "metadata"
+
+	// The buffer holds raw points for ~24h; the reduced tables hold 60s
+	// aggregates of dropped-label series.
+	SamplesV4BufferTableName          = "distributed_samples_v4_buffer"
+	SamplesV4BufferLocalTableName     = "samples_v4_buffer"
+	TimeseriesV4BufferTableName       = "distributed_time_series_v4_buffer"
+	TimeseriesV4BufferLocalTableName  = "time_series_v4_buffer"
+	SamplesV4ReducedLastTableName     = "distributed_samples_v4_reduced_last_60s"
+	SamplesV4ReducedSumTableName      = "distributed_samples_v4_reduced_sum_60s"
+	TimeseriesV4ReducedTableName      = "distributed_time_series_v4_reduced"
+	TimeseriesV4ReducedLocalTableName = "time_series_v4_reduced"
 )
 
 var (
@@ -49,8 +60,16 @@ var (
 // in that order.
 func WhichTSTableToUse(
 	start, end uint64,
+	useBuffer bool,
 	tableHints *metrictypes.MetricTableHints,
 ) (uint64, uint64, string, string) {
+	// the buffer holds the recent raw window for reduced metrics and has the same
+	// shape as time_series_v4; round the start to the hour like the v4 table.
+	if useBuffer {
+		start = start - (start % (oneHourInMilliseconds))
+		return start, end, TimeseriesV4BufferTableName, TimeseriesV4BufferLocalTableName
+	}
+
 	// if we have a hint for the table, we need to use it
 	// the hint will be used to override the default table selection logic
 	if tableHints != nil {
@@ -149,14 +168,20 @@ func WhichSamplesTableToUse(
 	start, end uint64,
 	metricType metrictypes.Type,
 	timeAggregation metrictypes.TimeAggregation,
+	useBuffer bool,
 	tableHints *metrictypes.MetricTableHints,
 ) (string, string) {
+	// the buffer holds the recent raw window for reduced metrics; same shape as samples_v4
+	if useBuffer {
+		return SamplesV4BufferTableName, SamplesV4BufferLocalTableName
+	}
+
 	// if we have a hint for the table, we need to use it
 	// the hint will be used to override the default table selection logic.
 	// SamplesTableName is the distributed name; derive the local via switch.
 	if tableHints != nil && tableHints.SamplesTableName != "" {
 		switch tableHints.SamplesTableName {
-		case SamplesV4TableName:
+		case SamplesV4TableName, SamplesV4BufferTableName:
 			return SamplesV4TableName, SamplesV4LocalTableName
 		case SamplesV4Agg5mTableName:
 			return SamplesV4Agg5mTableName, SamplesV4Agg5mLocalTableName
@@ -188,13 +213,10 @@ func WhichSamplesTableToUse(
 }
 
 func AggregationColumnForSamplesTable(
-	start, end uint64,
-	metricType metrictypes.Type,
+	tableName string,
 	temporality metrictypes.Temporality,
 	timeAggregation metrictypes.TimeAggregation,
-	tableHints *metrictypes.MetricTableHints,
 ) (string, error) {
-	tableName, _ := WhichSamplesTableToUse(start, end, metricType, timeAggregation, tableHints)
 	var aggregationColumn string
 	switch temporality {
 	case metrictypes.Delta:
@@ -202,7 +224,7 @@ func AggregationColumnForSamplesTable(
 		// although it doesn't make sense to use anyLast, avg, min, max, count on delta metrics,
 		// we are keeping it here to make sure that query will not be invalid
 		switch tableName {
-		case SamplesV4TableName:
+		case SamplesV4TableName, SamplesV4BufferTableName:
 			switch timeAggregation {
 			case metrictypes.TimeAggregationLatest:
 				aggregationColumn = "anyLast(value)"
@@ -244,7 +266,7 @@ func AggregationColumnForSamplesTable(
 		// for cumulative metrics, we only support `RATE`/`INCREASE`. The max value in window is
 		// used to calculate the sum which is then divided by the window size to get the rate
 		switch tableName {
-		case SamplesV4TableName:
+		case SamplesV4TableName, SamplesV4BufferTableName:
 			switch timeAggregation {
 			case metrictypes.TimeAggregationLatest:
 				aggregationColumn = "anyLast(value)"
@@ -284,7 +306,7 @@ func AggregationColumnForSamplesTable(
 		}
 	case metrictypes.Unspecified:
 		switch tableName {
-		case SamplesV4TableName:
+		case SamplesV4TableName, SamplesV4BufferTableName:
 			switch timeAggregation {
 			case metrictypes.TimeAggregationLatest:
 				aggregationColumn = "anyLast(value)"
@@ -332,6 +354,65 @@ func AggregationColumnForSamplesTable(
 	return aggregationColumn, nil
 }
 
+// WhichReducedSamplesTableToUse returns the 60s reduced samples table for a metric
+// type: the last_60s table for gauge-like series, the sum_60s table for counters
+// and histograms.
+func WhichReducedSamplesTableToUse(metricType metrictypes.Type) string {
+	if metricType == metrictypes.SumType || metricType == metrictypes.HistogramType {
+		return SamplesV4ReducedSumTableName
+	}
+	return SamplesV4ReducedLastTableName
+}
+
+// ReducedValueColumn returns the reduced value column (and the avg-denominator
+// weight) for a space aggregation. The reduced columns are pre-aggregated across
+// the original series, so the space aggregation picks the underlying value; the
+// sum table only has `sum`, so min/max across series have no column (ok=false).
+func ReducedValueColumn(metricType metrictypes.Type, space metrictypes.SpaceAggregation) (value, weight string, ok bool) {
+	if metricType == metrictypes.SumType || metricType == metrictypes.HistogramType {
+		switch space {
+		case metrictypes.SpaceAggregationSum:
+			return "`sum`", "", true
+		case metrictypes.SpaceAggregationAvg:
+			return "`sum`", "`count_series`", true
+		}
+		return "", "", false
+	}
+	switch space {
+	case metrictypes.SpaceAggregationSum:
+		return "`sum_last`", "", true
+	case metrictypes.SpaceAggregationAvg:
+		return "`sum_last`", "`count_series`", true
+	case metrictypes.SpaceAggregationMin:
+		return "`min`", "", true
+	case metrictypes.SpaceAggregationMax:
+		return "`max`", "", true
+	}
+	return "", "", false
+}
+
+// ReducedTimeAggregationColumn applies the time aggregation to the reduced `value`
+// column over the step's 60s buckets. latest uses argMax over the bucket timestamp
+// (the buckets have no read order); rate divides the per-step sum by the step.
+func ReducedTimeAggregationColumn(timeAggregation metrictypes.TimeAggregation, stepSec int64) string {
+	switch timeAggregation {
+	case metrictypes.TimeAggregationLatest:
+		return "argMax(value, unix_milli)"
+	case metrictypes.TimeAggregationAvg:
+		return "avg(value)"
+	case metrictypes.TimeAggregationMin:
+		return "min(value)"
+	case metrictypes.TimeAggregationMax:
+		return "max(value)"
+	case metrictypes.TimeAggregationCount:
+		return "count(value)"
+	case metrictypes.TimeAggregationRate:
+		return fmt.Sprintf("sum(value) / %d", stepSec)
+	default: // sum, increase
+		return "sum(value)"
+	}
+}
+
 func AggregationQueryForHistogramCountWithParams(param *metrictypes.ComparisonSpaceAggregationParam) (string, error) {
 	if param == nil {
 		return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "no aggregation param provided for histogram count")

pkg/types/authtypes/mapping.go

@@ -2,10 +2,6 @@ package authtypes
 
 import (
 	"encoding/json"
-	"strings"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/types"
 )
 
 type AttributeMapping struct {
@@ -51,83 +47,95 @@ func (attr *AttributeMapping) UnmarshalJSON(data []byte) error {
 }
 
 type RoleMapping struct {
-	// Default role any new SSO users. Defaults to "VIEWER"
+	// Default role assigned to new SSO users when no group mapping applies.
 	DefaultRole string `json:"defaultRole"`
-	// Map of IDP group names to SigNoz roles. Key is group name, value is SigNoz role
+
+	// Map of IDP group name to SigNoz role name.
 	GroupMappings map[string]string `json:"groupMappings"`
-	// If true, use the role claim directly from IDP instead of group mappings
+
+	// If true, use the role claim directly from IDP instead of group mappings.
 	UseRoleAttribute bool `json:"useRoleAttribute"`
 }
 
-func (typ *RoleMapping) UnmarshalJSON(data []byte) error {
-	type Alias RoleMapping
+func (roleMapping *RoleMapping) UnmarshalJSON(data []byte) error {
+	type alias RoleMapping
 
-	var temp Alias
+	var temp alias
 	if err := json.Unmarshal(data, &temp); err != nil {
 		return err
 	}
 
-	if temp.DefaultRole != "" {
-		if _, err := types.NewRole(strings.ToUpper(temp.DefaultRole)); err != nil {
-			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "invalid default role %s", temp.DefaultRole)
-		}
-	}
-
+	temp.DefaultRole = NormalizeRoleName(temp.DefaultRole)
 	for group, role := range temp.GroupMappings {
-		if _, err := types.NewRole(strings.ToUpper(role)); err != nil {
-			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "invalid role %s for group %s", role, group)
-		}
+		temp.GroupMappings[group] = NormalizeRoleName(role)
 	}
 
-	*typ = RoleMapping(temp)
+	*roleMapping = RoleMapping(temp)
 	return nil
 }
 
-func (roleMapping *RoleMapping) NewRoleFromCallbackIdentity(callbackIdentity *CallbackIdentity) types.Role {
+func (roleMapping *RoleMapping) NewRolesFromCallbackIdentity(callbackIdentity *CallbackIdentity, roleAttributeExists bool) []string {
 	if roleMapping == nil {
-		return types.RoleViewer
+		return []string{SigNozViewerRoleName}
 	}
 
-	if roleMapping.UseRoleAttribute && callbackIdentity.Role != "" {
-		if role, err := types.NewRole(strings.ToUpper(callbackIdentity.Role)); err == nil {
-			return role
-		}
+	if roleAttributeExists {
+		return []string{NormalizeRoleName(callbackIdentity.Role)}
 	}
 
 	if len(roleMapping.GroupMappings) > 0 && len(callbackIdentity.Groups) > 0 {
-		highestRole := types.RoleViewer
-		found := false
-
+		roleNames := make([]string, 0)
+		seen := make(map[string]struct{})
 		for _, group := range callbackIdentity.Groups {
-			if mappedRole, exists := roleMapping.GroupMappings[group]; exists {
-				found = true
-				if role, err := types.NewRole(strings.ToUpper(mappedRole)); err == nil {
-					if compareRoles(role, highestRole) > 0 {
-						highestRole = role
-					}
-				}
+			roleName, exists := roleMapping.GroupMappings[group]
+			if !exists {
+				continue
 			}
+			if _, duplicate := seen[roleName]; duplicate {
+				continue
+			}
+			seen[roleName] = struct{}{}
+			roleNames = append(roleNames, roleName)
 		}
-
-		if found {
-			return highestRole
+		if len(roleNames) > 0 {
+			return roleNames
 		}
 	}
 
+	return []string{roleMapping.DefaultRoleName()}
+}
+
+func (roleMapping *RoleMapping) DefaultRoleName() string {
+	if roleMapping.DefaultRole != "" {
+		return roleMapping.DefaultRole
+	}
+
+	return SigNozViewerRoleName
+}
+
+func (roleMapping *RoleMapping) RoleNames() []string {
+	if roleMapping == nil {
+		return nil
+	}
+
+	seen := make(map[string]struct{})
+	roleNames := make([]string, 0, len(roleMapping.GroupMappings)+1)
+
 	if roleMapping.DefaultRole != "" {
-		if role, err := types.NewRole(strings.ToUpper(roleMapping.DefaultRole)); err == nil {
-			return role
+		seen[roleMapping.DefaultRole] = struct{}{}
+		roleNames = append(roleNames, roleMapping.DefaultRole)
+	}
+
+	for _, roleName := range roleMapping.GroupMappings {
+		if roleName == "" {
+			continue
 		}
+		if _, duplicate := seen[roleName]; duplicate {
+			continue
+		}
+		seen[roleName] = struct{}{}
+		roleNames = append(roleNames, roleName)
 	}
 
-	return types.RoleViewer
-}
-
-func compareRoles(a, b types.Role) int {
-	order := map[types.Role]int{
-		types.RoleViewer: 0,
-		types.RoleEditor: 1,
-		types.RoleAdmin:  2,
-	}
-	return order[a] - order[b]
+	return roleNames
 }

pkg/types/authtypes/role.go

@@ -25,6 +25,7 @@ var (
 	ErrCodeRoleUnsupported                  = errors.MustNewCode("role_unsupported")
 	ErrCodeRoleHasUserAssignees             = errors.MustNewCode("role_has_user_assignees")
 	ErrCodeRoleHasServiceAccountAssignees   = errors.MustNewCode("role_has_service_account_assignees")
+	ErrCodeRoleHasAuthDomainMappings        = errors.MustNewCode("role_has_auth_domain_mappings")
 )
 
 var (
@@ -298,6 +299,20 @@ func MustGetSigNozManagedRoleFromExistingRole(role types.Role) string {
 	return managedRole
 }
 
+func NormalizeRoleName(role string) string {
+	legacyRole, err := types.NewRole(strings.ToUpper(role))
+	if err != nil {
+		return role
+	}
+
+	managedRole, ok := ExistingRoleToSigNozManagedRoleMap[legacyRole]
+	if !ok {
+		return role
+	}
+
+	return managedRole
+}
+
 type RoleStore interface {
 	Create(context.Context, *Role) error
 	Get(context.Context, valuer.UUID, valuer.UUID) (*Role, error)

pkg/types/querybuildertypes/querybuildertypesv5/builder_elements.go

@@ -480,7 +480,9 @@ type MetricAggregation struct {
 	// value filter to apply to the query
 	ValueFilter *metrictypes.MetricValueFilter `json:"-"`
 	// reduce to operator for metric scalar requests
-	ReduceTo ReduceTo `json:"reduceTo,omitempty"`
+	ReduceTo ReduceTo `json:"reduceTo,omitzero"`
+
+	Reduced bool `json:"-"`
 }
 
 // Copy creates a deep copy of MetricAggregation.

pkg/types/telemetrytypes/store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
 // MetadataStore is the interface for the telemetry metadata store.
@@ -26,12 +27,12 @@ type MetadataStore interface {
 	GetAllValues(ctx context.Context, fieldValueSelector *FieldValueSelector) (*TelemetryFieldValues, bool, error)
 
 	// FetchTemporality fetches the temporality for metric
-	FetchTemporality(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
+	FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
 
 	// FetchTemporalityMulti fetches the temporality for multiple metrics
-	FetchTemporalityMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
+	FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
 
-	FetchTemporalityAndTypeMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error)
+	FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error)
 
 	// ListLogsJSONIndexes lists the JSON indexes for the logs table.
 	ListLogsJSONIndexes(ctx context.Context, filters ...string) ([]TelemetryFieldKeySkipIndex, error)

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
 // MockMetadataStore implements the MetadataStore interface for testing purposes.
@@ -16,6 +17,7 @@ type MockMetadataStore struct {
 	AllValuesMap               map[string]*telemetrytypes.TelemetryFieldValues
 	TemporalityMap             map[string]metrictypes.Temporality
 	TypeMap                    map[string]metrictypes.Type
+	ReducedMap                 map[string]bool
 	PromotedPathsMap           map[string]bool
 	LogsJSONIndexes            []telemetrytypes.TelemetryFieldKeySkipIndex
 	ColumnEvolutionMetadataMap map[string][]*telemetrytypes.EvolutionEntry
@@ -306,7 +308,7 @@ func (m *MockMetadataStore) SetAllValues(lookupKey string, values *telemetrytype
 }
 
 // FetchTemporality fetches the temporality for a metric.
-func (m *MockMetadataStore) FetchTemporality(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error) {
+func (m *MockMetadataStore) FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error) {
 	if temporality, exists := m.TemporalityMap[metricName]; exists {
 		return temporality, nil
 	}
@@ -314,7 +316,7 @@ func (m *MockMetadataStore) FetchTemporality(ctx context.Context, queryTimeRange
 }
 
 // FetchTemporalityMulti fetches the temporality for multiple metrics.
-func (m *MockMetadataStore) FetchTemporalityMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error) {
+func (m *MockMetadataStore) FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error) {
 	result := make(map[string]metrictypes.Temporality)
 
 	for _, metricName := range metricNames {
@@ -329,9 +331,10 @@ func (m *MockMetadataStore) FetchTemporalityMulti(ctx context.Context, queryTime
 }
 
 // FetchTemporalityMulti fetches the temporality for multiple metrics.
-func (m *MockMetadataStore) FetchTemporalityAndTypeMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {
+func (m *MockMetadataStore) FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
 	temporalities := make(map[string]metrictypes.Temporality)
 	types := make(map[string]metrictypes.Type)
+	reduced := make(map[string]bool)
 
 	for _, metricName := range metricNames {
 		if temporality, exists := m.TemporalityMap[metricName]; exists {
@@ -344,9 +347,12 @@ func (m *MockMetadataStore) FetchTemporalityAndTypeMulti(ctx context.Context, qu
 		} else {
 			types[metricName] = metrictypes.UnspecifiedType
 		}
+		if m.ReducedMap[metricName] {
+			reduced[metricName] = true
+		}
 	}
 
-	return temporalities, types, nil
+	return temporalities, types, reduced, nil
 }
 
 // SetTemporality sets the temporality for a metric in the mock store.