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

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.

Author: ericallam

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.",

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/);
+  });
+});

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)

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.
    *