perf(clickhouse): stage ephemeral event attribute input

Adds a separate ephemeral input for task event attributes and
temporarily writes both attribute inputs. Existing materialization
behavior remains unchanged while the new input is deployed everywhere.

Mono-RevId: af3261bb83d40c1a6333e214f8263f096f7bad13

Author: carderne

internal-packages/clickhouse/schema/043_add_task_events_attributes_input.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+
+ALTER TABLE trigger_dev.task_events_v2
+    ADD COLUMN IF NOT EXISTS attributes_input JSON
+        EPHEMERAL defaultValueOfTypeName('JSON')
+        CODEC(ZSTD(1))
+        AFTER attributes;
+
+-- +goose Down
+
+ALTER TABLE trigger_dev.task_events_v2
+    DROP COLUMN IF EXISTS attributes_input;

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

@@ -112,4 +112,29 @@ describe("task events v2", () => {
       expect(indexes).toEqual([]);
     }
   );
+
+  clickhouseTest("adds an ephemeral attributes input", async ({ clickhouseContainer }) => {
+    const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
+    const readInputColumn = ch.reader.query({
+      name: "read-task-event-attributes-input-kind",
+      query: `SELECT default_kind, default_expression
+      FROM system.columns
+      WHERE database = 'trigger_dev'
+        AND table = 'task_events_v2'
+        AND name = 'attributes_input'`,
+      schema: z.object({
+        default_kind: z.string(),
+        default_expression: z.string(),
+      }),
+    });
+
+    const [columnError, columns] = await readInputColumn({});
+    expect(columnError).toBeNull();
+    expect(columns).toEqual([
+      {
+        default_kind: "EPHEMERAL",
+        default_expression: "defaultValueOfTypeName('JSON')",
+      },
+    ]);
+  });
 });

internal-packages/clickhouse/src/taskEvents.ts

@@ -1,6 +1,10 @@
 import type { ClickHouseSettings } from "@clickhouse/client";
 import { z } from "zod";
-import type { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
+import type {
+  ClickhouseInsertFunction,
+  ClickhouseReader,
+  ClickhouseWriter,
+} from "./client/types.js";
 
 export const TaskEventV1Input = z.object({
   environment_id: z.string(),
@@ -191,6 +195,7 @@ const TASK_EVENT_V2_INSERT_COLUMNS = [
   "kind",
   "status",
   "attributes",
+  "attributes_input",
   "metadata",
   "expires_at",
   "machine_id",
@@ -221,8 +226,22 @@ export const TaskEventV2Input = z.object({
 
 export type TaskEventV2Input = z.input<typeof TaskEventV2Input>;
 
-export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
-  return ch.insertUnsafe<TaskEventV2Input>({
+type TaskEventV2DualAttributesInput = TaskEventV2Input & {
+  attributes_input: unknown;
+};
+
+function withAttributesInput(event: TaskEventV2Input): TaskEventV2DualAttributesInput {
+  return {
+    ...event,
+    attributes_input: event.attributes,
+  };
+}
+
+export function insertTaskEventsV2(
+  ch: ClickhouseWriter,
+  settings?: ClickHouseSettings
+): ClickhouseInsertFunction<TaskEventV2Input> {
+  const insert = ch.insertUnsafe<TaskEventV2DualAttributesInput>({
     name: "insertTaskEventsV2",
     table: "trigger_dev.task_events_v2",
     columns: TASK_EVENT_V2_INSERT_COLUMNS,
@@ -235,6 +254,14 @@ export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSe
       ...settings,
     },
   });
+
+  return (events, options) => {
+    const values = Array.isArray(events)
+      ? events.map(withAttributesInput)
+      : withAttributesInput(events);
+
+    return insert(values, options);
+  };
 }
 
 export function getTraceSummaryQueryBuilderV2(ch: ClickhouseReader, settings?: ClickHouseSettings) {