Skip to content

perf(clickhouse): prepare task event attribute inserts #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anurag6569201 wants to merge 1 commit into
base: qa/agent-triggerdotdev-trigger-dev/pr-09-4860/base
Choose a base branch
from
Open

perf(clickhouse): prepare task event attribute inserts #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 22 additions & 6 deletions apps/webapp/app/services/admin/missingLlmModels.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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" },
{

@shipwright-agent shipwright-agent Bot Sep 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shipwright · HIGH

The migration from ClickHouse map accessor syntax (attributes.gen_ai.response.model.:String) to JSONExtractString(attributes_text, ...) changes query semantics.

Impact: The migration from ClickHouse map accessor syntax (attributes.gen_ai.response.model.:String) to JSONExtractString(attributes_text, ...) changes query semantics. If attributes_text is not populated for older rows or is populated asynchronously, these queries may silently return empty results, causing the missing-model detection to skip spans and potentially under-bill or misreport LLM usage. No backfill or data migra…

Suggested fix: Review the cited evidence, fix the risk if confirmed, and rerun Shipwright.

name: "model",
expression: "JSONExtractString(attributes_text, 'gen_ai', 'response', 'model')",
},
{
name: "system",
expression: "JSONExtractString(attributes_text, 'gen_ai', 'system')",
},
{ name: "cnt", expression: "count()" },
],
});
Expand All @@ -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" });
Expand Down Expand Up @@ -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");
Expand Down
2 changes: 1 addition & 1 deletion apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() }),
});
Expand Down
3 changes: 3 additions & 0 deletions internal-packages/clickhouse/src/client/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -1078,6 +1079,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
public insertUnsafe<TRecord extends Record<string, any>>(req: {
name: string;
table: string;
columns?: InsertParams["columns"];
settings?: ClickHouseSettings;
}): ClickhouseInsertFunction<TRecord> {
return async (events, options) => {
Expand Down Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions internal-packages/clickhouse/src/client/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -272,6 +273,7 @@ export interface ClickhouseWriter {
insertUnsafe<TRecord extends Record<string, any>>(req: {
name: string;
table: string;
columns?: InsertParams["columns"];
settings?: ClickHouseSettings;
}): ClickhouseInsertFunction<TRecord>;

Expand Down
67 changes: 67 additions & 0 deletions internal-packages/clickhouse/src/taskEvents.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 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([
{
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(startTime),
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(expiresAt),
},
]);
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}',

@shipwright-agent shipwright-agent Bot Sep 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shipwright · HIGH

The test asserts an exact JSON string for attributes_text, but JavaScript object key ordering is not guaranteed.

Impact: The test asserts an exact JSON string for attributes_text, but JavaScript object key ordering is not guaranteed. The input object { z: 1, a: 'hello', nested: { enabled: true } } may serialize in a different order across runtimes or V8 versions, causing intermittent CI failures. The test should parse attributes_text and compare objects.

Suggested fix: Review the cited evidence, fix the risk if confirmed, and rerun Shipwright.

@shipwright-agent shipwright-agent Bot Sep 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shipwright · MEDIUM

The test expects attributes_text to be '{"a":"hello","nested":{"enabled":true},"z":1}', but the input attributes object is { z: 1, a: "hello", nested: { enabled: true } }.

Impact: The test expects attributes_text to be '{"a":"hello","nested":{"enabled":true},"z":1}', but the input attributes object is { z: 1, a: "hello", nested: { enabled: true } }. JSON object key ordering is not guaranteed by the JavaScript runtime, so this assertion can fail intermittently depending on insertion order. The test should parse the JSON and compare objects rather than asserting an exact string.

Suggested fix: Fix the review finding before release.

has_inserted_at: 1,
},
]);
}
);
});
22 changes: 22 additions & 0 deletions internal-packages/clickhouse/src/taskEvents.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [

@shipwright-agent shipwright-agent Bot Sep 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shipwright · HIGH

The new TASK_EVENT_V2_INSERT_COLUMNS constant is defined as a bare array with a satisfies tuple constraint, but there is no comment explaining why explicit columns are required or

Impact: The new TASK_EVENT_V2_INSERT_COLUMNS constant is defined as a bare array with a satisfies tuple constraint, but there is no comment explaining why explicit columns are required or what happens if the table schema changes. A future maintainer adding a column to TaskEventV2Input may not realize they must also update this list, causing silent data loss or insert failures.

Suggested fix: Review the cited evidence, fix the risk if confirmed, and rerun Shipwright.

"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(),
Expand Down Expand Up @@ -204,6 +225,7 @@ export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSe
return ch.insertUnsafe<TaskEventV2Input>({
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,
Expand Down