Merge branch 'main' into issue-5535

.github/workflows/integrationci.yaml

@@ -56,6 +56,8 @@ jobs:
           - querier_json_body
           - querier_skip_resource_fingerprint
           - ttl
+          - clickhousecluster
+          - metricreduction
         sqlstore-provider:
           - postgres
           - sqlite

pkg/telemetrymetrics/reduced_test.go

@@ -39,48 +39,48 @@ func TestReducedStatementBuilder(t *testing.T) {
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`min`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`max`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric.sum", uint64(1746999600000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999600000), uint64(1747172760000)},
 			},
 		},
 		{
@@ -103,16 +103,16 @@ func TestReducedStatementBuilder(t *testing.T) {
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, `le`, argMax(`sum`, points.computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'le') AS `le` FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli, `le`) 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), false, "test.metric.bucket", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 		{
 			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.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},
+				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 points.reduced_fingerprint AS fingerprint, points.unix_milli AS unix_milli, argMax(`sum_last`, points.computed_at) AS value, argMax(`count_series`, points.computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.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.reduced_fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, unix_milli) 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), false, "test.metric", uint64(1746999900000), uint64(1747172760000)},
 			},
 		},
 	}

pkg/telemetrymetrics/statement_builder.go

@@ -338,19 +338,24 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 
 	// 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.Select("points.reduced_fingerprint AS fingerprint", "points.unix_milli AS unix_milli")
+	for _, g := range query.GroupBy {
+		dedup.SelectMore(fmt.Sprintf("`%s`", g.Name))
 	}
-	dedup.From(fmt.Sprintf("%s.%s", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	dedup.SelectMore(fmt.Sprintf("argMax(%s, points.computed_at) AS value", value))
+	if weight != "" {
+		dedup.SelectMore(fmt.Sprintf("argMax(%s, points.computed_at) AS weight", weight))
+	}
+	dedup.From(fmt.Sprintf("%s.%s AS points", DBName, WhichReducedSamplesTableToUse(agg.Type)))
+	dedup.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
 	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)
+	dedup.GroupBy("fingerprint", "unix_milli")
+	dedup.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
+	dedupQuery, dedupArgs := dedup.BuildWithFlavor(sqlbuilder.ClickHouse, timeSeriesCTEArgs...)
 
 	sb := sqlbuilder.NewSelectBuilder()
 	sb.Select("fingerprint")
@@ -364,13 +369,11 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
 		// 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.From(fmt.Sprintf("(%s)", dedupQuery))
 	sb.GroupBy("fingerprint", "ts")
 	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
 
-	initArgs := append(append([]any{}, dedupArgs...), timeSeriesCTEArgs...)
-	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, initArgs...)
+	q, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, dedupArgs...)
 	return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, true
 }
 

tests/fixtures/clickhouse.py

@@ -2,6 +2,7 @@ import os
 from collections.abc import Callable, Generator
 from datetime import datetime
 from typing import Any
+from uuid import uuid4
 
 import clickhouse_connect
 import clickhouse_connect.driver
@@ -10,37 +11,93 @@ import docker
 import docker.errors
 import pytest
 from testcontainers.clickhouse import ClickHouseContainer
-from testcontainers.core.container import Network
+from testcontainers.core.container import DockerContainer, Network
 
 from fixtures import reuse, types
 from fixtures.logger import setup_logger
 
 logger = setup_logger(__name__)
 
+CLICKHOUSE_USERNAME = "signoz"
+CLICKHOUSE_PASSWORD = "password"
 
-@pytest.fixture(name="clickhouse", scope="package")
-def clickhouse(
-    tmpfs: Generator[types.LegacyPath, Any],
-    network: Network,
-    zookeeper: types.TestContainerDocker,
-    request: pytest.FixtureRequest,
-    pytestconfig: pytest.Config,
-) -> types.TestContainerClickhouse:
-    """
-    Package-scoped fixture for Clickhouse TestContainer.
+CUSTOM_FUNCTION_CONFIG = """
+<functions>
+    <function>
+        <type>executable</type>
+        <name>histogramQuantile</name>
+        <return_type>Float64</return_type>
+        <argument>
+            <type>Array(Float64)</type>
+            <name>buckets</name>
+        </argument>
+        <argument>
+            <type>Array(Float64)</type>
+            <name>counts</name>
+        </argument>
+        <argument>
+            <type>Float64</type>
+            <name>quantile</name>
+        </argument>
+        <format>CSV</format>
+        <command>./histogramQuantile</command>
+    </function>
+</functions>
+"""
+
+# Distributed inserts to a remote shard are async by default. We force
+# sycn at the profile level for deterministic tests.
+CLUSTER_USERS_CONFIG = """
+<clickhouse>
+    <profiles>
+        <default>
+            <insert_distributed_sync>1</insert_distributed_sync>
+        </default>
+    </profiles>
+</clickhouse>
+"""
+
+
+def render_remote_servers(shard_hosts: list[tuple[str, int]], secret: str | None = None) -> str:
+    """Render the <remote_servers> block for a cluster named `cluster` with one
+    single-replica shard per (host, port).
     """
+    shards = "".join(
+        f"""
+                <shard>
+                    <replica>
+                        <host>{host}</host>
+                        <port>{port}</port>
+                    </replica>
+                </shard>"""
+        for host, port in shard_hosts
+    )
 
-    def create() -> types.TestContainerClickhouse:
-        version = request.config.getoption("--clickhouse-version")
+    # Multi-node clusters need `secret` because distributed queries otherwise
+    # authenticate as the `default` user, which the docker entrypoint restricts
+    # to localhost when a custom user is configured.
+    secret_block = (
+        f"""
+                    <secret>{secret}</secret>"""
+        if secret
+        else ""
+    )
 
-        container = ClickHouseContainer(
-            image=f"clickhouse/clickhouse-server:{version}",
-            port=9000,
-            username="signoz",
-            password="password",
-        )
+    return f"""
+            <remote_servers>
+                <cluster>{secret_block}{shards}
+                </cluster>
+            </remote_servers>"""
 
-        cluster_config = f"""
+
+def render_node_config(
+    zookeeper_address: str,
+    zookeeper_port: int,
+    shard: str,
+    remote_servers: str,
+    distributed_ddl_path: str = "/clickhouse/task_queue/ddl",
+) -> str:
+    return f"""
         <clickhouse>
             <logger>
                 <level>information</level>
@@ -55,33 +112,23 @@ def clickhouse(
             </logger>
 
             <macros>
-                <shard>01</shard>
+                <shard>{shard}</shard>
                 <replica>01</replica>
             </macros>
 
             <zookeeper>
                 <node>
-                    <host>{zookeeper.container_configs["2181"].address}</host>
-                    <port>{zookeeper.container_configs["2181"].port}</port>
+                    <host>{zookeeper_address}</host>
+                    <port>{zookeeper_port}</port>
                 </node>
             </zookeeper>
-
-            <remote_servers>
-                <cluster>
-                    <shard>
-                        <replica>
-                            <host>127.0.0.1</host>
-                            <port>9000</port>
-                        </replica>
-                    </shard>
-                </cluster>
-            </remote_servers>
+{remote_servers}
 
             <user_defined_executable_functions_config>*function.xml</user_defined_executable_functions_config>
             <user_scripts_path>/var/lib/clickhouse/user_scripts/</user_scripts_path>
 
             <distributed_ddl>
-                <path>/clickhouse/task_queue/ddl</path>
+                <path>{distributed_ddl_path}</path>
                 <profile>default</profile>
             </distributed_ddl>
 
@@ -122,38 +169,66 @@ def clickhouse(
         </clickhouse>
         """
 
-        custom_function_config = """
-        <functions>
-            <function>
-                <type>executable</type>
-                <name>histogramQuantile</name>
-                <return_type>Float64</return_type>
-                <argument>
-                    <type>Array(Float64)</type>
-                    <name>buckets</name>
-                </argument>
-                <argument>
-                    <type>Array(Float64)</type>
-                    <name>counts</name>
-                </argument>
-                <argument>
-                    <type>Float64</type>
-                    <name>quantile</name>
-                </argument>
-                <format>CSV</format>
-                <command>./histogramQuantile</command>
-            </function>
-        </functions>
-        """
 
-        tmp_dir = tmpfs("clickhouse")
+def install_histogram_quantile(container: ClickHouseContainer) -> None:
+    wrapped = container.get_wrapped_container()
+    exit_code, output = wrapped.exec_run(
+        [
+            "bash",
+            "-c",
+            (
+                'version="v0.0.1" && '
+                'node_os=$(uname -s | tr "[:upper:]" "[:lower:]") && '
+                "node_arch=$(uname -m | sed s/aarch64/arm64/ | sed s/x86_64/amd64/) && "
+                "cd /tmp && "
+                'wget -O histogram-quantile.tar.gz "https://github.com/SigNoz/signoz/releases/download/histogram-quantile%2F${version}/histogram-quantile_${node_os}_${node_arch}.tar.gz" && '
+                "tar -xzf histogram-quantile.tar.gz && "
+                "mkdir -p /var/lib/clickhouse/user_scripts && "
+                "mv histogram-quantile /var/lib/clickhouse/user_scripts/histogramQuantile && "
+                "chmod +x /var/lib/clickhouse/user_scripts/histogramQuantile"
+            ),
+        ],
+    )
+    if exit_code != 0:
+        raise RuntimeError(f"Failed to install histogramQuantile binary: {output.decode()}")
+
+
+def create_clickhouse(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhouse",
+    version: str | None = None,
+) -> types.TestContainerClickhouse:
+    coordinator = next(iter(keeper.container_configs.values()))
+
+    def create() -> types.TestContainerClickhouse:
+        clickhouse_version = version or request.config.getoption("--clickhouse-version")
+
+        container = ClickHouseContainer(
+            image=f"clickhouse/clickhouse-server:{clickhouse_version}",
+            port=9000,
+            username=CLICKHOUSE_USERNAME,
+            password=CLICKHOUSE_PASSWORD,
+        )
+
+        cluster_config = render_node_config(
+            zookeeper_address=coordinator.address,
+            zookeeper_port=coordinator.port,
+            shard="01",
+            remote_servers=render_remote_servers([("127.0.0.1", 9000)]),
+        )
+
+        tmp_dir = tmpfs(cache_key)
         cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
         with open(cluster_config_file_path, "w", encoding="utf-8") as f:
             f.write(cluster_config)
 
         custom_function_file_path = os.path.join(tmp_dir, "custom-function.xml")
         with open(custom_function_file_path, "w", encoding="utf-8") as f:
-            f.write(custom_function_config)
+            f.write(CUSTOM_FUNCTION_CONFIG)
 
         container.with_volume_mapping(cluster_config_file_path, "/etc/clickhouse-server/config.d/cluster.xml")
         container.with_volume_mapping(
@@ -163,27 +238,7 @@ def clickhouse(
         container.with_network(network)
         container.start()
 
-        # Download and install the histogramQuantile binary
-        wrapped = container.get_wrapped_container()
-        exit_code, output = wrapped.exec_run(
-            [
-                "bash",
-                "-c",
-                (
-                    'version="v0.0.1" && '
-                    'node_os=$(uname -s | tr "[:upper:]" "[:lower:]") && '
-                    "node_arch=$(uname -m | sed s/aarch64/arm64/ | sed s/x86_64/amd64/) && "
-                    "cd /tmp && "
-                    'wget -O histogram-quantile.tar.gz "https://github.com/SigNoz/signoz/releases/download/histogram-quantile%2F${version}/histogram-quantile_${node_os}_${node_arch}.tar.gz" && '
-                    "tar -xzf histogram-quantile.tar.gz && "
-                    "mkdir -p /var/lib/clickhouse/user_scripts && "
-                    "mv histogram-quantile /var/lib/clickhouse/user_scripts/histogramQuantile && "
-                    "chmod +x /var/lib/clickhouse/user_scripts/histogramQuantile"
-                ),
-            ],
-        )
-        if exit_code != 0:
-            raise RuntimeError(f"Failed to install histogramQuantile binary: {output.decode()}")
+        install_histogram_quantile(container)
 
         connection = clickhouse_connect.get_client(
             user=container.username,
@@ -253,7 +308,7 @@ def clickhouse(
     return reuse.wrap(
         request,
         pytestconfig,
-        "clickhouse",
+        cache_key,
         empty=lambda: types.TestContainerSQL(
             container=types.TestContainerDocker(id="", host_configs={}, container_configs={}),
             conn=None,
@@ -265,6 +320,334 @@ def clickhouse(
     )
 
 
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    zookeeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    """
+    Package-scoped fixture for Clickhouse TestContainer.
+    """
+    return create_clickhouse(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=zookeeper,
+        request=request,
+        pytestconfig=pytestconfig,
+    )
+
+
+def local_series_counts(
+    node_conns: list[clickhouse_connect.driver.client.Client],
+    table: str,
+    metric_name: str,
+) -> list[int]:
+    """Distinct series per node via the LOCAL (non-distributed) table."""
+    return [
+        int(
+            conn.query(
+                f"SELECT count(DISTINCT fingerprint) FROM signoz_metrics.{table} WHERE metric_name = %(metric_name)s",
+                parameters={"metric_name": metric_name},
+            ).result_rows[0][0]
+        )
+        for conn in node_conns
+    ]
+
+
+def assert_spans_shards(
+    node_conns: list[clickhouse_connect.driver.client.Client],
+    table: str,
+    metric_name: str,
+    total: int,
+) -> None:
+    """Guard for distributed tests: a green run on a cluster proves nothing
+    unless the seeded series actually landed on more than one shard."""
+    counts = local_series_counts(node_conns, table, metric_name)
+    assert sum(counts) == total, f"expected {total} series in {table} across shards, got {counts}"
+    assert min(counts) > 0, f"seeded series in {table} all landed on one shard: {counts}"
+
+
+@pytest.fixture(name="clickhouse_node_conns", scope="function")
+def clickhouse_node_conns(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[list[clickhouse_connect.driver.client.Client], Any]:
+    """Per-node clients (index 0 = the initiator) for asserting shard-local
+    state via the local, non-distributed tables. Empty for single-node
+    fixtures, which don't populate `nodes`."""
+    conns = [
+        clickhouse_connect.get_client(
+            user=clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME"],
+            password=clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD"],
+            host=node.host_configs["8123"].address,
+            port=node.host_configs["8123"].port,
+        )
+        for node in clickhouse.nodes
+    ]
+    yield conns
+    for conn in conns:
+        conn.close()
+
+
+KEEPER_CONFIG = """
+<clickhouse>
+    <listen_host>0.0.0.0</listen_host>
+    <keeper_server>
+        <tcp_port>9181</tcp_port>
+        <server_id>1</server_id>
+        <log_storage_path>/var/lib/clickhouse-keeper/coordination/log</log_storage_path>
+        <snapshot_storage_path>/var/lib/clickhouse-keeper/coordination/snapshots</snapshot_storage_path>
+        <coordination_settings>
+            <operation_timeout_ms>10000</operation_timeout_ms>
+            <session_timeout_ms>30000</session_timeout_ms>
+            <raft_logs_level>warning</raft_logs_level>
+        </coordination_settings>
+        <raft_configuration>
+            <server>
+                <id>1</id>
+                <hostname>localhost</hostname>
+                <port>9234</port>
+            </server>
+        </raft_configuration>
+    </keeper_server>
+</clickhouse>
+"""
+
+
+def create_clickhouse_keeper(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhousekeeper",
+    version: str | None = None,
+) -> types.TestContainerDocker:
+
+    def create() -> types.TestContainerDocker:
+        keeper_version = version or request.config.getoption("--clickhouse-version")
+
+        tmp_dir = tmpfs(cache_key)
+        keeper_config_file_path = os.path.join(tmp_dir, "keeper_config.xml")
+        with open(keeper_config_file_path, "w", encoding="utf-8") as f:
+            f.write(KEEPER_CONFIG)
+
+        container = DockerContainer(image=f"clickhouse/clickhouse-keeper:{keeper_version}")
+        container.with_volume_mapping(keeper_config_file_path, "/etc/clickhouse-keeper/keeper_config.xml")
+        container.with_exposed_ports(9181)
+        container.with_network(network=network)
+
+        container.start()
+        return types.TestContainerDocker(
+            id=container.get_wrapped_container().id,
+            host_configs={
+                "9181": types.TestContainerUrlConfig(
+                    scheme="tcp",
+                    address=container.get_container_host_ip(),
+                    port=container.get_exposed_port(9181),
+                )
+            },
+            container_configs={
+                "9181": types.TestContainerUrlConfig(
+                    scheme="tcp",
+                    address=container.get_wrapped_container().name,
+                    port=9181,
+                )
+            },
+        )
+
+    def delete(container: types.TestContainerDocker):
+        client = docker.from_env()
+        try:
+            client.containers.get(container_id=container.id).stop()
+            client.containers.get(container_id=container.id).remove(v=True)
+        except docker.errors.NotFound:
+            logger.info(
+                "Skipping removal of ClickHouse Keeper, Keeper(%s) not found. Maybe it was manually removed?",
+                {"id": container.id},
+            )
+
+    def restore(cache: dict) -> types.TestContainerDocker:
+        return types.TestContainerDocker.from_cache(cache)
+
+    return reuse.wrap(
+        request,
+        pytestconfig,
+        cache_key,
+        lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+        create,
+        delete,
+        restore,
+    )
+
+
+def create_clickhouse_cluster(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "clickhouse_cluster",
+    shards: int = 2,
+    version: str | None = None,
+) -> types.TestContainerClickhouse:
+    """
+    To some extent, taken inspiration from how ClickHouse's own integration
+    harness composes real clusters: deterministic hostnames
+    (network aliases), per-node shard macros, and a shared cluster definition
+    named `cluster`.
+
+    `conn`/`env` point at node 1 i.e the initiator every query-service query and
+    migration goes through. Per-node containers are exposed via `nodes` so
+    tests can assert shard-local state. `keeper` is any coordination service
+    (ZooKeeper or ClickHouse Keeper).
+    """
+    coordinator = next(iter(keeper.container_configs.values()))
+
+    def create() -> types.TestContainerClickhouse:
+        clickhouse_version = version or request.config.getoption("--clickhouse-version")
+
+        # Unique aliases per creation: docker allows duplicate network aliases
+        # (DNS round-robin), so a stale cluster must never share names with a
+        # fresh one.
+        suffix = uuid4().hex[:6]
+        aliases = [f"signoz-ch-{suffix}-{i:02d}" for i in range(1, shards + 1)]
+        remote_servers = render_remote_servers([(alias, 9000) for alias in aliases], secret=cache_key)
+        # Own DDL queue path: the keeper instance may be shared with other
+        # environments under --reuse; its DDL queue stays separate.
+        distributed_ddl_path = f"/clickhouse/{cache_key}-{suffix}/task_queue/ddl"
+
+        nodes: list[types.TestContainerDocker] = []
+        started: list[ClickHouseContainer] = []
+        try:
+            for i, alias in enumerate(aliases, start=1):
+                node_config = render_node_config(
+                    zookeeper_address=coordinator.address,
+                    zookeeper_port=coordinator.port,
+                    shard=f"{i:02d}",
+                    remote_servers=remote_servers,
+                    distributed_ddl_path=distributed_ddl_path,
+                )
+
+                tmp_dir = tmpfs(f"clickhouse-{suffix}-{i:02d}")
+                cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
+                with open(cluster_config_file_path, "w", encoding="utf-8") as f:
+                    f.write(node_config)
+                custom_function_file_path = os.path.join(tmp_dir, "custom-function.xml")
+                with open(custom_function_file_path, "w", encoding="utf-8") as f:
+                    f.write(CUSTOM_FUNCTION_CONFIG)
+                users_config_file_path = os.path.join(tmp_dir, "users.xml")
+                with open(users_config_file_path, "w", encoding="utf-8") as f:
+                    f.write(CLUSTER_USERS_CONFIG)
+
+                container = ClickHouseContainer(
+                    image=f"clickhouse/clickhouse-server:{clickhouse_version}",
+                    port=9000,
+                    username=CLICKHOUSE_USERNAME,
+                    password=CLICKHOUSE_PASSWORD,
+                )
+                container.with_volume_mapping(cluster_config_file_path, "/etc/clickhouse-server/config.d/cluster.xml")
+                container.with_volume_mapping(custom_function_file_path, "/etc/clickhouse-server/custom-function.xml")
+                container.with_volume_mapping(users_config_file_path, "/etc/clickhouse-server/users.d/integration-cluster.xml")
+                container.with_network(network)
+                container.with_network_aliases(alias)
+                container.start()
+                started.append(container)
+
+                install_histogram_quantile(container)
+
+                nodes.append(
+                    types.TestContainerDocker(
+                        id=container.get_wrapped_container().id,
+                        host_configs={
+                            "9000": types.TestContainerUrlConfig(
+                                "tcp",
+                                container.get_container_host_ip(),
+                                container.get_exposed_port(9000),
+                            ),
+                            "8123": types.TestContainerUrlConfig(
+                                "tcp",
+                                container.get_container_host_ip(),
+                                container.get_exposed_port(8123),
+                            ),
+                        },
+                        container_configs={
+                            "9000": types.TestContainerUrlConfig("tcp", alias, 9000),
+                            "8123": types.TestContainerUrlConfig("tcp", alias, 8123),
+                        },
+                    )
+                )
+        except Exception:
+            for container in started:
+                container.stop()
+            raise
+
+        connection = clickhouse_connect.get_client(
+            user=CLICKHOUSE_USERNAME,
+            password=CLICKHOUSE_PASSWORD,
+            host=nodes[0].host_configs["8123"].address,
+            port=nodes[0].host_configs["8123"].port,
+        )
+
+        return types.TestContainerClickhouse(
+            container=nodes[0],
+            conn=connection,
+            env={
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": f"tcp://{CLICKHOUSE_USERNAME}:{CLICKHOUSE_PASSWORD}@{aliases[0]}:{9000}",
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME": CLICKHOUSE_USERNAME,
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD": CLICKHOUSE_PASSWORD,
+                "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER": "cluster",
+            },
+            nodes=nodes,
+        )
+
+    def delete(resource: types.TestContainerClickhouse) -> None:
+        client = docker.from_env()
+        for node in resource.nodes or [resource.container]:
+            try:
+                client.containers.get(container_id=node.id).stop()
+                client.containers.get(container_id=node.id).remove(v=True)
+            except docker.errors.NotFound:
+                logger.info(
+                    "Skipping removal of Clickhouse cluster node, node(%s) not found. Maybe it was manually removed?",
+                    {"id": node.id},
+                )
+
+    def restore(cache: dict) -> types.TestContainerClickhouse:
+        nodes = [types.TestContainerDocker.from_cache(node) for node in cache["nodes"]]
+        env = cache["env"]
+        host_config = nodes[0].host_configs["8123"]
+
+        conn = clickhouse_connect.get_client(
+            user=env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_USERNAME"],
+            password=env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_PASSWORD"],
+            host=host_config.address,
+            port=host_config.port,
+        )
+
+        return types.TestContainerClickhouse(
+            container=nodes[0],
+            conn=conn,
+            env=env,
+            nodes=nodes,
+        )
+
+    return reuse.wrap(
+        request,
+        pytestconfig,
+        cache_key,
+        empty=lambda: types.TestContainerClickhouse(
+            container=types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+            conn=None,
+            env={},
+        ),
+        create=create,
+        delete=delete,
+        restore=restore,
+    )
+
+
 @pytest.fixture(name="check_query_log")
 def check_query_log(
     signoz: types.SigNoz,

tests/fixtures/http.py

@@ -18,19 +18,22 @@ from fixtures.logger import setup_logger
 logger = setup_logger(__name__)
 
 
-@pytest.fixture(name="zeus", scope="package")
-def zeus(
+ZEUS_NETWORK_ALIAS = "signoz-zeus-it"
+
+
+def create_zeus(
     network: Network,
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
+    cache_key: str = "zeus",
+    alias: str | None = None,
 ) -> types.TestContainerDocker:
-    """
-    Package-scoped fixture for running zeus
-    """
 
     def create() -> types.TestContainerDocker:
         container = WireMockContainer(image="wiremock/wiremock:2.35.1-1", secure=False)
         container.with_network(network)
+        if alias:
+            container.with_network_aliases(alias)
         container.start()
 
         return types.TestContainerDocker(
@@ -42,7 +45,7 @@ def zeus(
                     container.get_exposed_port(8080),
                 )
             },
-            container_configs={"8080": types.TestContainerUrlConfig("http", container.get_wrapped_container().name, 8080)},
+            container_configs={"8080": types.TestContainerUrlConfig("http", alias or container.get_wrapped_container().name, 8080)},
         )
 
     def delete(container: types.TestContainerDocker):
@@ -62,7 +65,7 @@ def zeus(
     return reuse.wrap(
         request,
         pytestconfig,
-        "zeus",
+        cache_key,
         lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
         create,
         delete,
@@ -70,6 +73,18 @@ def zeus(
     )
 
 
+@pytest.fixture(name="zeus", scope="package")
+def zeus(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    """
+    Package-scoped fixture for running zeus
+    """
+    return create_zeus(network=network, request=request, pytestconfig=pytestconfig)
+
+
 @pytest.fixture(name="gateway", scope="package")
 def gateway(
     network: Network,

tests/fixtures/metricreduction.py

@@ -0,0 +1,51 @@
+import datetime
+from collections.abc import Sequence
+
+from fixtures.metrics import MetricsBufferSample, MetricsBufferTimeSeries
+
+
+def build_ruled_gauge_buffer(
+    metric_name: str,
+    base_epoch: int,
+    services: Sequence[str],
+    pods_per_service: int,
+    minutes: int,
+    value: float = 1.0,
+) -> tuple[list[MetricsBufferTimeSeries], list[MetricsBufferSample]]:
+    """Collector-shaped buffer rows for a gauge under a reduction rule that
+    keeps `service`: per raw series a raw series row (is_reduced=false, full
+    labels, reduced_fingerprint -> group) plus the group's reduced series row
+    (is_reduced=true, kept labels), and one raw sample per series per minute
+    carrying both fingerprints. Returns (time_series, samples) for
+    insert_buffer_metrics."""
+    reduced_series = {
+        service: MetricsBufferTimeSeries(
+            metric_name=metric_name,
+            labels={"service": service},
+            timestamp=datetime.datetime.fromtimestamp(base_epoch, tz=datetime.UTC),
+            is_reduced=True,
+        )
+        for service in services
+    }
+    raw_series = [
+        MetricsBufferTimeSeries(
+            metric_name=metric_name,
+            labels={"service": service, "pod": f"pod-{service}-{i}"},
+            timestamp=datetime.datetime.fromtimestamp(base_epoch, tz=datetime.UTC),
+            reduced_fingerprint=reduced_series[service].fingerprint,
+        )
+        for service in services
+        for i in range(pods_per_service)
+    ]
+    samples = [
+        MetricsBufferSample(
+            metric_name=metric_name,
+            fingerprint=ts.fingerprint,
+            timestamp=datetime.datetime.fromtimestamp(base_epoch + minute * 60, tz=datetime.UTC),
+            value=value,
+            reduced_fingerprint=ts.reduced_fingerprint,
+        )
+        for ts in raw_series
+        for minute in range(minutes)
+    ]
+    return raw_series + list(reduced_series.values()), samples

tests/fixtures/metrics.py

@@ -11,6 +11,14 @@ import pytest
 from fixtures import types
 from fixtures.time import parse_timestamp
 
+_REDUCED_METRICS_TABLES_TO_TRUNCATE = [
+    "time_series_v4_reduced",
+    "samples_v4_reduced_last_60s",
+    "samples_v4_reduced_sum_60s",
+    "time_series_v4_buffer",
+    "samples_v4_buffer",
+]
+
 
 class MetricsTimeSeries(ABC):
     """Represents a row in the time_series_v4 table."""
@@ -414,6 +422,267 @@ class Metrics(ABC):
         return metrics
 
 
+class MetricsReducedTimeSeries(ABC):
+    """Represents a row in the time_series_v4_reduced table i.e what
+    the time_series_v4_reduced_mv materializes for a metric under a
+    reduction rule. One row per kept-label group. `fingerprint` holds the
+    reduced fingerprint and `labels` contains only the kept labels.
+
+    The fingerprint recipe (md5, like MetricsTimeSeries) does not match the
+    collector's real hash; it only needs to be consistent with the
+    reduced_fingerprint used in the reduced samples rows.
+    """
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        kept_labels: dict[str, str],
+        timestamp: datetime.datetime,
+        temporality: str = "Unspecified",
+        description: str = "",
+        unit: str = "",
+        type_: str = "Gauge",
+        is_monotonic: bool = False,
+        env: str = "default",
+    ) -> None:
+        kept_labels = dict(kept_labels)
+        kept_labels["__name__"] = metric_name
+        self.env = env
+        # mirror time_series_v4_reduced_mv: monotonic cumulative counters are
+        # reduced as deltas
+        if temporality == "Cumulative" and is_monotonic:
+            temporality = "Delta"
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.description = description
+        self.unit = unit
+        self.type = type_
+        self.is_monotonic = is_monotonic
+        self.labels = json.dumps(kept_labels, separators=(",", ":"))
+        self.attrs = kept_labels
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
+        self.normalized = False
+
+        fingerprint_str = metric_name + self.labels
+        self.fingerprint = np.uint64(int(hashlib.md5(fingerprint_str.encode()).hexdigest()[:16], 16))
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.description,
+            self.unit,
+            self.type,
+            self.is_monotonic,
+            self.fingerprint,
+            self.unix_milli,
+            self.labels,
+            self.attrs,
+            {},
+            {},
+            self.normalized,
+        ]
+
+
+class MetricsReducedSampleLast60s(ABC):
+    """Represents a row in the samples_v4_reduced_last_60s table. One 60s
+    bucket per reduced group, as the samples_v4_reduced_last_60s_mv refresh
+    would emit it (gauges and non-monotonic cumulative sums)."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        reduced_fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        sum_last: float,
+        min_value: float,
+        max_value: float,
+        sum_values: float,
+        count_series: int,
+        count_samples: int,
+        temporality: str = "Unspecified",
+        env: str = "default",
+        computed_at: datetime.datetime | None = None,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.reduced_fingerprint = reduced_fingerprint
+        # buckets are 60s-aligned: intDiv(unix_milli, 60000) * 60000
+        self.unix_milli = np.int64((int(timestamp.timestamp() * 1e3) // 60000) * 60000)
+        self.sum_last = np.float64(sum_last)
+        self.min = np.float64(min_value)
+        self.max = np.float64(max_value)
+        self.sum_values = np.float64(sum_values)
+        self.count_series = np.uint64(count_series)
+        self.count_samples = np.uint64(count_samples)
+        # the refresh stamps now(); default to shortly after the bucket closes
+        if computed_at is None:
+            computed_at = datetime.datetime.fromtimestamp(int(self.unix_milli) / 1e3, tz=datetime.UTC) + datetime.timedelta(seconds=180)
+        self.computed_at = computed_at
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.reduced_fingerprint,
+            self.unix_milli,
+            self.sum_last,
+            self.min,
+            self.max,
+            self.sum_values,
+            self.count_series,
+            self.count_samples,
+            self.computed_at,
+        ]
+
+
+class MetricsReducedSampleSum60s(ABC):
+    """Represents a row in the samples_v4_reduced_sum_60s table. One 60s
+    bucket per reduced group for delta counters and histograms."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        reduced_fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        sum_value: float,
+        count_series: int,
+        count_samples: int,
+        temporality: str = "Delta",
+        env: str = "default",
+        computed_at: datetime.datetime | None = None,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.reduced_fingerprint = reduced_fingerprint
+        self.unix_milli = np.int64((int(timestamp.timestamp() * 1e3) // 60000) * 60000)
+        self.sum = np.float64(sum_value)
+        self.count_series = np.uint64(count_series)
+        self.count_samples = np.uint64(count_samples)
+        if computed_at is None:
+            computed_at = datetime.datetime.fromtimestamp(int(self.unix_milli) / 1e3, tz=datetime.UTC) + datetime.timedelta(seconds=180)
+        self.computed_at = computed_at
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.reduced_fingerprint,
+            self.unix_milli,
+            self.sum,
+            self.count_series,
+            self.count_samples,
+            self.computed_at,
+        ]
+
+
+class MetricsBufferTimeSeries(ABC):
+    """Represents a row in the time_series_v4_buffer table. This is the collector's
+    universal landing target under cardinality control. For a ruled metric the
+    collector writes two rows per series: the raw one (is_reduced=false, full
+    labels, reduced_fingerprint pointing at its group) and the group's reduced
+    one (is_reduced=true, kept labels, fingerprint = reduced fingerprint)."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        labels: dict[str, str],
+        timestamp: datetime.datetime,
+        reduced_fingerprint: np.uint64 | int = 0,
+        is_reduced: bool = False,
+        temporality: str = "Unspecified",
+        description: str = "",
+        unit: str = "",
+        type_: str = "Gauge",
+        is_monotonic: bool = False,
+        env: str = "default",
+    ) -> None:
+        labels = dict(labels)
+        labels["__name__"] = metric_name
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.description = description
+        self.unit = unit
+        self.type = type_
+        self.is_monotonic = is_monotonic
+        self.reduced_fingerprint = np.uint64(reduced_fingerprint)
+        self.is_reduced = is_reduced
+        self.labels = json.dumps(labels, separators=(",", ":"))
+        self.attrs = labels
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
+        self.normalized = False
+
+        fingerprint_str = metric_name + self.labels
+        self.fingerprint = np.uint64(int(hashlib.md5(fingerprint_str.encode()).hexdigest()[:16], 16))
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.description,
+            self.unit,
+            self.type,
+            self.is_monotonic,
+            self.fingerprint,
+            self.reduced_fingerprint,
+            self.is_reduced,
+            self.unix_milli,
+            self.labels,
+            self.attrs,
+            {},
+            {},
+            self.normalized,
+        ]
+
+
+class MetricsBufferSample(ABC):
+    """Represents a row in the samples_v4_buffer table. Ruled samples carry
+    the raw fingerprint plus the group's reduced_fingerprint; unruled samples
+    have reduced_fingerprint = 0."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        metric_name: str,
+        fingerprint: np.uint64,
+        timestamp: datetime.datetime,
+        value: float,
+        reduced_fingerprint: np.uint64 | int = 0,
+        is_monotonic: bool = False,
+        temporality: str = "Unspecified",
+        env: str = "default",
+        flags: int = 0,
+    ) -> None:
+        self.env = env
+        self.temporality = temporality
+        self.metric_name = metric_name
+        self.fingerprint = fingerprint
+        self.reduced_fingerprint = np.uint64(reduced_fingerprint)
+        self.is_monotonic = is_monotonic
+        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
+        self.value = np.float64(value)
+        self.flags = np.uint32(flags)
+
+    def to_row(self) -> list:
+        return [
+            self.env,
+            self.temporality,
+            self.metric_name,
+            self.fingerprint,
+            self.reduced_fingerprint,
+            self.is_monotonic,
+            self.unix_milli,
+            self.value,
+            self.flags,
+        ]
+
+
 def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
     """
     Insert metrics into ClickHouse tables.
@@ -576,6 +845,163 @@ def insert_metrics(
     )
 
 
+def insert_reduced_metrics_to_clickhouse(
+    conn,
+    time_series: list[MetricsReducedTimeSeries],
+    last_samples: list[MetricsReducedSampleLast60s] | None = None,
+    sum_samples: list[MetricsReducedSampleSum60s] | None = None,
+) -> None:
+    """Insert reduced series into distributed_time_series_v4_reduced and 60s
+    buckets into the reduced samples tables. These tables exist only when
+    the schema migrator version includes the metrics cardinality-control
+    migration."""
+    if time_series:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_time_series_v4_reduced",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "description",
+                "unit",
+                "type",
+                "is_monotonic",
+                "fingerprint",
+                "unix_milli",
+                "labels",
+                "attrs",
+                "scope_attrs",
+                "resource_attrs",
+                "__normalized",
+            ],
+            data=[ts.to_row() for ts in time_series],
+        )
+
+    if last_samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_reduced_last_60s",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "reduced_fingerprint",
+                "unix_milli",
+                "sum_last",
+                "min",
+                "max",
+                "sum_values",
+                "count_series",
+                "count_samples",
+                "computed_at",
+            ],
+            data=[sample.to_row() for sample in last_samples],
+        )
+
+    if sum_samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_reduced_sum_60s",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "reduced_fingerprint",
+                "unix_milli",
+                "sum",
+                "count_series",
+                "count_samples",
+                "computed_at",
+            ],
+            data=[sample.to_row() for sample in sum_samples],
+        )
+
+
+def insert_buffer_metrics_to_clickhouse(
+    conn,
+    time_series: list[MetricsBufferTimeSeries],
+    samples: list[MetricsBufferSample],
+) -> None:
+    if time_series:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_time_series_v4_buffer",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "description",
+                "unit",
+                "type",
+                "is_monotonic",
+                "fingerprint",
+                "reduced_fingerprint",
+                "is_reduced",
+                "unix_milli",
+                "labels",
+                "attrs",
+                "scope_attrs",
+                "resource_attrs",
+                "__normalized",
+            ],
+            data=[ts.to_row() for ts in time_series],
+        )
+
+    if samples:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_samples_v4_buffer",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "fingerprint",
+                "reduced_fingerprint",
+                "is_monotonic",
+                "unix_milli",
+                "value",
+                "flags",
+            ],
+            data=[sample.to_row() for sample in samples],
+        )
+
+
+@pytest.fixture(name="insert_reduced_metrics", scope="function")
+def insert_reduced_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[..., None], Any]:
+    def _insert_reduced_metrics(
+        time_series: list[MetricsReducedTimeSeries],
+        last_samples: list[MetricsReducedSampleLast60s] | None = None,
+        sum_samples: list[MetricsReducedSampleSum60s] | None = None,
+    ) -> None:
+        insert_reduced_metrics_to_clickhouse(clickhouse.conn, time_series, last_samples, sum_samples)
+
+    yield _insert_reduced_metrics
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    for table in _REDUCED_METRICS_TABLES_TO_TRUNCATE:
+        clickhouse.conn.query(f"TRUNCATE TABLE signoz_metrics.{table} ON CLUSTER '{cluster}' SYNC")
+
+
+@pytest.fixture(name="insert_buffer_metrics", scope="function")
+def insert_buffer_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[..., None], Any]:
+    def _insert_buffer_metrics(
+        time_series: list[MetricsBufferTimeSeries],
+        samples: list[MetricsBufferSample],
+    ) -> None:
+        insert_buffer_metrics_to_clickhouse(clickhouse.conn, time_series, samples)
+
+    yield _insert_buffer_metrics
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    for table in _REDUCED_METRICS_TABLES_TO_TRUNCATE:
+        clickhouse.conn.query(f"TRUNCATE TABLE signoz_metrics.{table} ON CLUSTER '{cluster}' SYNC")
+
+
 @pytest.fixture(name="remove_metrics_ttl_and_storage_settings", scope="function")
 def remove_metrics_ttl_and_storage_settings(signoz: types.SigNoz):
     """

tests/fixtures/migrator.py

@@ -8,27 +8,30 @@ from fixtures.logger import setup_logger
 logger = setup_logger(__name__)
 
 
-def create_migrator(
+def create_migrator(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     network: Network,
     clickhouse: types.TestContainerClickhouse,
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
     cache_key: str = "migrator",
     env_overrides: dict | None = None,
+    version: str | None = None,
 ) -> types.Operation:
     """
     Factory function for running schema migrations.
-    Accepts optional env_overrides to customize the migrator environment.
+    Accepts optional env_overrides to customize the migrator environment, and
+    an optional version to pin a schema-migrator release different from the
+    --schema-migrator-version option.
     """
 
     def create() -> None:
-        version = request.config.getoption("--schema-migrator-version")
+        migrator_version = version or request.config.getoption("--schema-migrator-version")
         client = docker.from_env()
 
         environment = dict(env_overrides) if env_overrides else {}
 
         container = client.containers.run(
-            image=f"signoz/signoz-schema-migrator:{version}",
+            image=f"signoz/signoz-schema-migrator:{migrator_version}",
             command=f"sync --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN']}",
             detach=True,
             auto_remove=False,
@@ -47,7 +50,7 @@ def create_migrator(
         container.remove()
 
         container = client.containers.run(
-            image=f"signoz/signoz-schema-migrator:{version}",
+            image=f"signoz/signoz-schema-migrator:{migrator_version}",
             command=f"async --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN']}",
             detach=True,
             auto_remove=False,

tests/fixtures/querier.py

@@ -189,6 +189,35 @@ def make_query_request(
     )
 
 
+def aligned_epoch(ago: timedelta, step_seconds: int = DEFAULT_STEP_INTERVAL) -> int:
+    """Epoch seconds for `now - ago`, floored to a step boundary so seeded
+    points land exactly on the query's toStartOfInterval buckets."""
+    return (int((datetime.now(tz=UTC) - ago).timestamp()) // step_seconds) * step_seconds
+
+
+def query_metric_values(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    token: str,
+    metric_name: str,
+    start_epoch: int,
+    end_epoch: int,
+    time_agg: str,
+    space_agg: str,
+    step_interval: int = DEFAULT_STEP_INTERVAL,
+) -> list[dict]:
+    """Run a single metrics builder query over [start_epoch, end_epoch) in
+    epoch seconds and return its series values sorted by timestamp."""
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=start_epoch * 1000,
+        end_ms=end_epoch * 1000,
+        queries=[build_builder_query("A", metric_name, time_agg, space_agg, step_interval=step_interval)],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    return sorted(get_series_values(response.json(), "A"), key=lambda v: v["timestamp"])
+
+
 def build_builder_query(
     name: str,
     metric_name: str,

tests/fixtures/types.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Literal
 from urllib.parse import urljoin
 
@@ -84,11 +84,16 @@ class TestContainerClickhouse:
     container: TestContainerDocker
     conn: clickhouse_connect.driver.client.Client
     env: dict[str, str]
+    # Per-node containers when running a multi-node cluster. Empty for the
+    # default single-node setup; nodes[0] is the node `conn`/`env` point at
+    # (the initiator every query goes through).
+    nodes: list[TestContainerDocker] = field(default_factory=list)
 
     def __cache__(self) -> dict:
         return {
             "container": self.container.__cache__(),
             "env": self.env,
+            "nodes": [node.__cache__() for node in self.nodes],
         }
 
     def __log__(self) -> str:

tests/integration/tests/clickhousecluster/01_basic_cluster.py

@@ -0,0 +1,52 @@
+import clickhouse_connect.driver.client
+
+from fixtures import types
+
+TOTAL_ROWS = 64
+
+
+def test_topology(
+    clickhouse: types.TestContainerClickhouse,
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    aliases = {node.container_configs["9000"].address for node in clickhouse.nodes}
+
+    # Every node sees the same 2-shard cluster definition and identifies
+    # exactly itself as the local replica
+
+    for i, conn in enumerate(clickhouse_node_conns, start=1):
+        rows = conn.query("SELECT shard_num, host_name, is_local FROM system.clusters WHERE cluster = 'cluster' ORDER BY shard_num").result_rows
+        assert [row[0] for row in rows] == [1, 2], f"node {i}: expected 2 shards, got {rows}"
+        assert {row[1] for row in rows} == aliases, f"node {i}: cluster hosts {rows} != node aliases {aliases}"
+        local = [row[0] for row in rows if row[2]]
+        assert local == [i], f"node {i}: expected to be local for shard {i} only, got {local}"
+
+
+def test_replicated_distributed_round_trip(
+    clickhouse: types.TestContainerClickhouse,
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    # ON CLUSTER DDL reaches both nodes, Replicated engines register with the
+    # keeper via per-node macros, and a sharded Distributed insert scatters rows
+    # across shards while the distributed read returns the union.
+    conn = clickhouse.conn
+    try:
+        conn.query("CREATE DATABASE IF NOT EXISTS it_cluster ON CLUSTER 'cluster'")
+        conn.query("CREATE TABLE it_cluster.events ON CLUSTER 'cluster' (id UInt64, payload String) ENGINE = ReplicatedMergeTree ORDER BY id")
+        conn.query("CREATE TABLE it_cluster.distributed_events ON CLUSTER 'cluster' AS it_cluster.events ENGINE = Distributed('cluster', 'it_cluster', 'events', cityHash64(id))")
+
+        conn.insert(
+            database="it_cluster",
+            table="distributed_events",
+            column_names=["id", "payload"],
+            data=[[i, f"payload-{i:03d}"] for i in range(TOTAL_ROWS)],
+        )
+
+        distributed_count = int(conn.query("SELECT count() FROM it_cluster.distributed_events").result_rows[0][0])
+        assert distributed_count == TOTAL_ROWS
+
+        local_counts = [int(node_conn.query("SELECT count() FROM it_cluster.events").result_rows[0][0]) for node_conn in clickhouse_node_conns]
+        assert sum(local_counts) == TOTAL_ROWS, f"local counts {local_counts} do not add up to {TOTAL_ROWS}"
+        assert min(local_counts) > 0, f"all rows landed on one shard: {local_counts}"
+    finally:
+        conn.query("DROP DATABASE IF EXISTS it_cluster ON CLUSTER 'cluster' SYNC")

tests/integration/tests/clickhousecluster/conftest.py

@@ -0,0 +1,47 @@
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.clickhouse import create_clickhouse_cluster, create_clickhouse_keeper
+
+CLICKHOUSE_VERSION = "25.12.5"
+
+
+@pytest.fixture(name="keeper", scope="package")
+def keeper_cluster(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_clickhouse_keeper(
+        tmpfs=tmpfs,
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="keeper_cluster",
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse_cluster(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    return create_clickhouse_cluster(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=keeper,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="clickhouse_cluster",
+        shards=2,
+        version=CLICKHOUSE_VERSION,
+    )

tests/integration/tests/metricreduction/01_reduced_gauge.py

@@ -0,0 +1,203 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import clickhouse_connect.driver.client
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.clickhouse import assert_spans_shards
+from fixtures.metrics import (
+    Metrics,
+    MetricsReducedSampleLast60s,
+    MetricsReducedTimeSeries,
+)
+from fixtures.querier import aligned_epoch, query_metric_values
+
+
+def test_stitch_across_epoch(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+    insert_reduced_metrics: Callable[..., None],
+    clickhouse_node_conns: list[clickhouse_connect.driver.client.Client],
+) -> None:
+    """Before the rule activates, samples live in the raw tables; after, only
+    the reduced 60s tables have data. One query spanning the boundary must
+    stitch the two branches into a continuous series with no gap and no double
+    counting: 32 raw series at 2.0 collapse into 16 groups whose sum_last is
+    4.0, so the summed value stays 320 per step across the epoch. Enough
+    series to guarantee both shards hold data (guarded below), so the totals
+    also prove the raw and reduced joins execute shard-local."""
+    metric_name = "test_reduction_stitch"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+    services = [f"svc-{i:02d}" for i in range(16)]
+
+    # first 30 minutes: raw samples (2 pods per service, one sample per minute)
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": service, "pod": f"{service}-pod-{pod}"},
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                value=2.0,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for service in services
+            for pod in range(2)
+            for minute in range(30)
+        ]
+    )
+
+    # next 30 minutes: reduced 60s buckets (one group per service)
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch + 30 * 60, tz=UTC),
+        )
+        for service in services
+    ]
+    insert_reduced_metrics(
+        time_series,
+        [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + (30 + minute) * 60, tz=UTC),
+                sum_last=4.0,
+                min_value=2.0,
+                max_value=2.0,
+                sum_values=4.0,
+                count_series=2,
+                count_samples=2,
+            )
+            for ts in time_series
+            for minute in range(30)
+        ],
+    )
+
+    assert_spans_shards(clickhouse_node_conns, "time_series_v4", metric_name, total=len(services) * 2)
+    assert_spans_shards(clickhouse_node_conns, "time_series_v4_reduced", metric_name, total=len(services))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 3600, "sum", "sum", step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(12)]
+    assert [v["value"] for v in values] == [320.0] * 12
+
+
+@pytest.mark.parametrize(
+    "space_agg, expected",
+    [
+        ("sum", 12.0),  # sum_last: 4 + 8
+        ("avg", 3.0),  # sum(sum_last) / sum(count_series): 12 / 4
+        ("min", 1.0),  # min(min)
+        ("max", 6.0),  # max(max)
+    ],
+)
+def test_space_aggregations(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+    space_agg: str,
+    expected: float,
+) -> None:
+    """Space aggregations read the reduced pre-aggregated columns: sum/avg
+    from sum_last with the count_series weight, min/max from the min/max
+    columns."""
+    metric_name = f"test_reduction_space_{space_agg}"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    groups = [
+        # (service, sum_last, min, max, count_series)
+        ("a", 4.0, 1.0, 3.0, 2),
+        ("b", 8.0, 2.0, 6.0, 2),
+    ]
+    time_series = {
+        service: MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+        )
+        for service, _, _, _, _ in groups
+    }
+    insert_reduced_metrics(
+        list(time_series.values()),
+        [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=time_series[service].fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_last=sum_last,
+                min_value=min_value,
+                max_value=max_value,
+                sum_values=sum_last,
+                count_series=count_series,
+                count_samples=count_series,
+            )
+            for service, sum_last, min_value, max_value, count_series in groups
+            for minute in range(20)
+        ],
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 20 * 60, "avg", space_agg, step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [expected] * 4
+
+
+def test_dedup_latest_computed_at_wins(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+) -> None:
+    """The refreshable MVs re-emit every bucket on each refresh with a newer
+    computed_at (APPEND mode); reads must dedup to the latest version per
+    (series, bucket). Recompute the same buckets with a newer computed_at and
+    a different value: only the newer value may be counted."""
+    metric_name = "test_reduction_dedup"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+        )
+        for service in ("a", "b")
+    ]
+
+    def buckets(sum_last: float, computed_at_offset_seconds: int) -> list[MetricsReducedSampleLast60s]:
+        return [
+            MetricsReducedSampleLast60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_last=sum_last,
+                min_value=sum_last,
+                max_value=sum_last,
+                sum_values=sum_last,
+                count_series=1,
+                count_samples=1,
+                computed_at=datetime.fromtimestamp(base_epoch + minute * 60 + computed_at_offset_seconds, tz=UTC),
+            )
+            for ts in time_series
+            for minute in range(10)
+        ]
+
+    # first refresh emits 1.0; a later refresh recomputes the same buckets to 5.0
+    insert_reduced_metrics(time_series, buckets(sum_last=1.0, computed_at_offset_seconds=120))
+    insert_reduced_metrics(time_series, buckets(sum_last=5.0, computed_at_offset_seconds=180))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 10 * 60, "sum", "sum", step_interval=300)
+
+    # 2 groups x 5 buckets x 5.0 per step; 1.0 rows must not contribute
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(2)]
+    assert [v["value"] for v in values] == [50.0] * 2

tests/integration/tests/metricreduction/02_reduced_counter.py

@@ -0,0 +1,70 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import (
+    MetricsReducedSampleSum60s,
+    MetricsReducedTimeSeries,
+)
+from fixtures.querier import aligned_epoch, query_metric_values
+
+
+@pytest.mark.parametrize(
+    "time_agg, expected",
+    [
+        # 2 groups x 5 buckets x 30.0 per 300s step
+        ("rate", 1.0),  # 300 / 300s
+        ("increase", 300.0),
+    ],
+)
+def test_counter_rate_and_increase(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_reduced_metrics: Callable[..., None],
+    time_agg: str,
+    expected: float,
+) -> None:
+    metric_name = f"test_reduction_counter_{time_agg}"
+    base_epoch = aligned_epoch(timedelta(hours=30), step_seconds=300)
+
+    # monotonic cumulative counter: MetricsReducedTimeSeries mirrors the
+    # collector's temporality rewrite to Delta
+    time_series = [
+        MetricsReducedTimeSeries(
+            metric_name=metric_name,
+            kept_labels={"service": service},
+            timestamp=datetime.fromtimestamp(base_epoch, tz=UTC),
+            temporality="Cumulative",
+            type_="Sum",
+            is_monotonic=True,
+        )
+        for service in ("a", "b")
+    ]
+    assert all(ts.temporality == "Delta" for ts in time_series)
+
+    insert_reduced_metrics(
+        time_series,
+        sum_samples=[
+            MetricsReducedSampleSum60s(
+                metric_name=metric_name,
+                reduced_fingerprint=ts.fingerprint,
+                timestamp=datetime.fromtimestamp(base_epoch + minute * 60, tz=UTC),
+                sum_value=30.0,
+                count_series=2,
+                count_samples=2,
+                temporality="Delta",
+            )
+            for ts in time_series
+            for minute in range(20)
+        ],
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + 20 * 60, time_agg, "sum", step_interval=300)
+
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [expected] * 4

tests/integration/tests/metricreduction/03_reduced_buffer_window.py

@@ -0,0 +1,70 @@
+from collections.abc import Callable
+from datetime import timedelta
+from http import HTTPStatus
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metricreduction import build_ruled_gauge_buffer
+from fixtures.querier import (
+    aligned_epoch,
+    build_builder_query,
+    get_all_series,
+    index_series_by_label,
+    make_query_request,
+    query_metric_values,
+)
+
+SERVICES = ("a", "b")
+PODS_PER_SERVICE = 2
+MINUTES = 20
+
+
+def test_recent_window_reads_buffer_totals(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_buffer_metrics: Callable[..., None],
+) -> None:
+    metric_name = "test_reduction_buffer_totals"
+    # samples span [now-25m, now-5m); the query window sits inside the last 24h
+    base_epoch = aligned_epoch(timedelta(minutes=25), step_seconds=300)
+    insert_buffer_metrics(*build_ruled_gauge_buffer(metric_name, base_epoch, SERVICES, PODS_PER_SERVICE, MINUTES))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    values = query_metric_values(signoz, token, metric_name, base_epoch, base_epoch + MINUTES * 60, "sum", "sum", step_interval=300)
+
+    # 4 raw series x 5 samples x 1.0 per step: full raw resolution, and the
+    # is_reduced=true series rows must not join in (their fingerprints match
+    # no samples, and the ts CTE filters them out)
+    assert [v["timestamp"] for v in values] == [(base_epoch + step * 300) * 1000 for step in range(4)]
+    assert [v["value"] for v in values] == [float(len(SERVICES) * PODS_PER_SERVICE * 5)] * 4
+
+
+def test_recent_window_group_by_raw_label(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_buffer_metrics: Callable[..., None],
+) -> None:
+    """Group-by resolves against the raw buffer series rows (full labels), so
+    grouping by the kept label still sees every raw series underneath."""
+    metric_name = "test_reduction_buffer_groupby"
+    base_epoch = aligned_epoch(timedelta(minutes=25), step_seconds=300)
+    insert_buffer_metrics(*build_ruled_gauge_buffer(metric_name, base_epoch, SERVICES, PODS_PER_SERVICE, MINUTES))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=base_epoch * 1000,
+        end_ms=(base_epoch + MINUTES * 60) * 1000,
+        queries=[build_builder_query("A", metric_name, "sum", "sum", step_interval=300, group_by=["service"])],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    series_by_service = index_series_by_label(get_all_series(response.json(), "A"), "service")
+    assert set(series_by_service.keys()) == set(SERVICES)
+    for service in SERVICES:
+        values = sorted(series_by_service[service]["values"], key=lambda v: v["timestamp"])
+        # 2 pods x 5 samples x 1.0 per step
+        assert [v["value"] for v in values] == [float(PODS_PER_SERVICE * 5)] * 4

tests/integration/tests/metricreduction/conftest.py

@@ -0,0 +1,114 @@
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.auth import register_admin
+from fixtures.clickhouse import create_clickhouse_cluster, create_clickhouse_keeper
+from fixtures.http import ZEUS_NETWORK_ALIAS, create_zeus
+from fixtures.migrator import create_migrator
+from fixtures.signoz import create_signoz
+
+SCHEMA_MIGRATOR_VERSION = "v0.144.6-rc.2"
+CLICKHOUSE_VERSION = "25.12.5"
+
+
+@pytest.fixture(name="keeper", scope="package")
+def keeper_metricreduction(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_clickhouse_keeper(
+        tmpfs=tmpfs,
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="keeper_metricreduction",
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="zeus", scope="package")
+def zeus_metricreduction(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    return create_zeus(
+        network=network,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="zeus_metricreduction",
+        alias=ZEUS_NETWORK_ALIAS,
+    )
+
+
+@pytest.fixture(name="clickhouse", scope="package")
+def clickhouse_metricreduction(
+    tmpfs: Generator[types.LegacyPath, Any],
+    network: Network,
+    keeper: types.TestContainerDocker,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerClickhouse:
+    return create_clickhouse_cluster(
+        tmpfs=tmpfs,
+        network=network,
+        keeper=keeper,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="clickhouse_metricreduction",
+        shards=2,
+        version=CLICKHOUSE_VERSION,
+    )
+
+
+@pytest.fixture(name="migrator", scope="package")
+def migrator_metricreduction(
+    network: Network,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.Operation:
+    return create_migrator(
+        network=network,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="migrator_metricreduction",
+        version=SCHEMA_MIGRATOR_VERSION,
+    )
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_metricreduction(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz_metricreduction",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_ENABLE__METRICS__REDUCTION": True,
+        },
+    )
+
+
+@pytest.fixture(name="create_user_admin", scope="package")
+def create_user_admin_metricreduction(signoz: types.SigNoz, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.Operation:
+    return register_admin(signoz, request, pytestconfig, cache_key="create_user_admin_metricreduction")