Skip to content

Commit 0350e00

Browse files
carderneTrigger.dev RepoOps
carderne
authored and
Trigger.dev RepoOps
committed
perf(clickhouse): let writers supply task event attributes_text
Lets the task event writer supply the serialized attributes text directly instead of having the database compute it, while still computing it for writers that omit it. Mono-RevId: 77dd8468faa40afac2d94b17b0c5a61e4fe72aa5
1 parent ad821ea commit 0350e00

2 files changed

Lines changed: 150 additions & 86 deletions

File tree

‎internal-packages/clickhouse/schema/044_default_task_events_attributes_text.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
-- +goose Up
2+
3+
-- Metadata-only change. attributes_text moves from MATERIALIZED to DEFAULT so a
4+
-- writer may supply the serialized attributes itself. A writer that omits the
5+
-- column still gets toJSONString(attributes), exactly as before, so old and new
6+
-- writers can coexist against the same table.
7+
ALTER TABLE trigger_dev.task_events_v2
8+
MODIFY COLUMN IF EXISTS attributes_text String
9+
DEFAULT toJSONString(attributes);
10+
11+
-- +goose Down
12+
13+
ALTER TABLE trigger_dev.task_events_v2
14+
MODIFY COLUMN IF EXISTS attributes_text String
15+
MATERIALIZED toJSONString(attributes);

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

Lines changed: 135 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -6,53 +6,82 @@ function clickhouseDate(value: Date) {
66
return value.toISOString().replace("T", " ").replace("Z", "");
77
}
88

9+
function baseEvent(spanId: string) {
10+
const startTime = new Date("2026-09-01T10:00:00.000Z");
11+
const expiresAt = new Date(Date.now() + 90 * 24 * 60 * 60 * 1000);
12+
13+
return {
14+
environment_id: "env_attributes_text",
15+
organization_id: "org_attributes_text",
16+
project_id: "project_attributes_text",
17+
task_identifier: "attributes-text",
18+
run_id: "run_attributes_text",
19+
start_time: clickhouseDate(startTime),
20+
duration: "1000000",
21+
trace_id: "trace_attributes_text",
22+
span_id: spanId,
23+
parent_span_id: "",
24+
message: "Attributes text",
25+
kind: "SPAN",
26+
status: "OK",
27+
metadata: "{}",
28+
expires_at: clickhouseDate(expiresAt),
29+
};
30+
}
31+
32+
const columnKindSchema = z.object({
33+
name: z.string(),
34+
default_kind: z.string(),
35+
default_expression: z.string(),
36+
});
37+
38+
function readAttributesText(ch: ClickHouse) {
39+
return ch.reader.query({
40+
name: "read-task-event-attributes",
41+
query: `SELECT attributes_text,
42+
toUInt8(inserted_at > toDateTime64('2020-01-01 00:00:00', 3)) AS has_inserted_at
43+
FROM trigger_dev.task_events_v2
44+
WHERE environment_id = {environmentId: String}
45+
AND span_id = {spanId: String}`,
46+
params: z.object({ environmentId: z.string(), spanId: z.string() }),
47+
schema: z.object({ attributes_text: z.string(), has_inserted_at: z.number() }),
48+
});
49+
}
50+
51+
function readColumnKinds(ch: ClickHouse) {
52+
return ch.reader.query({
53+
name: "read-task-event-attribute-column-kinds",
54+
query: `SELECT name, default_kind, default_expression
55+
FROM system.columns
56+
WHERE database = 'trigger_dev'
57+
AND table = 'task_events_v2'
58+
AND name IN ('attributes', 'attributes_input', 'attributes_text')
59+
ORDER BY name`,
60+
schema: columnKindSchema,
61+
});
62+
}
63+
964
describe("task events v2", () => {
1065
clickhouseTest(
11-
"stores materialized attributes with explicit insert columns",
66+
"computes attributes_text when the writer omits it",
1267
async ({ clickhouseContainer }) => {
1368
const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
14-
const startTime = new Date("2026-09-01T10:00:00.000Z");
15-
const expiresAt = new Date(Date.now() + 90 * 24 * 60 * 60 * 1000);
16-
const spanId = "span_materialized_attributes";
69+
const spanId = "span_computed_attributes";
1770

1871
const [insertError] = await ch.taskEventsV2.insert([
1972
{
20-
environment_id: "env_ephemeral_attributes",
21-
organization_id: "org_ephemeral_attributes",
22-
project_id: "project_ephemeral_attributes",
23-
task_identifier: "ephemeral-attributes",
24-
run_id: "run_ephemeral_attributes",
25-
start_time: clickhouseDate(startTime),
26-
duration: "1000000",
27-
trace_id: "trace_ephemeral_attributes",
28-
span_id: spanId,
29-
parent_span_id: "",
30-
message: "Ephemeral attributes",
31-
kind: "SPAN",
32-
status: "OK",
73+
...baseEvent(spanId),
3374
attributes: {
3475
z: 1,
3576
a: "hello",
3677
nested: { enabled: true },
3778
},
38-
metadata: "{}",
39-
expires_at: clickhouseDate(expiresAt),
4079
},
4180
]);
4281
expect(insertError).toBeNull();
4382

44-
const readAttributes = ch.reader.query({
45-
name: "read-task-event-attributes",
46-
query: `SELECT attributes_text,
47-
toUInt8(inserted_at > toDateTime64('2020-01-01 00:00:00', 3)) AS has_inserted_at
48-
FROM trigger_dev.task_events_v2
49-
WHERE environment_id = {environmentId: String}
50-
AND span_id = {spanId: String}`,
51-
params: z.object({ environmentId: z.string(), spanId: z.string() }),
52-
schema: z.object({ attributes_text: z.string(), has_inserted_at: z.number() }),
53-
});
54-
const [readError, rows] = await readAttributes({
55-
environmentId: "env_ephemeral_attributes",
83+
const [readError, rows] = await readAttributesText(ch)({
84+
environmentId: "env_attributes_text",
5685
spanId,
5786
});
5887
expect(readError).toBeNull();
@@ -62,79 +91,99 @@ describe("task events v2", () => {
6291
has_inserted_at: 1,
6392
},
6493
]);
94+
}
95+
);
96+
97+
clickhouseTest(
98+
"accepts attributes_text supplied by the writer",
99+
async ({ clickhouseContainer }) => {
100+
const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
101+
const spanId = "span_supplied_attributes";
65102

66-
const readColumnKinds = ch.reader.query({
67-
name: "read-task-event-attribute-column-kinds",
68-
query: `SELECT name, default_kind, default_expression
69-
FROM system.columns
70-
WHERE database = 'trigger_dev'
71-
AND table = 'task_events_v2'
72-
AND name IN ('attributes', 'attributes_text')
73-
ORDER BY name`,
74-
schema: z.object({
75-
name: z.string(),
76-
default_kind: z.string(),
77-
default_expression: z.string(),
78-
}),
103+
// A writer that serializes attributes itself can send the column directly.
104+
// The DEFAULT expression only applies when the column is omitted.
105+
const insert = ch.writer.insertUnsafe<Record<string, unknown>>({
106+
name: "insertTaskEventsV2WithAttributesText",
107+
table: "trigger_dev.task_events_v2",
108+
columns: [
109+
"environment_id",
110+
"organization_id",
111+
"project_id",
112+
"task_identifier",
113+
"run_id",
114+
"start_time",
115+
"duration",
116+
"trace_id",
117+
"span_id",
118+
"parent_span_id",
119+
"message",
120+
"kind",
121+
"status",
122+
"attributes",
123+
"attributes_text",
124+
"metadata",
125+
"expires_at",
126+
],
127+
settings: { enable_json_type: 1 },
79128
});
80-
const [columnError, columns] = await readColumnKinds({});
81-
expect(columnError).toBeNull();
82-
expect(columns).toEqual([
83-
{
84-
name: "attributes",
85-
default_kind: "",
86-
default_expression: "",
87-
},
129+
130+
const [insertError] = await insert([
88131
{
89-
name: "attributes_text",
90-
default_kind: "MATERIALIZED",
91-
default_expression: "toJSONString(attributes)",
132+
...baseEvent(spanId),
133+
attributes: { supplied: false },
134+
attributes_text: '{"supplied":true}',
92135
},
93136
]);
137+
expect(insertError).toBeNull();
94138

95-
const readRemovedIndexes = ch.reader.query({
96-
name: "read-removed-task-event-text-indexes",
97-
query: `SELECT name
98-
FROM system.data_skipping_indices
99-
WHERE database = 'trigger_dev'
100-
AND table = 'task_events_v2'
101-
AND name IN (
102-
'idx_attributes_text_search',
103-
'idx_attributes_text',
104-
'idx_message_text_search',
105-
'message_text_search'
106-
)
107-
ORDER BY name`,
108-
schema: z.object({ name: z.string() }),
139+
const [readError, rows] = await readAttributesText(ch)({
140+
environmentId: "env_attributes_text",
141+
spanId,
109142
});
110-
const [indexError, indexes] = await readRemovedIndexes({});
111-
expect(indexError).toBeNull();
112-
expect(indexes).toEqual([]);
143+
expect(readError).toBeNull();
144+
expect(rows).toEqual([{ attributes_text: '{"supplied":true}', has_inserted_at: 1 }]);
113145
}
114146
);
115147

116-
clickhouseTest("adds an ephemeral attributes input", async ({ clickhouseContainer }) => {
148+
clickhouseTest("attributes_text is a DEFAULT column", async ({ clickhouseContainer }) => {
117149
const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
118-
const readInputColumn = ch.reader.query({
119-
name: "read-task-event-attributes-input-kind",
120-
query: `SELECT default_kind, default_expression
121-
FROM system.columns
122-
WHERE database = 'trigger_dev'
123-
AND table = 'task_events_v2'
124-
AND name = 'attributes_input'`,
125-
schema: z.object({
126-
default_kind: z.string(),
127-
default_expression: z.string(),
128-
}),
129-
});
130150

131-
const [columnError, columns] = await readInputColumn({});
151+
const [columnError, columns] = await readColumnKinds(ch)({});
132152
expect(columnError).toBeNull();
133153
expect(columns).toEqual([
154+
{ name: "attributes", default_kind: "", default_expression: "" },
134155
{
156+
name: "attributes_input",
135157
default_kind: "EPHEMERAL",
136158
default_expression: "defaultValueOfTypeName('JSON')",
137159
},
160+
{
161+
name: "attributes_text",
162+
default_kind: "DEFAULT",
163+
default_expression: "toJSONString(attributes)",
164+
},
138165
]);
139166
});
167+
168+
clickhouseTest("has no attributes text indexes", async ({ clickhouseContainer }) => {
169+
const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
170+
const readRemovedIndexes = ch.reader.query({
171+
name: "read-removed-task-event-text-indexes",
172+
query: `SELECT name
173+
FROM system.data_skipping_indices
174+
WHERE database = 'trigger_dev'
175+
AND table = 'task_events_v2'
176+
AND name IN (
177+
'idx_attributes_text_search',
178+
'idx_attributes_text',
179+
'idx_message_text_search',
180+
'message_text_search'
181+
)
182+
ORDER BY name`,
183+
schema: z.object({ name: z.string() }),
184+
});
185+
const [indexError, indexes] = await readRemovedIndexes({});
186+
expect(indexError).toBeNull();
187+
expect(indexes).toEqual([]);
188+
});
140189
});

0 commit comments

Comments
 (0)