Skip to content

Commit 9fc88e0

Browse files
cardernecarderne
committed
perf(clickhouse): reduce task event storage overhead
1 parent 43ecf15 commit 9fc88e0

7 files changed

Lines changed: 189 additions & 7 deletions

File tree

‎apps/webapp/app/services/admin/missingLlmModels.server.ts‎

Lines changed: 22 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,14 @@ export async function getMissingLlmModels(
2626
name: "missingLlmModels",
2727
table: "trigger_dev.task_events_v2",
2828
columns: [
29-
{ name: "model", expression: "attributes.gen_ai.response.model.:String" },
30-
{ name: "system", expression: "attributes.gen_ai.system.:String" },
29+
{
30+
name: "model",
31+
expression: "JSONExtractString(attributes_text, 'gen_ai', 'response', 'model')",
32+
},
33+
{
34+
name: "system",
35+
expression: "JSONExtractString(attributes_text, 'gen_ai', 'system')",
36+
},
3137
{ name: "cnt", expression: "count()" },
3238
],
3339
});
@@ -39,10 +45,15 @@ export async function getMissingLlmModels(
3945
});
4046

4147
// Only spans that have a model set
42-
qb.where("attributes.gen_ai.response.model.:String != {empty: String}", { empty: "" });
48+
qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') != {empty: String}", {
49+
empty: "",
50+
});
4351

4452
// Only spans that were NOT cost-enriched (trigger.llm.total_cost is NULL)
45-
qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {});
53+
qb.where(
54+
"JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL",
55+
{}
56+
);
4657

4758
// Only completed spans
4859
qb.where("kind = {kind: String}", { kind: "SPAN" });
@@ -107,8 +118,13 @@ export async function getMissingModelSamples(opts: {
107118
const qb = createBuilder();
108119

109120
qb.where("inserted_at >= {since: DateTime64(3)}", { since: formatDateTime(since) });
110-
qb.where("attributes.gen_ai.response.model.:String = {model: String}", { model: opts.model });
111-
qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {});
121+
qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') = {model: String}", {
122+
model: opts.model,
123+
});
124+
qb.where(
125+
"JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL",
126+
{}
127+
);
112128
qb.where("kind = {kind: String}", { kind: "SPAN" });
113129
qb.where("status = {status: String}", { status: "OK" });
114130
qb.orderBy("start_time DESC");

‎apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ describe("ClickhouseEventRepository JSON parse recovery", () => {
8989
const queryEvents = clickhouse.reader.query({
9090
name: "event-recovery-check",
9191
query:
92-
"SELECT span_id, toJSONString(attributes) AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}",
92+
"SELECT span_id, attributes_text AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}",
9393
schema: z.object({ span_id: z.string(), attributes_json: z.string() }),
9494
params: z.object({ env_id: z.string() }),
9595
});

‎internal-packages/clickhouse/schema/042_reduce_task_events_v2_storage_overhead.sql‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
-- +goose Up
2+
3+
-- Full-text search is served outside the source event table. Keeping these
4+
-- indexes here adds work to every event insert and merge without serving reads.
5+
ALTER TABLE trigger_dev.task_events_v2
6+
DROP INDEX IF EXISTS idx_attributes_text_search;
7+
8+
ALTER TABLE trigger_dev.task_events_v2
9+
DROP INDEX IF EXISTS idx_message_text_search;
10+
11+
-- attributes remains an insert input for attributes_text, but is no longer
12+
-- stored. Writers must include attributes in an explicit insert column list
13+
-- because implicit INSERT column lists exclude EPHEMERAL columns.
14+
ALTER TABLE trigger_dev.task_events_v2
15+
MODIFY COLUMN attributes JSON EPHEMERAL;
16+
17+
-- +goose Down
18+
19+
ALTER TABLE trigger_dev.task_events_v2
20+
MODIFY COLUMN attributes JSON CODEC(ZSTD(1));
21+
22+
ALTER TABLE trigger_dev.task_events_v2
23+
ADD INDEX IF NOT EXISTS idx_attributes_text_search lower(attributes_text)
24+
TYPE ngrambf_v1(3, 32768, 2, 0)
25+
GRANULARITY 1;
26+
27+
ALTER TABLE trigger_dev.task_events_v2
28+
ADD INDEX IF NOT EXISTS idx_message_text_search lower(message)
29+
TYPE ngrambf_v1(3, 32768, 2, 0)
30+
GRANULARITY 1;

‎internal-packages/clickhouse/src/client/client.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
type ClickHouseSettings,
66
createClient,
77
type BaseQueryParams,
8+
type InsertParams,
89
type InsertResult,
910
} from "@clickhouse/client";
1011
import type { Counter, Histogram, Meter, Span, Tracer, UpDownCounter } from "@internal/tracing";
@@ -1078,6 +1079,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
10781079
public insertUnsafe<TRecord extends Record<string, any>>(req: {
10791080
name: string;
10801081
table: string;
1082+
columns?: InsertParams["columns"];
10811083
settings?: ClickHouseSettings;
10821084
}): ClickhouseInsertFunction<TRecord> {
10831085
return async (events, options) => {
@@ -1109,6 +1111,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
11091111
const [clickhouseError, result] = await tryCatch(
11101112
this.client.insert({
11111113
table: req.table,
1114+
columns: req.columns,
11121115
format: "JSONEachRow",
11131116
values: eventsArray,
11141117
query_id: queryId,

‎internal-packages/clickhouse/src/client/types.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
type ClickHouseSettings,
66
type BaseQueryParams,
77
type CommandResult,
8+
type InsertParams,
89
type InsertResult,
910
} from "@clickhouse/client";
1011
import type { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js";
@@ -272,6 +273,7 @@ export interface ClickhouseWriter {
272273
insertUnsafe<TRecord extends Record<string, any>>(req: {
273274
name: string;
274275
table: string;
276+
columns?: InsertParams["columns"];
275277
settings?: ClickHouseSettings;
276278
}): ClickhouseInsertFunction<TRecord>;
277279

‎internal-packages/clickhouse/src/taskEvents.test.ts‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
import { clickhouseTest } from "@internal/testcontainers";
2+
import { z } from "zod";
3+
import { ClickHouse } from "./index.js";
4+
5+
function clickhouseDate(value: Date) {
6+
return value.toISOString().replace("T", " ").replace("Z", "");
7+
}
8+
9+
describe("task events v2", () => {
10+
clickhouseTest(
11+
"stores materialized attributes from the ephemeral JSON input",
12+
async ({ clickhouseContainer }) => {
13+
const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
14+
const now = new Date("2026-09-01T10:00:00.000Z");
15+
const spanId = "span_ephemeral_attributes";
16+
17+
const [insertError] = await ch.taskEventsV2.insert([
18+
{
19+
environment_id: "env_ephemeral_attributes",
20+
organization_id: "org_ephemeral_attributes",
21+
project_id: "project_ephemeral_attributes",
22+
task_identifier: "ephemeral-attributes",
23+
run_id: "run_ephemeral_attributes",
24+
start_time: clickhouseDate(now),
25+
duration: "1000000",
26+
trace_id: "trace_ephemeral_attributes",
27+
span_id: spanId,
28+
parent_span_id: "",
29+
message: "Ephemeral attributes",
30+
kind: "SPAN",
31+
status: "OK",
32+
attributes: {
33+
z: 1,
34+
a: "hello",
35+
nested: { enabled: true },
36+
},
37+
metadata: "{}",
38+
expires_at: clickhouseDate(new Date(now.getTime() + 90 * 24 * 60 * 60 * 1000)),
39+
},
40+
]);
41+
expect(insertError).toBeNull();
42+
43+
const readAttributes = ch.reader.query({
44+
name: "read-ephemeral-task-event-attributes",
45+
query: `SELECT attributes_text,
46+
toUInt8(inserted_at > toDateTime64('2020-01-01 00:00:00', 3)) AS has_inserted_at
47+
FROM trigger_dev.task_events_v2
48+
WHERE environment_id = {environmentId: String}
49+
AND span_id = {spanId: String}`,
50+
params: z.object({ environmentId: z.string(), spanId: z.string() }),
51+
schema: z.object({ attributes_text: z.string(), has_inserted_at: z.number() }),
52+
});
53+
const [readError, rows] = await readAttributes({
54+
environmentId: "env_ephemeral_attributes",
55+
spanId,
56+
});
57+
expect(readError).toBeNull();
58+
expect(rows).toEqual([
59+
{
60+
attributes_text: '{"a":"hello","nested":{"enabled":true},"z":1}',
61+
has_inserted_at: 1,
62+
},
63+
]);
64+
65+
const readColumnKinds = ch.reader.query({
66+
name: "read-task-event-attribute-column-kinds",
67+
query: `SELECT name, default_kind, default_expression
68+
FROM system.columns
69+
WHERE database = 'trigger_dev'
70+
AND table = 'task_events_v2'
71+
AND name IN ('attributes', 'attributes_text')
72+
ORDER BY name`,
73+
schema: z.object({
74+
name: z.string(),
75+
default_kind: z.string(),
76+
default_expression: z.string(),
77+
}),
78+
});
79+
const [columnError, columns] = await readColumnKinds({});
80+
expect(columnError).toBeNull();
81+
expect(columns).toEqual([
82+
{
83+
name: "attributes",
84+
default_kind: "EPHEMERAL",
85+
default_expression: "defaultValueOfTypeName('JSON')",
86+
},
87+
{
88+
name: "attributes_text",
89+
default_kind: "MATERIALIZED",
90+
default_expression: "toJSONString(attributes)",
91+
},
92+
]);
93+
94+
const readRemovedIndexes = ch.reader.query({
95+
name: "read-removed-task-event-text-indexes",
96+
query: `SELECT name
97+
FROM system.data_skipping_indices
98+
WHERE database = 'trigger_dev'
99+
AND table = 'task_events_v2'
100+
AND name IN ('idx_attributes_text_search', 'idx_message_text_search')
101+
ORDER BY name`,
102+
schema: z.object({ name: z.string() }),
103+
});
104+
const [indexError, indexes] = await readRemovedIndexes({});
105+
expect(indexError).toBeNull();
106+
expect(indexes).toEqual([]);
107+
}
108+
);
109+
});

‎internal-packages/clickhouse/src/taskEvents.ts‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -176,6 +176,27 @@ export function getSpanDetailsQueryBuilder(ch: ClickhouseReader, settings?: Clic
176176
// V2 Table Functions (partitioned by inserted_at instead of start_time)
177177
// ============================================================================
178178

179+
const TASK_EVENT_V2_INSERT_COLUMNS = [
180+
"environment_id",
181+
"organization_id",
182+
"project_id",
183+
"task_identifier",
184+
"run_id",
185+
"start_time",
186+
"duration",
187+
"trace_id",
188+
"span_id",
189+
"parent_span_id",
190+
"message",
191+
"kind",
192+
"status",
193+
"attributes",
194+
"metadata",
195+
"expires_at",
196+
"machine_id",
197+
"inserted_at",
198+
] satisfies [string, ...string[]];
199+
179200
export const TaskEventV2Input = z.object({
180201
environment_id: z.string(),
181202
organization_id: z.string(),
@@ -204,6 +225,7 @@ export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSe
204225
return ch.insertUnsafe<TaskEventV2Input>({
205226
name: "insertTaskEventsV2",
206227
table: "trigger_dev.task_events_v2",
228+
columns: TASK_EVENT_V2_INSERT_COLUMNS,
207229
settings: {
208230
enable_json_type: 1,
209231
type_json_skip_duplicated_paths: 1,

0 commit comments

Comments
 (0)