fix: resolve aggregate column for exp histograms before samples table

pkg/statementbuilder/metricsstatementbuilder/statement_builder.go

@@ -428,20 +428,24 @@ func (b *StatementBuilder) buildTemporalAggDeltaFastPath(
 		sb.SelectMore(fmt.Sprintf("`%s`", GroupByColumnAlias(i, g.Name)))
 	}
 
-	aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(
-		samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
-	)
-	if err != nil {
-		return "", nil, err
-	}
-	if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
-		// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
-		aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
-	}
-
+	var aggCol string
 	if query.Aggregations[0].SpaceAggregation.IsPercentile() &&
 		query.Aggregations[0].Type == metrictypes.ExpHistogramType {
+		// merging sketches already spans every series in the step, so neither a
+		// samples-table value column nor the rate divisor applies
 		aggCol = fmt.Sprintf("quantilesDDMerge(0.01, %f)(sketch)[1]", query.Aggregations[0].SpaceAggregation.Percentile())
+	} else {
+		col, err := metricstelemetryschema.AggregationColumnForSamplesTable(
+			samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
+		)
+		if err != nil {
+			return "", nil, err
+		}
+		aggCol = col
+		if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
+			// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
+			aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
+		}
 	}
 
 	sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))

pkg/statementbuilder/metricsstatementbuilder/stmt_builder_test.go

@@ -126,6 +126,64 @@ func TestStatementBuilder(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			name:        "test_exp_histogram_percentile_delta",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "signoz_latency",
+						Type:             metrictypes.ExpHistogramType,
+						Temporality:      metrictypes.Delta,
+						SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
+					},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "service.name",
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
+				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
+			},
+			expectedErr: nil,
+		},
+		{
+			// the sketch merge spans the whole step, so `rate` must not add a /step divisor
+			name:        "test_exp_histogram_percentile_delta_rate_time_aggregation",
+			requestType: qbtypes.RequestTypeTimeSeries,
+			query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+				Signal:       telemetrytypes.SignalMetrics,
+				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+				Aggregations: []qbtypes.MetricAggregation{
+					{
+						MetricName:       "signoz_latency",
+						Type:             metrictypes.ExpHistogramType,
+						Temporality:      metrictypes.Delta,
+						TimeAggregation:  metrictypes.TimeAggregationRate,
+						SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
+					},
+				},
+				GroupBy: []qbtypes.GroupByKey{
+					{
+						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+							Name: "service.name",
+						},
+					},
+				},
+			},
+			expected: qbtypes.Statement{
+				Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
+				Args:  []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
+			},
+			expectedErr: nil,
+		},
 		{
 			name:        "test_histogram_percentile1",
 			requestType: qbtypes.RequestTypeTimeSeries,

tests/fixtures/metrics.py

@@ -132,7 +132,12 @@ class MetricsSample(ABC):
 
 
 class MetricsExpHist(ABC):
-    """Represents a row in the exp_hist table for exponential histograms."""
+    """Represents a row in the exp_hist table for exponential histograms.
+
+    Carries the raw observations rather than a serialized sketch: the `sketch`
+    column is an AggregateFunction state that only ClickHouse can build, so
+    `observations` is what gets folded into one on insert. Must be non-empty.
+    """
 
     env: str
     temporality: str
@@ -143,7 +148,7 @@ class MetricsExpHist(ABC):
     sum: np.float64
     min: np.float64
     max: np.float64
-    sketch: bytes
+    observations: list[int]
     flags: np.uint32
 
     def __init__(
@@ -151,11 +156,7 @@ class MetricsExpHist(ABC):
         metric_name: str,
         fingerprint: np.uint64,
         timestamp: datetime.datetime,
-        count: int,
-        sum_value: float,
-        min_value: float,
-        max_value: float,
-        sketch: bytes = b"",
+        observations: list[int],
         temporality: str = "Unspecified",
         env: str = "default",
         flags: int = 0,
@@ -165,28 +166,13 @@ class MetricsExpHist(ABC):
         self.metric_name = metric_name
         self.fingerprint = fingerprint
         self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
-        self.count = np.uint64(count)
-        self.sum = np.float64(sum_value)
-        self.min = np.float64(min_value)
-        self.max = np.float64(max_value)
-        self.sketch = sketch
+        self.observations = observations
+        self.count = np.uint64(len(observations))
+        self.sum = np.float64(sum(observations))
+        self.min = np.float64(min(observations))
+        self.max = np.float64(max(observations))
         self.flags = np.uint32(flags)
 
-    def to_row(self) -> list:
-        return [
-            self.env,
-            self.temporality,
-            self.metric_name,
-            self.fingerprint,
-            self.unix_milli,
-            self.count,
-            self.sum,
-            self.min,
-            self.max,
-            self.sketch,
-            self.flags,
-        ]
-
 
 class MetricsMetadata(ABC):
     """Represents a row in the metadata table for metric metadata."""
@@ -429,6 +415,73 @@ class Metrics(ABC):
         return metrics
 
 
+class ExpHistogramMetrics(ABC):
+    """High-level exponential histogram representation. Produces both time series
+    and exp_hist entries."""
+
+    metric_name: str
+    labels: dict[str, str]
+    temporality: str
+    timestamp: datetime.datetime
+    observations: list[int]
+
+    @property
+    def time_series(self) -> MetricsTimeSeries:
+        return self._time_series
+
+    @property
+    def exp_hist(self) -> MetricsExpHist:
+        return self._exp_hist
+
+    def __init__(
+        self,
+        metric_name: str,
+        observations: list[int],
+        labels: dict[str, str] = {},
+        timestamp: datetime.datetime | None = None,
+        temporality: str = "Delta",
+        flags: int = 0,
+        description: str = "",
+        unit: str = "",
+        env: str = "default",
+        resource_attributes: dict[str, str] = {},
+        scope_attributes: dict[str, str] = {},
+    ) -> None:
+        if timestamp is None:
+            timestamp = datetime.datetime.now()
+        self.metric_name = metric_name
+        self.labels = labels
+        self.temporality = temporality
+        self.timestamp = timestamp
+        self.observations = observations
+
+        self._time_series = MetricsTimeSeries(
+            metric_name=metric_name,
+            labels=labels,
+            timestamp=timestamp,
+            temporality=temporality,
+            description=description,
+            unit=unit,
+            # the querier resolves the metric type from this column, and only an
+            # ExponentialHistogram here routes the query to the sketch read
+            type_="ExponentialHistogram",
+            is_monotonic=False,
+            env=env,
+            resource_attrs=resource_attributes,
+            scope_attrs=scope_attributes,
+        )
+
+        self._exp_hist = MetricsExpHist(
+            metric_name=metric_name,
+            fingerprint=self._time_series.fingerprint,
+            timestamp=timestamp,
+            observations=observations,
+            temporality=temporality,
+            env=env,
+            flags=flags,
+        )
+
+
 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
@@ -853,6 +906,86 @@ def insert_metrics(
     )
 
 
+def insert_exp_histogram_metrics_to_clickhouse(conn, metrics: list[ExpHistogramMetrics]) -> None:
+    """
+    Insert exponential histograms into ClickHouse tables.
+    Handles insertion into:
+    - distributed_time_series_v4 (time series metadata)
+    - distributed_exp_hist (per-point sketches)
+    """
+    time_series_map: dict[tuple[int, int], MetricsTimeSeries] = {}
+    for metric in metrics:
+        fp = int(metric.time_series.fingerprint)
+        hour_bucket = int(metric.time_series.unix_milli) // 3_600_000
+        if (fp, hour_bucket) not in time_series_map:
+            metric.time_series.unix_milli = np.int64(hour_bucket * 3_600_000)
+            time_series_map[(fp, hour_bucket)] = metric.time_series
+
+    if len(time_series_map) > 0:
+        conn.insert(
+            database="signoz_metrics",
+            table="distributed_time_series_v4",
+            column_names=[
+                "env",
+                "temporality",
+                "metric_name",
+                "description",
+                "unit",
+                "type",
+                "is_monotonic",
+                "fingerprint",
+                "unix_milli",
+                "labels",
+                "attrs",
+                "scope_attrs",
+                "resource_attrs",
+            ],
+            data=[ts.to_row() for ts in time_series_map.values()],
+        )
+
+    # `sketch` is AggregateFunction(quantilesDD(...), UInt64) — the state has to be
+    # folded server-side, it cannot be sent as a literal. The quantilesDDState
+    # parameters must match the column's exactly or the INSERT is rejected.
+    for metric in metrics:
+        hist = metric.exp_hist
+        conn.command(
+            "INSERT INTO signoz_metrics.distributed_exp_hist "
+            "(env, temporality, metric_name, fingerprint, unix_milli, count, sum, min, max, sketch, flags) "
+            "SELECT %(env)s, %(temporality)s, %(metric_name)s, %(fingerprint)s, %(unix_milli)s, "
+            "%(count)s, %(sum)s, %(min)s, %(max)s, "
+            "quantilesDDState(0.01, 0.5, 0.75, 0.9, 0.95, 0.99)(toUInt64(observation)), %(flags)s "
+            "FROM (SELECT arrayJoin(%(observations)s) AS observation)",
+            parameters={
+                "env": hist.env,
+                "temporality": hist.temporality,
+                "metric_name": hist.metric_name,
+                "fingerprint": int(hist.fingerprint),
+                "unix_milli": int(hist.unix_milli),
+                "count": int(hist.count),
+                "sum": float(hist.sum),
+                "min": float(hist.min),
+                "max": float(hist.max),
+                "observations": hist.observations,
+                "flags": int(hist.flags),
+            },
+        )
+
+
+@pytest.fixture(name="insert_exp_histogram_metrics", scope="function")
+def insert_exp_histogram_metrics(
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[Callable[[list[ExpHistogramMetrics]], None], Any]:
+    def _insert_exp_histogram_metrics(metrics: list[ExpHistogramMetrics]) -> None:
+        insert_exp_histogram_metrics_to_clickhouse(clickhouse.conn, metrics)
+
+    yield _insert_exp_histogram_metrics
+
+    truncate_metrics_tables(
+        clickhouse.conn,
+        clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"],
+    )
+
+
 def insert_reduced_metrics_to_clickhouse(
     conn,
     time_series: list[MetricsReducedTimeSeries],

tests/integration/tests/queriermetrics/14_exp_histogram.py

@@ -0,0 +1,129 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import ExpHistogramMetrics
+from fixtures.querier import (
+    build_builder_query,
+    get_all_series,
+    get_series_values,
+    make_query_request,
+)
+
+# quantilesDD carries 0.01 relative accuracy and the log-spaced observations put
+# neighbouring ranks ~1.25% apart, so a percentile can land a few percent off
+PERCENTILE_TOLERANCE = 0.05
+
+
+@pytest.mark.parametrize(
+    "space_aggregation, frontend_first, frontend_last, backend_first, backend_last",
+    [
+        ("p50", 118, 153, 711, 921),
+        ("p95", 1108, 1435, 6651, 8613),
+        ("p99", 1352, 1751, 8113, 10507),
+    ],
+)
+@pytest.mark.parametrize("time_aggregation", ["", "rate"])
+def test_exp_histogram_percentile_delta_grouped(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
+    time_aggregation: str,
+    space_aggregation: str,
+    frontend_first: float,
+    frontend_last: float,
+    backend_first: float,
+    backend_last: float,
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    metric_name = "test_exp_histogram_latency"
+
+    insert_exp_histogram_metrics(
+        [
+            ExpHistogramMetrics(
+                metric_name=metric_name,
+                # log-spaced latencies with a long tail, drifting ~30% higher across
+                # the hour so each point carries a distinct distribution
+                observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
+                labels={"service.name": service},
+                timestamp=now - timedelta(minutes=60 - minute),
+                temporality="Delta",
+            )
+            for service, base in (("frontend", 10), ("backend", 60))
+            for minute in range(60)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query(
+        "A",
+        metric_name,
+        time_aggregation,
+        space_aggregation,
+        temporality="delta",
+        group_by=["service.name"],
+    )
+
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    all_series = get_all_series(response.json(), "A")
+    values_by_service = {series["labels"][0]["value"]: [point["value"] for point in sorted(series["values"], key=lambda point: point["timestamp"])] for series in all_series}
+    assert set(values_by_service.keys()) == {"frontend", "backend"}, f"got series {set(values_by_service.keys())}"
+
+    for service, first, last in (
+        ("frontend", frontend_first, frontend_last),
+        ("backend", backend_first, backend_last),
+    ):
+        values = values_by_service[service]
+        assert len(values) >= 55, f"{service}: expected a point per minute, got {len(values)}"
+        assert values[0] == pytest.approx(first, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the oldest point: got {values[0]}, want ~{first}"
+        assert values[-1] == pytest.approx(last, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the newest point: got {values[-1]}, want ~{last}"
+        # every observation drifts up minute over minute, so the sketch must too
+        assert values == sorted(values), f"{service} {space_aggregation} is not non-decreasing: {values}"
+
+
+def test_exp_histogram_percentile_delta_merges_across_series(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
+) -> None:
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
+    end_ms = int(now.timestamp() * 1000)
+    metric_name = "test_exp_histogram_latency_merged"
+
+    insert_exp_histogram_metrics(
+        [
+            ExpHistogramMetrics(
+                metric_name=metric_name,
+                observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
+                labels={"service.name": service},
+                timestamp=now - timedelta(minutes=60 - minute),
+                temporality="Delta",
+            )
+            for service, base in (("frontend", 10), ("backend", 60))
+            for minute in range(60)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query("A", metric_name, "", "p95", temporality="delta")
+
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    # both services' sketches merge into one, so p95 sits well above the frontend's
+    # own p95 (~1108) and below the backend's (~6651)
+    values = [point["value"] for point in sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])]
+    assert len(values) >= 55, f"expected a point per minute, got {len(values)}"
+    assert values[0] == pytest.approx(5188, rel=PERCENTILE_TOLERANCE), f"oldest point: got {values[0]}, want ~5188"
+    assert values[-1] == pytest.approx(6718, rel=PERCENTILE_TOLERANCE), f"newest point: got {values[-1]}, want ~6718"