From d6b02fae7d21a80357e071e61ec26f368157738b Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Tue, 1 Sep 2026 11:50:55 +0100 Subject: [PATCH 1/2] perf(clickhouse): prepare task event attribute inserts --- .../services/admin/missingLlmModels.server.ts | 28 ++++++-- ...ckhouseEventRepositoryJsonRecovery.test.ts | 2 +- .../clickhouse/src/client/client.ts | 3 + .../clickhouse/src/client/types.ts | 2 + .../clickhouse/src/taskEvents.test.ts | 67 +++++++++++++++++++ .../clickhouse/src/taskEvents.ts | 22 ++++++ 6 files changed, 117 insertions(+), 7 deletions(-) create mode 100644 internal-packages/clickhouse/src/taskEvents.test.ts diff --git a/apps/webapp/app/services/admin/missingLlmModels.server.ts b/apps/webapp/app/services/admin/missingLlmModels.server.ts index 2ad5bffb520..c555f7e1aa1 100644 --- a/apps/webapp/app/services/admin/missingLlmModels.server.ts +++ b/apps/webapp/app/services/admin/missingLlmModels.server.ts @@ -26,8 +26,14 @@ export async function getMissingLlmModels( name: "missingLlmModels", table: "trigger_dev.task_events_v2", columns: [ - { name: "model", expression: "attributes.gen_ai.response.model.:String" }, - { name: "system", expression: "attributes.gen_ai.system.:String" }, + { + name: "model", + expression: "JSONExtractString(attributes_text, 'gen_ai', 'response', 'model')", + }, + { + name: "system", + expression: "JSONExtractString(attributes_text, 'gen_ai', 'system')", + }, { name: "cnt", expression: "count()" }, ], }); @@ -39,10 +45,15 @@ export async function getMissingLlmModels( }); // Only spans that have a model set - qb.where("attributes.gen_ai.response.model.:String != {empty: String}", { empty: "" }); + qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') != {empty: String}", { + empty: "", + }); // Only spans that were NOT cost-enriched (trigger.llm.total_cost is NULL) - qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {}); + qb.where( + "JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL", + {} + ); // Only completed spans qb.where("kind = {kind: String}", { kind: "SPAN" }); @@ -107,8 +118,13 @@ export async function getMissingModelSamples(opts: { const qb = createBuilder(); qb.where("inserted_at >= {since: DateTime64(3)}", { since: formatDateTime(since) }); - qb.where("attributes.gen_ai.response.model.:String = {model: String}", { model: opts.model }); - qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {}); + qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') = {model: String}", { + model: opts.model, + }); + qb.where( + "JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL", + {} + ); qb.where("kind = {kind: String}", { kind: "SPAN" }); qb.where("status = {status: String}", { status: "OK" }); qb.orderBy("start_time DESC"); diff --git a/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts index 375d7c4498d..3b04a571226 100644 --- a/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts +++ b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts @@ -89,7 +89,7 @@ describe("ClickhouseEventRepository JSON parse recovery", () => { const queryEvents = clickhouse.reader.query({ name: "event-recovery-check", query: - "SELECT span_id, toJSONString(attributes) AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}", + "SELECT span_id, attributes_text AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}", schema: z.object({ span_id: z.string(), attributes_json: z.string() }), params: z.object({ env_id: z.string() }), }); diff --git a/internal-packages/clickhouse/src/client/client.ts b/internal-packages/clickhouse/src/client/client.ts index 96db1f4a529..a208f4395cd 100644 --- a/internal-packages/clickhouse/src/client/client.ts +++ b/internal-packages/clickhouse/src/client/client.ts @@ -5,6 +5,7 @@ import { type ClickHouseSettings, createClient, type BaseQueryParams, + type InsertParams, type InsertResult, } from "@clickhouse/client"; import type { Counter, Histogram, Meter, Span, Tracer, UpDownCounter } from "@internal/tracing"; @@ -1078,6 +1079,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { public insertUnsafe>(req: { name: string; table: string; + columns?: InsertParams["columns"]; settings?: ClickHouseSettings; }): ClickhouseInsertFunction { return async (events, options) => { @@ -1109,6 +1111,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { const [clickhouseError, result] = await tryCatch( this.client.insert({ table: req.table, + columns: req.columns, format: "JSONEachRow", values: eventsArray, query_id: queryId, diff --git a/internal-packages/clickhouse/src/client/types.ts b/internal-packages/clickhouse/src/client/types.ts index 6cfbe35fd48..b6b16fbd34f 100644 --- a/internal-packages/clickhouse/src/client/types.ts +++ b/internal-packages/clickhouse/src/client/types.ts @@ -5,6 +5,7 @@ import { type ClickHouseSettings, type BaseQueryParams, type CommandResult, + type InsertParams, type InsertResult, } from "@clickhouse/client"; import type { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js"; @@ -272,6 +273,7 @@ export interface ClickhouseWriter { insertUnsafe>(req: { name: string; table: string; + columns?: InsertParams["columns"]; settings?: ClickHouseSettings; }): ClickhouseInsertFunction; diff --git a/internal-packages/clickhouse/src/taskEvents.test.ts b/internal-packages/clickhouse/src/taskEvents.test.ts new file mode 100644 index 00000000000..50eb3ff0c45 --- /dev/null +++ b/internal-packages/clickhouse/src/taskEvents.test.ts @@ -0,0 +1,67 @@ +import { clickhouseTest } from "@internal/testcontainers"; +import { z } from "zod"; +import { ClickHouse } from "./index.js"; + +function clickhouseDate(value: Date) { + return value.toISOString().replace("T", " ").replace("Z", ""); +} + +describe("task events v2", () => { + clickhouseTest( + "stores materialized attributes with explicit insert columns", + async ({ clickhouseContainer }) => { + const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" }); + const now = new Date("2026-09-01T10:00:00.000Z"); + const spanId = "span_ephemeral_attributes"; + + const [insertError] = await ch.taskEventsV2.insert([ + { + environment_id: "env_ephemeral_attributes", + organization_id: "org_ephemeral_attributes", + project_id: "project_ephemeral_attributes", + task_identifier: "ephemeral-attributes", + run_id: "run_ephemeral_attributes", + start_time: clickhouseDate(now), + duration: "1000000", + trace_id: "trace_ephemeral_attributes", + span_id: spanId, + parent_span_id: "", + message: "Ephemeral attributes", + kind: "SPAN", + status: "OK", + attributes: { + z: 1, + a: "hello", + nested: { enabled: true }, + }, + metadata: "{}", + expires_at: clickhouseDate(new Date(now.getTime() + 90 * 24 * 60 * 60 * 1000)), + }, + ]); + expect(insertError).toBeNull(); + + const readAttributes = ch.reader.query({ + name: "read-ephemeral-task-event-attributes", + query: `SELECT attributes_text, + toUInt8(inserted_at > toDateTime64('2020-01-01 00:00:00', 3)) AS has_inserted_at + FROM trigger_dev.task_events_v2 + WHERE environment_id = {environmentId: String} + AND span_id = {spanId: String}`, + params: z.object({ environmentId: z.string(), spanId: z.string() }), + schema: z.object({ attributes_text: z.string(), has_inserted_at: z.number() }), + }); + const [readError, rows] = await readAttributes({ + environmentId: "env_ephemeral_attributes", + spanId, + }); + expect(readError).toBeNull(); + expect(rows).toEqual([ + { + attributes_text: '{"a":"hello","nested":{"enabled":true},"z":1}', + has_inserted_at: 1, + }, + ]); + + } + ); +}); diff --git a/internal-packages/clickhouse/src/taskEvents.ts b/internal-packages/clickhouse/src/taskEvents.ts index c79a048d481..3e59d4587c5 100644 --- a/internal-packages/clickhouse/src/taskEvents.ts +++ b/internal-packages/clickhouse/src/taskEvents.ts @@ -176,6 +176,27 @@ export function getSpanDetailsQueryBuilder(ch: ClickhouseReader, settings?: Clic // V2 Table Functions (partitioned by inserted_at instead of start_time) // ============================================================================ +const TASK_EVENT_V2_INSERT_COLUMNS = [ + "environment_id", + "organization_id", + "project_id", + "task_identifier", + "run_id", + "start_time", + "duration", + "trace_id", + "span_id", + "parent_span_id", + "message", + "kind", + "status", + "attributes", + "metadata", + "expires_at", + "machine_id", + "inserted_at", +] satisfies [string, ...string[]]; + export const TaskEventV2Input = z.object({ environment_id: z.string(), organization_id: z.string(), @@ -204,6 +225,7 @@ export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSe return ch.insertUnsafe({ name: "insertTaskEventsV2", table: "trigger_dev.task_events_v2", + columns: TASK_EVENT_V2_INSERT_COLUMNS, settings: { enable_json_type: 1, type_json_skip_duplicated_paths: 1, From 113b0d66f77c81717f28bae50f150227bcc79652 Mon Sep 17 00:00:00 2001 From: Chris Arderne Date: Wed, 2 Sep 2026 09:08:16 +0100 Subject: [PATCH 2/2] test(clickhouse): keep task event fixture within ttl --- internal-packages/clickhouse/src/taskEvents.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal-packages/clickhouse/src/taskEvents.test.ts b/internal-packages/clickhouse/src/taskEvents.test.ts index 50eb3ff0c45..8a2599bf90d 100644 --- a/internal-packages/clickhouse/src/taskEvents.test.ts +++ b/internal-packages/clickhouse/src/taskEvents.test.ts @@ -11,7 +11,8 @@ describe("task events v2", () => { "stores materialized attributes with explicit insert columns", async ({ clickhouseContainer }) => { const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" }); - const now = new Date("2026-09-01T10:00:00.000Z"); + const startTime = new Date("2026-09-01T10:00:00.000Z"); + const expiresAt = new Date(Date.now() + 90 * 24 * 60 * 60 * 1000); const spanId = "span_ephemeral_attributes"; const [insertError] = await ch.taskEventsV2.insert([ @@ -21,7 +22,7 @@ describe("task events v2", () => { project_id: "project_ephemeral_attributes", task_identifier: "ephemeral-attributes", run_id: "run_ephemeral_attributes", - start_time: clickhouseDate(now), + start_time: clickhouseDate(startTime), duration: "1000000", trace_id: "trace_ephemeral_attributes", span_id: spanId, @@ -35,7 +36,7 @@ describe("task events v2", () => { nested: { enabled: true }, }, metadata: "{}", - expires_at: clickhouseDate(new Date(now.getTime() + 90 * 24 * 60 * 60 * 1000)), + expires_at: clickhouseDate(expiresAt), }, ]); expect(insertError).toBeNull(); @@ -61,7 +62,6 @@ describe("task events v2", () => { has_inserted_at: 1, }, ]); - } ); });