perf(clickhouse): serialize task event attributes_text in the writer

Span and log attributes are now serialized by the server before being
stored, so the attribute text shown in the dashboard is exactly what
your task sent.

Mono-RevId: b789b40bd993d5096f0e10ca57554ecb0bfd12e9

Author: carderne

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

@@ -63,10 +63,10 @@ function readColumnKinds(ch: ClickHouse) {
 
 describe("task events v2", () => {
   clickhouseTest(
-    "computes attributes_text when the writer omits it",
+    "stores attributes_text serialized by the writer",
     async ({ clickhouseContainer }) => {
       const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
-      const spanId = "span_computed_attributes";
+      const spanId = "span_serialized_attributes";
 
       const [insertError] = await ch.taskEventsV2.insert([
         {
@@ -87,23 +87,23 @@ describe("task events v2", () => {
       expect(readError).toBeNull();
       expect(rows).toEqual([
         {
-          attributes_text: '{"a":"hello","nested":{"enabled":true},"z":1}',
+          attributes_text: '{"z":1,"a":"hello","nested":{"enabled":true}}',
           has_inserted_at: 1,
         },
       ]);
     }
   );
 
   clickhouseTest(
-    "accepts attributes_text supplied by the writer",
+    "computes attributes_text for a writer that omits it",
     async ({ clickhouseContainer }) => {
       const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
-      const spanId = "span_supplied_attributes";
+      const spanId = "span_computed_attributes";
 
-      // A writer that serializes attributes itself can send the column directly.
-      // The DEFAULT expression only applies when the column is omitted.
+      // An older writer that only sends the JSON column still gets attributes_text
+      // from the DEFAULT expression, so both writer generations can coexist.
       const insert = ch.writer.insertUnsafe<Record<string, unknown>>({
-        name: "insertTaskEventsV2WithAttributesText",
+        name: "insertTaskEventsV2WithoutAttributesText",
         table: "trigger_dev.task_events_v2",
         columns: [
           "environment_id",
@@ -120,7 +120,6 @@ describe("task events v2", () => {
           "kind",
           "status",
           "attributes",
-          "attributes_text",
           "metadata",
           "expires_at",
         ],
@@ -130,8 +129,7 @@ describe("task events v2", () => {
       const [insertError] = await insert([
         {
           ...baseEvent(spanId),
-          attributes: { supplied: false },
-          attributes_text: '{"supplied":true}',
+          attributes: { z: 1, a: "hello" },
         },
       ]);
       expect(insertError).toBeNull();
@@ -141,7 +139,7 @@ describe("task events v2", () => {
         spanId,
       });
       expect(readError).toBeNull();
-      expect(rows).toEqual([{ attributes_text: '{"supplied":true}', has_inserted_at: 1 }]);
+      expect(rows).toEqual([{ attributes_text: '{"a":"hello","z":1}', has_inserted_at: 1 }]);
     }
   );
 

internal-packages/clickhouse/src/taskEvents.ts

@@ -195,7 +195,7 @@ const TASK_EVENT_V2_INSERT_COLUMNS = [
   "kind",
   "status",
   "attributes",
-  "attributes_input",
+  "attributes_text",
   "metadata",
   "expires_at",
   "machine_id",
@@ -226,22 +226,33 @@ export const TaskEventV2Input = z.object({
 
 export type TaskEventV2Input = z.input<typeof TaskEventV2Input>;
 
-type TaskEventV2DualAttributesInput = TaskEventV2Input & {
-  attributes_input: unknown;
+type TaskEventV2Row = TaskEventV2Input & {
+  attributes_text: string;
 };
 
-function withAttributesInput(event: TaskEventV2Input): TaskEventV2DualAttributesInput {
+// attributes_text is serialized by the writer rather than computed by ClickHouse,
+// so the stored text is exactly what was sent and the table does not have to
+// parse and re-encode the attributes on every insert.
+export function serializeTaskEventAttributes(attributes: unknown): string {
+  if (attributes === null || attributes === undefined) {
+    return "{}";
+  }
+
+  return JSON.stringify(attributes) ?? "{}";
+}
+
+function toTaskEventV2Row(event: TaskEventV2Input): TaskEventV2Row {
   return {
     ...event,
-    attributes_input: event.attributes,
+    attributes_text: serializeTaskEventAttributes(event.attributes),
   };
 }
 
 export function insertTaskEventsV2(
   ch: ClickhouseWriter,
   settings?: ClickHouseSettings
 ): ClickhouseInsertFunction<TaskEventV2Input> {
-  const insert = ch.insertUnsafe<TaskEventV2DualAttributesInput>({
+  const insert = ch.insertUnsafe<TaskEventV2Row>({
     name: "insertTaskEventsV2",
     table: "trigger_dev.task_events_v2",
     columns: TASK_EVENT_V2_INSERT_COLUMNS,
@@ -256,9 +267,7 @@ export function insertTaskEventsV2(
   });
 
   return (events, options) => {
-    const values = Array.isArray(events)
-      ? events.map(withAttributesInput)
-      : withAttributesInput(events);
+    const values = Array.isArray(events) ? events.map(toTaskEventV2Row) : toTaskEventV2Row(events);
 
     return insert(values, options);
   };