feat(clickhouse,tsql): queue metrics rollups and single-scan ranking

ericallam · ericallam · commit d60696c2ea89 · 2026-07-04T23:14:53.000+01:00
Adds two rollups fed from the raw landing table: a per-queue 5-minute
tier and an environment-level 1-minute tier (gauges plus TDigest wait
quantiles). Ranking now reads the 5m tier and returns the page and the
ranked total in one windowed query instead of two scans.

The 5m materialized view reads raw rather than cascading off the 10s
table: deltaSumTimestamp states hold a single first/last segment, so
merging states in an MV's hash-ordered GROUP BY double-counts bridging
spans. For the same reason the env tier carries no counter columns, and
env-wide counter totals must group by queue before summing.
diff --git a/internal-packages/clickhouse/schema/035_create_queue_metrics_v1.sql b/internal-packages/clickhouse/schema/035_create_queue_metrics_v1.sql
@@ -92,7 +92,113 @@ SELECT
 FROM trigger_dev.queue_metrics_raw_v1
 GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start;
 
+-- (4) Env-level 1m rollup (no queue dimension) for header tiles/saturation charts.
+-- No counter deltas on purpose: cross-queue deltaSumTimestamp state merges mix unrelated
+-- odometers (env totals must GROUP BY queue then sum). TDigest because an env-level
+-- reservoir absorbs every sample in the environment.
+CREATE TABLE IF NOT EXISTS trigger_dev.env_metrics_1m_v1
+(
+  organization_id  LowCardinality(String),
+  project_id       LowCardinality(String),
+  environment_id   String CODEC(ZSTD(1)),
+  bucket_start     DateTime CODEC(Delta(4), ZSTD(1)),
+
+  max_env_queued   SimpleAggregateFunction(max, UInt32),
+  max_env_running  SimpleAggregateFunction(max, UInt32),
+  max_env_limit    SimpleAggregateFunction(max, UInt32),
+  throttled_count  SimpleAggregateFunction(sum, UInt64),
+
+  wait_ms_sum      SimpleAggregateFunction(sum, UInt64),
+  wait_ms_count    SimpleAggregateFunction(sum, UInt64),
+  wait_quantiles   AggregateFunction(quantilesTDigest(0.5, 0.9, 0.95, 0.99), UInt32)
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY toDate(bucket_start)
+ORDER BY (organization_id, project_id, environment_id, bucket_start)
+TTL bucket_start + INTERVAL 30 DAY
+SETTINGS ttl_only_drop_parts = 1;
+
+-- (5) MV: raw -> env rollup.
+CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.env_metrics_1m_mv_v1
+TO trigger_dev.env_metrics_1m_v1 AS
+SELECT
+  organization_id, project_id, environment_id,
+  toStartOfInterval(event_time, INTERVAL 1 MINUTE) AS bucket_start,
+  max(env_queued)         AS max_env_queued,
+  max(env_running)        AS max_env_running,
+  max(env_limit)          AS max_env_limit,
+  sum(throttled)          AS throttled_count,
+  sumIf(wait_ms, op = 'started')                 AS wait_ms_sum,
+  countIf(op = 'started' AND wait_ms > 0)        AS wait_ms_count,
+  quantilesTDigestStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0) AS wait_quantiles
+FROM trigger_dev.queue_metrics_raw_v1
+GROUP BY organization_id, project_id, environment_id, bucket_start;
+
+-- (6) Per-queue 5m rollup, exact column mirror of queue_metrics_v1, for ranking and
+-- env-wide GROUP BY queue reads at long ranges.
+CREATE TABLE IF NOT EXISTS trigger_dev.queue_metrics_5m_v1
+(
+  organization_id  LowCardinality(String),
+  project_id       LowCardinality(String),
+  environment_id   String CODEC(ZSTD(1)),
+  queue_name       String CODEC(ZSTD(1)),
+  bucket_start     DateTime CODEC(Delta(4), ZSTD(1)),
+
+  enqueue_delta    AggregateFunction(deltaSumTimestamp, UInt64, UInt64),
+  started_delta    AggregateFunction(deltaSumTimestamp, UInt64, UInt64),
+  ack_delta        AggregateFunction(deltaSumTimestamp, UInt64, UInt64),
+  nack_delta       AggregateFunction(deltaSumTimestamp, UInt64, UInt64),
+  dlq_delta        AggregateFunction(deltaSumTimestamp, UInt64, UInt64),
+  throttled_count  SimpleAggregateFunction(sum, UInt64),
+
+  max_queued       SimpleAggregateFunction(max, UInt32),
+  max_running      SimpleAggregateFunction(max, UInt32),
+  max_limit        SimpleAggregateFunction(max, UInt32),
+  max_env_queued   SimpleAggregateFunction(max, UInt32),
+  max_env_running  SimpleAggregateFunction(max, UInt32),
+  max_env_limit    SimpleAggregateFunction(max, UInt32),
+
+  wait_ms_sum      SimpleAggregateFunction(sum, UInt64),
+  wait_ms_count    SimpleAggregateFunction(sum, UInt64),
+  wait_quantiles   AggregateFunction(quantiles(0.5, 0.9, 0.95, 0.99), UInt32)
+)
+ENGINE = AggregatingMergeTree()
+PARTITION BY toDate(bucket_start)
+ORDER BY (organization_id, project_id, environment_id, queue_name, bucket_start)
+TTL bucket_start + INTERVAL 30 DAY
+SETTINGS ttl_only_drop_parts = 1;
+
+-- (7) MV: raw -> 5m rollup. MUST read raw, never cascade off queue_metrics_v1 with
+-- -MergeState: MV GROUP BY merges states in hash order, and out-of-time-order
+-- deltaSumTimestamp merges double-count bridging spans (verified 3x inflation).
+CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_5m_mv_v1
+TO trigger_dev.queue_metrics_5m_v1 AS
+SELECT
+  organization_id, project_id, environment_id, queue_name,
+  toStartOfInterval(event_time, INTERVAL 5 MINUTE) AS bucket_start,
+  deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue') AS enqueue_delta,
+  deltaSumTimestampStateIf(cumulative, order_key, op = 'started') AS started_delta,
+  deltaSumTimestampStateIf(cumulative, order_key, op = 'ack')     AS ack_delta,
+  deltaSumTimestampStateIf(cumulative, order_key, op = 'nack')    AS nack_delta,
+  deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq')     AS dlq_delta,
+  sum(throttled)          AS throttled_count,
+  max(queued)             AS max_queued,
+  max(running)            AS max_running,
+  max(queue_limit)        AS max_limit,
+  max(env_queued)         AS max_env_queued,
+  max(env_running)        AS max_env_running,
+  max(env_limit)          AS max_env_limit,
+  sumIf(wait_ms, op = 'started')                 AS wait_ms_sum,
+  countIf(op = 'started' AND wait_ms > 0)        AS wait_ms_count,
+  quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0) AS wait_quantiles
+FROM trigger_dev.queue_metrics_raw_v1
+GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start;
+
 -- +goose Down
+DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1;
+DROP TABLE IF EXISTS trigger_dev.queue_metrics_5m_v1;
+DROP VIEW IF EXISTS trigger_dev.env_metrics_1m_mv_v1;
+DROP TABLE IF EXISTS trigger_dev.env_metrics_1m_v1;
 DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1;
 DROP TABLE IF EXISTS trigger_dev.queue_metrics_v1;
 DROP TABLE IF EXISTS trigger_dev.queue_metrics_raw_v1;
diff --git a/internal-packages/clickhouse/src/index.ts b/internal-packages/clickhouse/src/index.ts
@@ -36,7 +36,8 @@ import {
   insertQueueMetricsRaw,
   getQueueListMetricsSummary,
   getQueueDepthSparklines,
-  getQueueRankingPage,
+  getQueueRanking,
+  getQueueRankingNames,
   getQueueRankingCount,
 } from "./queueMetrics.js";
 import {
@@ -273,7 +274,8 @@ export class ClickHouse {
       insertRaw: insertQueueMetricsRaw(this.writer),
       listSummary: getQueueListMetricsSummary(this.reader),
       depthSparklines: getQueueDepthSparklines(this.reader),
-      rankingPage: getQueueRankingPage(this.reader),
+      ranking: getQueueRanking(this.reader),
+      rankingNames: getQueueRankingNames(this.reader),
       rankingCount: getQueueRankingCount(this.reader),
     };
   }
diff --git a/internal-packages/clickhouse/src/queueMetrics.test.ts b/internal-packages/clickhouse/src/queueMetrics.test.ts
@@ -213,4 +213,122 @@ describe("queue_metrics_v1", () => {
       await ch.close();
     }
   );
+
+  clickhouseTest(
+    "5m and env rollups agree with the 10s tier, and cross-queue totals sum per queue",
+    async ({ clickhouseContainer }) => {
+      const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
+
+      // Own org so the env-level read (no queue filter) stays isolated from other tests.
+      const rollOrg = "org_qm_roll";
+      const rows: QueueMetricsRawV1Input[] = [
+        ...counter("started", "roll-a", 7, [100, 150, 200, 250, 300, 350, 400]),
+        ...counter("started", "roll-b", 3, [500, 600, 700]),
+        { ...base("gauge", "roll-a"), running: 4, queued: 9, env_running: 30, env_limit: 50 },
+        { ...base("gauge", "roll-b"), running: 2, queued: 1, env_running: 45, env_limit: 50 },
+      ].map((row) => ({ ...row, organization_id: rollOrg }));
+      const [insertError] = await ch.queueMetrics.insertRaw(rows, SYNC);
+      expect(insertError).toBeNull();
+
+      const perQueue = (table: string) =>
+        ch.reader.query({
+          name: "per-queue-both-tiers",
+          query: `SELECT queue_name, deltaSumTimestampMerge(started_delta) AS started
+          FROM ${table}
+          WHERE queue_name IN ('roll-a', 'roll-b')
+          GROUP BY queue_name ORDER BY queue_name`,
+          schema: z.object({ queue_name: z.string(), started: z.coerce.number() }),
+        })({});
+      const [e10, rows10] = await perQueue("trigger_dev.queue_metrics_v1");
+      const [e5m, rows5m] = await perQueue("trigger_dev.queue_metrics_5m_v1");
+      expect(e10).toBeNull();
+      expect(e5m).toBeNull();
+      expect(rows10).toEqual([
+        { queue_name: "roll-a", started: 7 },
+        { queue_name: "roll-b", started: 3 },
+      ]);
+      expect(rows5m).toEqual(rows10);
+
+      // Env-wide totals: sum of per-queue merges (a single merge across queues would mix
+      // odometers and double-count).
+      const [envTotalError, envTotal] = await ch.reader.query({
+        name: "env-total-per-queue-sum",
+        query: `SELECT sum(started) AS started FROM (
+            SELECT queue_name, deltaSumTimestampMerge(started_delta) AS started
+            FROM trigger_dev.queue_metrics_5m_v1
+            WHERE queue_name IN ('roll-a', 'roll-b')
+            GROUP BY queue_name
+          )`,
+        schema: z.object({ started: z.coerce.number() }),
+      })({});
+      expect(envTotalError).toBeNull();
+      expect(envTotal![0]!.started).toBe(10);
+
+      const [envError, envRows] = await ch.reader.query({
+        name: "env-rollup-read",
+        query: `SELECT
+            max(max_env_running) AS max_env_running,
+            max(max_env_limit) AS max_env_limit,
+            round(quantilesTDigestMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[4]) AS wait_p99
+          FROM trigger_dev.env_metrics_1m_v1
+          WHERE organization_id = {org: String}`,
+        schema: z.object({
+          max_env_running: z.coerce.number(),
+          max_env_limit: z.coerce.number(),
+          wait_p99: z.coerce.number(),
+        }),
+        params: z.object({ org: z.string() }),
+      })({ org: rollOrg });
+      expect(envError).toBeNull();
+      expect(envRows![0]!.max_env_running).toBe(45);
+      expect(envRows![0]!.max_env_limit).toBe(50);
+      expect(envRows![0]!.wait_p99).toBeGreaterThanOrEqual(600);
+      expect(envRows![0]!.wait_p99).toBeLessThanOrEqual(1000);
+
+      await ch.close();
+    }
+  );
+
+  clickhouseTest(
+    "merged ranking returns the page and the windowed total in one query",
+    async ({ clickhouseContainer }) => {
+      const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
+
+      const gauge = (queue: string, queued: number, running: number): QueueMetricsRawV1Input => ({
+        ...base("gauge", queue),
+        queued,
+        running,
+      });
+      const [insertError] = await ch.queueMetrics.insertRaw(
+        [gauge("rank-low", 1, 0), gauge("rank-high", 50, 3), gauge("rank-mid", 10, 2)],
+        SYNC
+      );
+      expect(insertError).toBeNull();
+
+      const args = {
+        organizationId: ORG,
+        projectId: PROJECT,
+        environmentId: ENV,
+        startTime: "2026-06-30 11:50:00",
+        nameContains: "rank-",
+        byQueuedOnly: 0,
+      };
+      const [pageError, page] = await ch.queueMetrics.ranking({ ...args, limit: 2, offset: 0 });
+      expect(pageError).toBeNull();
+      expect(page).toEqual([
+        { queue_name: "rank-high", ranked_total: 3 },
+        { queue_name: "rank-mid", ranked_total: 3 },
+      ]);
+
+      const [countError, count] = await ch.queueMetrics.rankingCount(args);
+      expect(countError).toBeNull();
+      expect(count![0]!.ranked).toBe(3);
+
+      const [namesError, names] = await ch.queueMetrics.rankingNames({ ...args, limit: 10 });
+      expect(namesError).toBeNull();
+      expect(names!.map((r) => r.queue_name)).toEqual(["rank-high", "rank-mid", "rank-low"]);
+
+      await ch.close();
+    }
+  );
 });
diff --git a/internal-packages/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts
@@ -50,6 +50,12 @@ const QueueMetricsSummaryRow = z.object({
   started_count: z.coerce.number(),
 });
 
+// Callers align window bounds to the bucket grid so repeated loads share cache entries.
+const QUEUE_METRICS_CACHE_SETTINGS = {
+  use_query_cache: 1,
+  query_cache_ttl: 30,
+} as const;
+
 /** Per-queue rollups over a window, for a fixed set of queues (the visible page). */
 export function getQueueListMetricsSummary(reader: ClickhouseReader) {
   return reader.query({
@@ -70,6 +76,7 @@ export function getQueueListMetricsSummary(reader: ClickhouseReader) {
       GROUP BY queue_name`,
     params: QueueMetricsListParams,
     schema: QueueMetricsSummaryRow,
+    settings: QUEUE_METRICS_CACHE_SETTINGS,
   });
 }
 
@@ -102,6 +109,7 @@ export function getQueueDepthSparklines(reader: ClickhouseReader) {
       ORDER BY bucket`,
     params: QueueDepthSparklineParams,
     schema: QueueDepthSparklineRow,
+    settings: QUEUE_METRICS_CACHE_SETTINGS,
   });
 }
 
@@ -119,29 +127,61 @@ const QueueRankingParams = z.object({
 
 const QueueRankingRow = z.object({
   queue_name: z.string(),
+  ranked_total: z.coerce.number(),
 });
 
+// Ranking reads the 5m rollup: a 15-minute window there costs ~30x fewer rows than the
+// 10s table.
 const RANKING_WHERE = `organization_id = {organizationId: String}
         AND project_id = {projectId: String}
         AND environment_id = {environmentId: String}
         AND bucket_start >= {startTime: DateTime}
         AND queue_name != '__overflow__'
         AND ({nameContains: String} = '' OR positionCaseInsensitive(queue_name, {nameContains: String}) > 0)`;
 
-/** Queue names ranked by recent activity, for relevance-ordered list pages. */
-export function getQueueRankingPage(reader: ClickhouseReader) {
+/**
+ * One page of queue names ranked by recent activity, with the total ranked count on
+ * every row (window function), so page + count cost a single scan.
+ */
+export function getQueueRanking(reader: ClickhouseReader) {
   return reader.query({
-    name: "getQueueRankingPage",
-    query: `SELECT queue_name
-      FROM trigger_dev.queue_metrics_v1
-      WHERE ${RANKING_WHERE}
-      GROUP BY queue_name
-      ORDER BY
-        if({byQueuedOnly: UInt8} = 1, max(max_queued), max(max_queued) + max(max_running)) DESC,
-        queue_name ASC
+    name: "getQueueRanking",
+    query: `SELECT queue_name, count() OVER () AS ranked_total
+      FROM (
+        SELECT queue_name
+        FROM trigger_dev.queue_metrics_5m_v1
+        WHERE ${RANKING_WHERE}
+        GROUP BY queue_name
+        ORDER BY
+          if({byQueuedOnly: UInt8} = 1, max(max_queued), max(max_queued) + max(max_running)) DESC,
+          queue_name ASC
+      )
       LIMIT {limit: UInt32} OFFSET {offset: UInt32}`,
     params: QueueRankingParams,
     schema: QueueRankingRow,
+    settings: QUEUE_METRICS_CACHE_SETTINGS,
+  });
+}
+
+const QueueRankingNamesParams = QueueRankingParams.omit({ byQueuedOnly: true, offset: true });
+
+const QueueRankingNameRow = z.object({
+  queue_name: z.string(),
+});
+
+/** All ranked queue names (activity order), used to exclude them from the alphabetical tail. */
+export function getQueueRankingNames(reader: ClickhouseReader) {
+  return reader.query({
+    name: "getQueueRankingNames",
+    query: `SELECT queue_name
+      FROM trigger_dev.queue_metrics_5m_v1
+      WHERE ${RANKING_WHERE}
+      GROUP BY queue_name
+      ORDER BY max(max_queued) + max(max_running) DESC, queue_name ASC
+      LIMIT {limit: UInt32}`,
+    params: QueueRankingNamesParams,
+    schema: QueueRankingNameRow,
+    settings: QUEUE_METRICS_CACHE_SETTINGS,
   });
 }
 
@@ -155,15 +195,16 @@ const QueueRankingCountRow = z.object({
   ranked: z.coerce.number(),
 });
 
-/** How many queues have activity in the ranking window (the ranked head of the list). */
+/** Ranked-queue count alone, for pages past the ranked head (approximate uniq is fine). */
 export function getQueueRankingCount(reader: ClickhouseReader) {
   return reader.query({
     name: "getQueueRankingCount",
-    query: `SELECT uniqExact(queue_name) AS ranked
-      FROM trigger_dev.queue_metrics_v1
+    query: `SELECT uniq(queue_name) AS ranked
+      FROM trigger_dev.queue_metrics_5m_v1
       WHERE ${RANKING_WHERE}`,
     params: QueueRankingCountParams,
     schema: QueueRankingCountRow,
+    settings: QUEUE_METRICS_CACHE_SETTINGS,
   });
 }
 
diff --git a/internal-packages/tsql/src/query/functions.ts b/internal-packages/tsql/src/query/functions.ts
@@ -645,6 +645,13 @@ export const TSQL_AGGREGATIONS: Record<string, TSQLFunctionMeta> = {
     maxParams: 1,
     aggregate: true,
   },
+  quantilesTDigestMerge: {
+    clickhouseName: "quantilesTDigestMerge",
+    minArgs: 1,
+    maxArgs: 1,
+    minParams: 1,
+    aggregate: true,
+  },
   sumMerge: { clickhouseName: "sumMerge", minArgs: 1, maxArgs: 1, aggregate: true },
   avgMerge: { clickhouseName: "avgMerge", minArgs: 1, maxArgs: 1, aggregate: true },
   countMerge: { clickhouseName: "countMerge", minArgs: 1, maxArgs: 1, aggregate: true },
diff --git a/internal-packages/tsql/src/query/printer.test.ts b/internal-packages/tsql/src/query/printer.test.ts
diff --git a/internal-packages/tsql/src/query/schema.ts b/internal-packages/tsql/src/query/schema.ts
