fix(metrics): use local table for fingerprint ctes (#11931)

* fix(metrics): use local table for fingerprint ctes

* chore: remove queries file

* chore: update reduced_test.go

docs/api/openapi.yml

@@ -3571,7 +3571,7 @@ components:
       - user
       - system
       - integration
-      type: object
+      type: string
     DashboardtypesSpanGaps:
       properties:
         fillLessThan:

pkg/modules/metricsexplorer/implmetricsexplorer/module.go

@@ -1100,24 +1100,25 @@ func (m *module) fetchMetricsStatsWithSamples(
 		reducedSumSB.Where(reducedSumSB.Between("unix_milli", req.Start, req.End))
 		reducedSumSB.Where("NOT startsWith(metric_name, 'signoz')")
 
-		if filterWhereClause == nil {
-			reducedTsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
-			reducedTsSB.Where(reducedTsSB.Between("unix_milli", start, end))
-			reducedTsSB.Where("NOT startsWith(metric_name, 'signoz')")
-			reducedTsSB.GroupBy("metric_name")
-		} else {
+		// separate query for reduced series counts
+		reducedTsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
+		reducedTsSB.Where(reducedTsSB.Between("unix_milli", start, end))
+		reducedTsSB.Where("NOT startsWith(metric_name, 'signoz')")
+		reducedTsSB.GroupBy("metric_name")
+
+		if filterWhereClause != nil {
+			reducedTsSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
+
+			// samples uses a separate cte with local table
 			reducedFpSB := sqlbuilder.NewSelectBuilder()
-			reducedFpSB.Select("metric_name", "fingerprint")
-			reducedFpSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
+			reducedFpSB.Select("fingerprint")
+			reducedFpSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedLocalTableName))
 			reducedFpSB.Where(reducedFpSB.Between("unix_milli", start, end))
 			reducedFpSB.Where("NOT startsWith(metric_name, 'signoz')")
 			reducedFpSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
-			reducedFpSB.GroupBy("metric_name", "fingerprint")
+			reducedFpSB.GroupBy("fingerprint")
 			ctes = append(ctes, sqlbuilder.CTEQuery("__reduced_filtered_fingerprints").As(reducedFpSB))
 
-			reducedTsSB.From("__reduced_filtered_fingerprints")
-			reducedTsSB.GroupBy("metric_name")
-
 			reducedLastSB.Where("reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints)")
 			reducedSumSB.Where("reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints)")
 		}
@@ -1422,7 +1423,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
 		if reductionEnabled {
 			reducedFingerprintSB := sqlbuilder.NewSelectBuilder()
 			reducedFingerprintSB.Select("fingerprint")
-			reducedFingerprintSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
+			reducedFingerprintSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedLocalTableName))
 			reducedFingerprintSB.Where(reducedFingerprintSB.Between("unix_milli", start, end))
 			reducedFingerprintSB.Where("NOT startsWith(metric_name, 'signoz')")
 			reducedFingerprintSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))

pkg/modules/metricsexplorer/implmetricsexplorer/module_test.go

@@ -32,11 +32,11 @@ const (
 	testEndMillis                  int64 = 1700003600000 // +1h
 	statsNoFilterSQL                     = "WITH __time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND metric_name IN (SELECT DISTINCT metric_name FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz')) GROUP BY metric_name), __reduced_time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name) SELECT COALESCE(ts.metric_name, rts.metric_name, s.metric_name, rs.metric_name) AS metric_name, COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) AS timeseries, COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) AS samples, COUNT(*) OVER() AS total FROM __time_series_counts ts FULL OUTER JOIN __reduced_time_series_counts rts ON ts.metric_name = rts.metric_name FULL OUTER JOIN __sample_counts s ON COALESCE(ts.metric_name, rts.metric_name) = s.metric_name FULL OUTER JOIN __reduced_sample_counts rs ON COALESCE(ts.metric_name, rts.metric_name, s.metric_name) = rs.metric_name WHERE (COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) > 0 OR COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) > 0) ORDER BY samples DESC, metric_name ASC LIMIT ? OFFSET ? SETTINGS join_use_nulls = 1"
 	statsOrderTimeseriesSQL              = "WITH __time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND metric_name IN (SELECT DISTINCT metric_name FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz')) GROUP BY metric_name), __reduced_time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name) SELECT COALESCE(ts.metric_name, rts.metric_name, s.metric_name, rs.metric_name) AS metric_name, COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) AS timeseries, COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) AS samples, COUNT(*) OVER() AS total FROM __time_series_counts ts FULL OUTER JOIN __reduced_time_series_counts rts ON ts.metric_name = rts.metric_name FULL OUTER JOIN __sample_counts s ON COALESCE(ts.metric_name, rts.metric_name) = s.metric_name FULL OUTER JOIN __reduced_sample_counts rs ON COALESCE(ts.metric_name, rts.metric_name, s.metric_name) = rs.metric_name WHERE (COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) > 0 OR COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) > 0) ORDER BY timeseries ASC, metric_name ASC LIMIT ? OFFSET ? SETTINGS join_use_nulls = 1"
-	statsWithFilterSQL                   = "WITH __time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name), __filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND fingerprint IN (SELECT fingerprint FROM __filtered_fingerprints) GROUP BY metric_name), __reduced_filtered_fingerprints AS (SELECT metric_name, fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name, fingerprint), __reduced_time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM __reduced_filtered_fingerprints GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name) SELECT COALESCE(ts.metric_name, rts.metric_name, s.metric_name, rs.metric_name) AS metric_name, COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) AS timeseries, COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) AS samples, COUNT(*) OVER() AS total FROM __time_series_counts ts FULL OUTER JOIN __reduced_time_series_counts rts ON ts.metric_name = rts.metric_name FULL OUTER JOIN __sample_counts s ON COALESCE(ts.metric_name, rts.metric_name) = s.metric_name FULL OUTER JOIN __reduced_sample_counts rs ON COALESCE(ts.metric_name, rts.metric_name, s.metric_name) = rs.metric_name WHERE (COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) > 0 OR COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) > 0) ORDER BY samples DESC, metric_name ASC LIMIT ? OFFSET ? SETTINGS join_use_nulls = 1"
+	statsWithFilterSQL                   = "WITH __time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name), __filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND fingerprint IN (SELECT fingerprint FROM __filtered_fingerprints) GROUP BY metric_name), __reduced_filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint), __reduced_time_series_counts AS (SELECT metric_name, uniq(fingerprint) AS timeseries FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name) SELECT COALESCE(ts.metric_name, rts.metric_name, s.metric_name, rs.metric_name) AS metric_name, COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) AS timeseries, COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) AS samples, COUNT(*) OVER() AS total FROM __time_series_counts ts FULL OUTER JOIN __reduced_time_series_counts rts ON ts.metric_name = rts.metric_name FULL OUTER JOIN __sample_counts s ON COALESCE(ts.metric_name, rts.metric_name) = s.metric_name FULL OUTER JOIN __reduced_sample_counts rs ON COALESCE(ts.metric_name, rts.metric_name, s.metric_name) = rs.metric_name WHERE (COALESCE(ts.timeseries, 0) + COALESCE(rts.timeseries, 0) > 0 OR COALESCE(s.samples, 0) + COALESCE(rs.samples, 0) > 0) ORDER BY samples DESC, metric_name ASC LIMIT ? OFFSET ? SETTINGS join_use_nulls = 1"
 	treemapTimeseriesNoFilterSQL         = "WITH __total_time_series AS (SELECT uniq(fingerprint) AS total_time_series FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ?), __metric_totals AS (SELECT metric_name, uniq(fingerprint) AS total_value FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name), __reduced_total_time_series AS (SELECT uniq(fingerprint) AS total_time_series FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ?), __reduced_metric_totals AS (SELECT metric_name, uniq(fingerprint) AS total_value FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') GROUP BY metric_name) SELECT COALESCE(mt.metric_name, rmt.metric_name) AS metric_name, COALESCE(mt.total_value, 0) + COALESCE(rmt.total_value, 0) AS total_value, CASE WHEN (tts.total_time_series + rtts.total_time_series) = 0 THEN 0 ELSE ((COALESCE(mt.total_value, 0) + COALESCE(rmt.total_value, 0)) * 100.0 / (tts.total_time_series + rtts.total_time_series)) END AS percentage FROM __metric_totals mt FULL OUTER JOIN __reduced_metric_totals rmt ON mt.metric_name = rmt.metric_name JOIN __total_time_series tts ON 1=1 JOIN __reduced_total_time_series rtts ON 1=1 ORDER BY percentage DESC LIMIT ? SETTINGS join_use_nulls = 1"
 	treemapTimeseriesWithFilterSQL       = "WITH __total_time_series AS (SELECT uniq(fingerprint) AS total_time_series FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ?), __metric_totals AS (SELECT metric_name, uniq(fingerprint) AS total_value FROM signoz_metrics.distributed_time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name), __reduced_total_time_series AS (SELECT uniq(fingerprint) AS total_time_series FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ?), __reduced_metric_totals AS (SELECT metric_name, uniq(fingerprint) AS total_value FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name) SELECT COALESCE(mt.metric_name, rmt.metric_name) AS metric_name, COALESCE(mt.total_value, 0) + COALESCE(rmt.total_value, 0) AS total_value, CASE WHEN (tts.total_time_series + rtts.total_time_series) = 0 THEN 0 ELSE ((COALESCE(mt.total_value, 0) + COALESCE(rmt.total_value, 0)) * 100.0 / (tts.total_time_series + rtts.total_time_series)) END AS percentage FROM __metric_totals mt FULL OUTER JOIN __reduced_metric_totals rmt ON mt.metric_name = rmt.metric_name JOIN __total_time_series tts ON 1=1 JOIN __reduced_total_time_series rtts ON 1=1 ORDER BY percentage DESC LIMIT ? SETTINGS join_use_nulls = 1"
 	treemapSamplesNoFilterSQL            = "WITH __metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4 WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __reduced_metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4_reduced WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates) GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name), __total_samples AS (SELECT count(*) AS total_samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ?), __reduced_total_samples AS (SELECT sum(cnt) AS total_samples FROM ((SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ?) UNION ALL (SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ?)) AS reduced_total), __all_candidates AS (SELECT DISTINCT metric_name FROM ((SELECT metric_name FROM __metric_candidates) UNION ALL (SELECT metric_name FROM __reduced_metric_candidates)) AS candidates) SELECT ac.metric_name, COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0) AS samples, CASE WHEN (ts.total_samples + rts.total_samples) = 0 THEN 0 ELSE ((COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0)) * 100.0 / (ts.total_samples + rts.total_samples)) END AS percentage FROM __all_candidates ac LEFT JOIN __sample_counts sc ON ac.metric_name = sc.metric_name LEFT JOIN __reduced_sample_counts rsc ON ac.metric_name = rsc.metric_name JOIN __total_samples ts ON 1=1 JOIN __reduced_total_samples rts ON 1=1 ORDER BY percentage DESC LIMIT ? SETTINGS join_use_nulls = 1"
-	treemapSamplesWithFilterSQL          = "WITH __metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4 WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __reduced_metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4_reduced WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? AND metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates) GROUP BY fingerprint), __reduced_filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) GROUP BY fingerprint), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates) AND fingerprint IN (SELECT fingerprint FROM __filtered_fingerprints) GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name), __total_samples AS (SELECT count(*) AS total_samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ?), __reduced_total_samples AS (SELECT sum(cnt) AS total_samples FROM ((SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ?) UNION ALL (SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ?)) AS reduced_total), __all_candidates AS (SELECT DISTINCT metric_name FROM ((SELECT metric_name FROM __metric_candidates) UNION ALL (SELECT metric_name FROM __reduced_metric_candidates)) AS candidates) SELECT ac.metric_name, COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0) AS samples, CASE WHEN (ts.total_samples + rts.total_samples) = 0 THEN 0 ELSE ((COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0)) * 100.0 / (ts.total_samples + rts.total_samples)) END AS percentage FROM __all_candidates ac LEFT JOIN __sample_counts sc ON ac.metric_name = sc.metric_name LEFT JOIN __reduced_sample_counts rsc ON ac.metric_name = rsc.metric_name JOIN __total_samples ts ON 1=1 JOIN __reduced_total_samples rts ON 1=1 ORDER BY percentage DESC LIMIT ? SETTINGS join_use_nulls = 1"
+	treemapSamplesWithFilterSQL          = "WITH __metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4 WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __reduced_metric_candidates AS (SELECT metric_name FROM signoz_metrics.distributed_time_series_v4_reduced WHERE NOT startsWith(metric_name, 'signoz') AND unix_milli BETWEEN ? AND ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY metric_name ORDER BY uniq(fingerprint) DESC LIMIT ?), __filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4 WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? AND metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates) GROUP BY fingerprint), __reduced_filtered_fingerprints AS (SELECT fingerprint FROM signoz_metrics.time_series_v4_reduced WHERE unix_milli BETWEEN ? AND ? AND NOT startsWith(metric_name, 'signoz') AND JSONExtractString(labels, 'host.name') = ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) GROUP BY fingerprint), __sample_counts AS (SELECT metric_name, count(*) AS samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates) AND fingerprint IN (SELECT fingerprint FROM __filtered_fingerprints) GROUP BY metric_name), __reduced_sample_counts AS (SELECT metric_name, sum(cnt) AS samples FROM ((SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name) UNION ALL (SELECT metric_name, uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ? AND metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates) AND reduced_fingerprint IN (SELECT fingerprint FROM __reduced_filtered_fingerprints) GROUP BY metric_name)) AS reduced_samples GROUP BY metric_name), __total_samples AS (SELECT count(*) AS total_samples FROM signoz_metrics.distributed_samples_v4 WHERE unix_milli BETWEEN ? AND ?), __reduced_total_samples AS (SELECT sum(cnt) AS total_samples FROM ((SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE unix_milli BETWEEN ? AND ?) UNION ALL (SELECT uniq(reduced_fingerprint, unix_milli) AS cnt FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE unix_milli BETWEEN ? AND ?)) AS reduced_total), __all_candidates AS (SELECT DISTINCT metric_name FROM ((SELECT metric_name FROM __metric_candidates) UNION ALL (SELECT metric_name FROM __reduced_metric_candidates)) AS candidates) SELECT ac.metric_name, COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0) AS samples, CASE WHEN (ts.total_samples + rts.total_samples) = 0 THEN 0 ELSE ((COALESCE(sc.samples, 0) + COALESCE(rsc.samples, 0)) * 100.0 / (ts.total_samples + rts.total_samples)) END AS percentage FROM __all_candidates ac LEFT JOIN __sample_counts sc ON ac.metric_name = sc.metric_name LEFT JOIN __reduced_sample_counts rsc ON ac.metric_name = rsc.metric_name JOIN __total_samples ts ON 1=1 JOIN __reduced_total_samples rts ON 1=1 ORDER BY percentage DESC LIMIT ? SETTINGS join_use_nulls = 1"
 
 	// Raw-only SQL produced when the metrics-reduction feature flag is OFF. These
 	// must stay byte-for-byte identical to the pre-reduction queries.
@@ -193,7 +193,7 @@ func TestGetStats(t *testing.T) {
 	}{
 		{name: "NoFilter_FastPathSQL", expectSQL: statsNoFilterSQL, argCount: 14, reductionEnabled: true},
 		{name: "WhitespaceFilter_FastPathSQL", opts: []statsOpt{withStatsFilter("   ")}, expectSQL: statsNoFilterSQL, argCount: 14, reductionEnabled: true},
-		{name: "WithFilter_FingerprintSQL", opts: []statsOpt{withStatsFilter("host.name = 'foo'")}, seedKey: "host.name", expectSQL: statsWithFilterSQL, argCount: 17, reductionEnabled: true},
+		{name: "WithFilter_FingerprintSQL", opts: []statsOpt{withStatsFilter("host.name = 'foo'")}, seedKey: "host.name", expectSQL: statsWithFilterSQL, argCount: 20, reductionEnabled: true},
 		{name: "OrderByTimeseriesAsc", opts: []statsOpt{withStatsOrderBy("timeseries", qbtypes.OrderDirectionAsc)}, expectSQL: statsOrderTimeseriesSQL, argCount: 14, reductionEnabled: true},
 		{name: "OrderByInvalid", opts: []statsOpt{withStatsOrderBy("nonsense", qbtypes.OrderDirectionAsc)}, noQuery: true, wantCode: errors.CodeInvalidInput},
 		{name: "QueryError", queryErr: assert.AnError, expectSQL: statsNoFilterSQL, argCount: 14, reductionEnabled: true, wantCode: errors.CodeInternal},

pkg/telemetrymetrics/reduced_test.go

@@ -39,7 +39,7 @@ 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.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -47,7 +47,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "gauge_avg_avg",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -55,7 +55,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "gauge_min_min",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMin, metrictypes.SpaceAggregationMin),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(min) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, min(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`min`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, min(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -63,7 +63,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "gauge_max_max",
 			query: reducedQuery("test.metric", metrictypes.GaugeType, metrictypes.Unspecified, metrictypes.TimeAggregationMax, metrictypes.SpaceAggregationMax),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(value) AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`max`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, max(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -71,7 +71,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "counter_sum_rate",
 			query: reducedQuery("test.metric.sum", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationRate, metrictypes.SpaceAggregationSum),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) / (ts - lagInFrame(ts, 1) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) / 300 AS per_series_value FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -79,7 +79,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "counter_avg_increase",
 			query: reducedQuery("test.metric", metrictypes.SumType, metrictypes.Cumulative, metrictypes.TimeAggregationIncrease, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT ts, multiIf(row_number() OVER rate_window = 1, nan, (per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) < 0, per_series_value, per_series_value - lagInFrame(per_series_value, 1) OVER rate_window) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, max(max) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_sum_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -103,7 +103,7 @@ 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.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint, `le`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts, `le`), __spatial_aggregation_cte AS (SELECT ts, `le`, sum(per_series_value) AS value FROM __temporal_aggregation_cte GROUP BY ts, `le`) SELECT ts, histogramQuantile(arrayMap(x -> toFloat64(x), groupArray(le)), groupArray(value), 0.990) AS value FROM __spatial_aggregation_cte GROUP BY ts ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},
@@ -111,7 +111,7 @@ func TestReducedStatementBuilder(t *testing.T) {
 			name:  "summary_avg",
 			query: reducedQuery("test.metric", metrictypes.SummaryType, metrictypes.Unspecified, metrictypes.TimeAggregationAvg, metrictypes.SpaceAggregationAvg),
 			expected: qbtypes.Statement{
-				Query: "SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, sum(sum) / sum(count) AS per_series_value FROM signoz_metrics.distributed_samples_v4_agg_5m AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.time_series_v4_1day WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) UNION ALL SELECT * FROM (WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(300)) AS ts, avg(value) AS per_series_value, avg(weight) AS per_series_weight FROM (SELECT reduced_fingerprint AS fingerprint, unix_milli, argMax(`sum_last`, computed_at) AS value, argMax(`count_series`, computed_at) AS weight FROM signoz_metrics.distributed_samples_v4_reduced_last_60s WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY reduced_fingerprint, unix_milli) AS points INNER JOIN (SELECT fingerprint FROM signoz_metrics.distributed_time_series_v4_reduced WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND __normalized = ? GROUP BY fingerprint) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint GROUP BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, sum(per_series_value) / sum(per_series_weight) AS value FROM __temporal_aggregation_cte GROUP BY ts) SELECT * FROM __spatial_aggregation_cte ORDER BY ts) ORDER BY ts",
+				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},
 			},
 		},

pkg/telemetrymetrics/statement_builder.go

@@ -296,7 +296,7 @@ func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
 		}
 	}
 
-	sb.From(fmt.Sprintf("%s.%s", DBName, TimeseriesV4ReducedTableName))
+	sb.From(fmt.Sprintf("%s.%s", DBName, TimeseriesV4ReducedLocalTableName))
 	sb.Select("fingerprint")
 	for _, g := range query.GroupBy {
 		col, err := b.fm.ColumnExpressionFor(ctx, start, end, &g.TelemetryFieldKey, keys)

pkg/types/dashboardtypes/source.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/swaggest/jsonschema-go"
 )
 
 type Source struct {
@@ -22,6 +23,23 @@ func (Source) Enum() []any {
 	return []any{SourceUser, SourceSystem, SourceIntegration}
 }
 
+// JSONSchema exposes Source as a string enum. Without this the reflector sees the
+// unexported valuer.String field and emits `type: object`. The enum values are
+// derived from Enum() so the list of sources lives in exactly one place.
+func (Source) JSONSchema() (jsonschema.Schema, error) {
+	sources := Source{}.Enum()
+	enum := make([]any, 0, len(sources))
+	for _, source := range sources {
+		enum = append(enum, source.(Source).StringValue())
+	}
+
+	schema := jsonschema.Schema{}
+	schema.WithType(jsonschema.String.Type())
+	schema.WithEnum(enum...)
+
+	return schema, nil
+}
+
 func (s Source) IsValid() bool {
 	return slices.ContainsFunc(s.Enum(), func(v any) bool { return v == s })
 }