feat(tsql,webapp): reject cross-queue merges of per-queue counter states

ericallam · ericallam · commit c8af713debdc · 2026-07-05T10:13:18.000+01:00
deltaSumTimestamp states are kept per queue, and merging them across
queues silently returns wrong totals, on the dashboard and the public
query API alike. Columns can now declare a mergeGroupKey, and the
compiler rejects queries that merge such a column without grouping by
that key or pinning it to a single value. The error names the column,
explains the failure, and includes a corrected example query.
diff --git a/apps/webapp/app/v3/querySchemas.ts b/apps/webapp/app/v3/querySchemas.ts
@@ -668,6 +668,7 @@ export const queueMetricsSchema: TableSchema = {
     // invalid (mixes unrelated odometers): totals must GROUP BY queue, then sum outside.
     enqueue_delta: {
       name: "enqueue_delta",
+      mergeGroupKey: "queue",
       ...column("String", {
         description:
           "Runs enqueued (cumulative-counter delta). Read with deltaSumTimestampMerge(enqueue_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues.",
@@ -678,6 +679,7 @@ export const queueMetricsSchema: TableSchema = {
     },
     started_delta: {
       name: "started_delta",
+      mergeGroupKey: "queue",
       ...column("String", {
         description:
           "Runs dequeued/started (throughput). Read with deltaSumTimestampMerge(started_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues.",
@@ -689,6 +691,7 @@ export const queueMetricsSchema: TableSchema = {
     },
     ack_delta: {
       name: "ack_delta",
+      mergeGroupKey: "queue",
       ...column("String", {
         description:
           "Runs acked (completed). Read with deltaSumTimestampMerge(ack_delta) grouped by queue; sum per-queue results for totals.",
@@ -699,6 +702,7 @@ export const queueMetricsSchema: TableSchema = {
     },
     nack_delta: {
       name: "nack_delta",
+      mergeGroupKey: "queue",
       ...column("String", {
         description:
           "Runs nacked. Read with deltaSumTimestampMerge(nack_delta) grouped by queue; sum per-queue results for totals.",
@@ -709,6 +713,7 @@ export const queueMetricsSchema: TableSchema = {
     },
     dlq_delta: {
       name: "dlq_delta",
+      mergeGroupKey: "queue",
       ...column("String", {
         description:
           "Runs dead-lettered. Read with deltaSumTimestampMerge(dlq_delta) grouped by queue; sum per-queue results for totals.",
diff --git a/internal-packages/tsql/src/query/printer.test.ts b/internal-packages/tsql/src/query/printer.test.ts
@@ -3995,6 +3995,7 @@ describe("cross-queue counter totals via subquery (env-wide throughput shape)",
       queue_name: { name: "queue_name", ...column("String") },
       started_delta: {
         name: "started_delta",
+        mergeGroupKey: "queue_name",
         ...column("String"),
         groupable: false,
         sortable: false,
@@ -4047,3 +4048,96 @@ describe("cross-queue counter totals via subquery (env-wide throughput shape)",
     expect(Object.values(params)).toContain("org_test123");
   });
 });
+
+describe("mergeGroupKey validation", () => {
+  const metricsSchema: TableSchema = {
+    name: "metrics",
+    clickhouseName: "trigger_dev.queue_metrics_v1",
+    timeConstraint: "bucket_at",
+    columns: {
+      bucket_at: { name: "bucket_at", ...column("DateTime64") },
+      queue: { name: "queue", clickhouseName: "queue_name", ...column("String") },
+      started_delta: {
+        name: "started_delta",
+        mergeGroupKey: "queue",
+        ...column("String"),
+        groupable: false,
+        sortable: false,
+        filterable: false,
+      },
+      organization_id: { name: "organization_id", ...column("String") },
+    },
+    tenantColumns: { organizationId: "organization_id" },
+  };
+
+  function compile(
+    query: string,
+    enforced: Record<string, unknown> = { organization_id: { op: "eq", value: "org_x" } }
+  ) {
+    const context = createPrinterContext({
+      schema: createSchemaRegistry([metricsSchema]),
+      enforcedWhereClause: enforced as never,
+      timeRange: {
+        from: new Date("2024-01-01T00:00:00Z"),
+        to: new Date("2024-01-08T00:00:00Z"),
+      },
+    });
+    return printToClickHouse(parseTSQLSelect(query), context);
+  }
+
+  it("rejects an ungrouped, unpinned merge with an actionable message", () => {
+    expect(() =>
+      compile(
+        "SELECT timeBucket() AS t, deltaSumTimestampMerge(started_delta) AS started FROM metrics GROUP BY t"
+      )
+    ).toThrowError(
+      /Merging 'started_delta' across every queue[\s\S]*GROUP BY queue\)[\s\S]*WHERE queue = 'my-queue'[\s\S]*inner GROUP BY t, queue and outer GROUP BY t/
+    );
+  });
+
+  it("allows the merge when queue is in the GROUP BY", () => {
+    const { sql } = compile(
+      "SELECT timeBucket() AS t, queue, deltaSumTimestampMerge(started_delta) AS started FROM metrics GROUP BY t, queue"
+    );
+    expect(sql).toContain("deltaSumTimestampMerge(started_delta)");
+  });
+
+  it("allows the merge when queue is pinned by an equality filter", () => {
+    const { sql } = compile(
+      "SELECT deltaSumTimestampMerge(started_delta) AS started FROM metrics WHERE queue = 'emails'"
+    );
+    expect(sql).toContain("deltaSumTimestampMerge(started_delta)");
+  });
+
+  it("allows the merge when the enforced clause pins queue to one value", () => {
+    const { sql } = compile(
+      "SELECT deltaSumTimestampMerge(started_delta) AS started FROM metrics",
+      { organization_id: { op: "eq", value: "org_x" }, queue: { op: "in", values: ["emails"] } }
+    );
+    expect(sql).toContain("deltaSumTimestampMerge(started_delta)");
+  });
+
+  it("rejects the merge when the enforced clause spans several queues", () => {
+    expect(() =>
+      compile("SELECT deltaSumTimestampMerge(started_delta) AS started FROM metrics", {
+        organization_id: { op: "eq", value: "org_x" },
+        queue: { op: "in", values: ["emails", "webhooks"] },
+      })
+    ).toThrowError(/only combine correctly within one queue/);
+  });
+
+  it("allows a grouped inner merge summed by the outer query", () => {
+    const { sql } = compile(
+      "SELECT t, sum(started) AS started FROM (SELECT timeBucket() AS t, queue, deltaSumTimestampMerge(started_delta) AS started FROM metrics GROUP BY t, queue) GROUP BY t ORDER BY t"
+    );
+    expect(sql).toContain("GROUP BY t, queue_name");
+  });
+
+  it("rejects an ungrouped merge inside a subquery", () => {
+    expect(() =>
+      compile(
+        "SELECT t, sum(started) AS started FROM (SELECT timeBucket() AS t, deltaSumTimestampMerge(started_delta) AS started FROM metrics GROUP BY t) GROUP BY t"
+      )
+    ).toThrowError(/only combine correctly within one queue/);
+  });
+});
diff --git a/internal-packages/tsql/src/query/printer.ts b/internal-packages/tsql/src/query/printer.ts
@@ -385,6 +385,8 @@ export class ClickHousePrinter {
       nextJoin = nextJoin.next_join;
     }
 
+    this.validateMergeScopedColumns(node);
+
     // Extract SELECT column aliases BEFORE visiting columns
     // This allows ORDER BY/HAVING to reference aliased columns
     const savedAliases = this.selectAliases;
@@ -1883,6 +1885,134 @@ export class ClickHousePrinter {
     // Note: projectId and environmentId are optional - no validation needed
   }
 
+  /**
+   * Reject queries that merge a scope-keyed aggregate state column (`mergeGroupKey`)
+   * across values of its key: such merges silently return wrong numbers. Valid shapes
+   * group by the key column or pin it to a single value (in the query's WHERE or via
+   * the enforced clause). Runs per SELECT scope; subqueries validate themselves.
+   */
+  private validateMergeScopedColumns(node: SelectQuery): void {
+    for (const tableSchema of this.tableContexts.values()) {
+      for (const column of Object.values(tableSchema.columns)) {
+        const key = column.mergeGroupKey;
+        if (!key) continue;
+        if (!this.scopeReferencesColumn(node, column.name)) continue;
+        if (this.groupByIncludesColumn(node, key)) continue;
+        if (this.wherePinsColumn(node.where, key)) continue;
+        if (this.enforcedPinsColumn(tableSchema, key)) continue;
+        throw new QueryError(
+          `Merging '${column.name}' across every ${key} returns wrong totals: its aggregate ` +
+            `states are kept per ${key} and only combine correctly within one ${key}. Either ` +
+            `add '${key}' to the GROUP BY and sum the per-${key} results in an outer query, ` +
+            `for example: SELECT sum(v) AS total FROM (SELECT ${key}, ` +
+            `deltaSumTimestampMerge(${column.name}) AS v FROM ${tableSchema.name} ` +
+            `GROUP BY ${key}). Or filter to a single ${key}, for example: ` +
+            `WHERE ${key} = 'my-${key}'. For a time series, bucket both layers: ` +
+            `inner GROUP BY t, ${key} and outer GROUP BY t.`
+        );
+      }
+    }
+  }
+
+  private scopeReferencesColumn(node: SelectQuery, name: string): boolean {
+    const parts: unknown[] = [
+      node.select,
+      node.prewhere,
+      node.where,
+      node.group_by,
+      node.having,
+      node.order_by,
+    ];
+    return parts.some((part) => this.expressionReferencesColumn(part, name));
+  }
+
+  private expressionReferencesColumn(
+    expr: unknown,
+    name: string,
+    seen = new WeakSet<object>()
+  ): boolean {
+    if (expr === null || typeof expr !== "object") return false;
+    if (seen.has(expr)) return false;
+    seen.add(expr);
+    if (Array.isArray(expr)) {
+      return expr.some((item) => this.expressionReferencesColumn(item, name, seen));
+    }
+    const candidate = expr as { expression_type?: string; chain?: unknown[] };
+    if (
+      candidate.expression_type === "select_query" ||
+      candidate.expression_type === "select_set_query"
+    ) {
+      return false;
+    }
+    if (
+      candidate.expression_type === "field" &&
+      Array.isArray(candidate.chain) &&
+      candidate.chain[candidate.chain.length - 1] === name
+    ) {
+      return true;
+    }
+    return Object.entries(expr).some(
+      ([property, value]) =>
+        property !== "type" &&
+        property !== "parent" &&
+        this.expressionReferencesColumn(value, name, seen)
+    );
+  }
+
+  private groupByIncludesColumn(node: SelectQuery, name: string): boolean {
+    return (node.group_by ?? []).some((expr) => {
+      const field = expr as Field;
+      return (
+        field.expression_type === "field" &&
+        Array.isArray(field.chain) &&
+        field.chain[field.chain.length - 1] === name
+      );
+    });
+  }
+
+  // Pins only count on the top-level AND chain: a pin inside an OR guarantees nothing.
+  private wherePinsColumn(where: Expression | undefined, name: string): boolean {
+    if (!where) return false;
+    if (where.expression_type === "and") {
+      return (where as And).exprs.some((expr) => this.wherePinsColumn(expr, name));
+    }
+    if (where.expression_type !== "compare_operation") return false;
+    const cmp = where as CompareOperation;
+    const isKeyField = (side: Expression) => {
+      const field = side as Field;
+      return (
+        field.expression_type === "field" &&
+        Array.isArray(field.chain) &&
+        field.chain[field.chain.length - 1] === name
+      );
+    };
+    const fieldSide = [cmp.left, cmp.right].find(isKeyField);
+    if (!fieldSide) return false;
+    if (cmp.op === CompareOperationOp.Eq) return true;
+    if (cmp.op === CompareOperationOp.In || cmp.op === CompareOperationOp.GlobalIn) {
+      const other = fieldSide === cmp.left ? cmp.right : cmp.left;
+      if ((other as Constant).expression_type === "constant") return true;
+      const tuple = other as Tuple;
+      return tuple.expression_type === "tuple" && tuple.exprs.length === 1;
+    }
+    return false;
+  }
+
+  private enforcedPinsColumn(tableSchema: TableSchema, key: string): boolean {
+    const names = [key];
+    const clickhouseName = tableSchema.columns[key]?.clickhouseName;
+    if (clickhouseName) names.push(clickhouseName);
+    for (const name of names) {
+      const condition = this.context.enforcedWhereClause[name] as
+        | { op?: string; values?: unknown[] }
+        | undefined;
+      if (!condition) continue;
+      if (condition.op === "eq") return true;
+      if (condition.op === "in" && condition.values?.length === 1) return true;
+    }
+    return false;
+  }
+
   /**
    * Format a Date as a ClickHouse-compatible DateTime64 string.
    * ClickHouse expects format: 'YYYY-MM-DD HH:MM:SS.mmm' (in UTC)
diff --git a/internal-packages/tsql/src/query/schema.ts b/internal-packages/tsql/src/query/schema.ts
@@ -128,6 +128,12 @@ export interface ColumnSchema {
    * = counter (missing buckets get 0).
    */
   fillMode?: "zero" | "carry";
+  /**
+   * Aggregate-state column whose states only merge correctly within one value of the
+   * named column (e.g. per-queue counter states). Queries referencing it must GROUP BY
+   * that column or pin it to a single value; other shapes fail to compile.
+   */
+  mergeGroupKey?: string;
   /**
    * Example value for documentation purposes.
    *
