fix(webapp,clickhouse): store task output as serialized String in ClickHouse

Task run output is stored in a new output_raw String column instead of the
native JSON output column. Deeply nested output could push the JSON column's
accumulated type past ClickHouse 26.2's input_format_binary_max_type_complexity
limit, failing the replication insert so the terminal row never landed and the
run appeared stuck.

A String has constant binary type complexity regardless of payload shape, so
the failure mode is gone for both writes and reads. TRQL path access compiles
to JSON_VALUE over output_raw, and bare reads plus full-text search use the
String column directly (a new ngram index keeps search fast). The TRQL surface
is unchanged. error and the payload column are untouched.

Author: matt-aitken

.server-changes/clickhouse-output-string.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Store task run output as serialized JSON text in ClickHouse instead of the native JSON column. Deeply nested output could exceed ClickHouse 26.2's `input_format_binary_max_type_complexity` limit, causing some runs to fail replication and appear stuck.

apps/webapp/app/services/runsReplicationService.server.ts

@@ -1061,6 +1061,11 @@ export class RunsReplicationService {
     _version: bigint
   ): Promise<TaskRunInsertArray> {
     const output = await this.#prepareJson(run.output, run.outputType);
+    // The serialized output is written to the String `output_raw` column rather than the
+    // native JSON `output` column. A String has constant binary type complexity, so deeply
+    // nested payloads can no longer breach the type-complexity ceiling and silently drop the
+    // terminal row on insert. The JSON `output` column is left empty to keep its type trivial.
+    const outputRaw = serializeJsonRaw(output.data);
     const errorData = { data: run.error };
 
     // Calculate error fingerprint for failed runs
@@ -1100,7 +1105,7 @@ export class RunsReplicationService {
       run.usageDurationMs ?? 0, // usage_duration_ms
       run.costInCents ?? 0, // cost_in_cents
       run.baseCostInCents ?? 0, // base_cost_in_cents
-      output, // output
+      { data: undefined }, // output (left empty; serialized value lives in output_raw)
       errorData, // error
       errorFingerprint, // error_fingerprint
       run.runTags ?? [], // tags
@@ -1130,6 +1135,7 @@ export class RunsReplicationService {
       annotations?.rootTriggerSource ?? "", // root_trigger_source
       annotations?.taskKind ?? "", // task_kind
       run.isWarmStart ?? null, // is_warm_start
+      outputRaw, // output_raw
     ];
   }
 
@@ -1369,3 +1375,19 @@ function lsnToUInt64(lsn: string): bigint {
   const [seg, off] = lsn.split("/");
   return (BigInt("0x" + seg) << 32n) | BigInt("0x" + off);
 }
+
+/**
+ * Serialize an already-parsed JSON value to the text stored in a `*_raw` String column.
+ * Returns an empty string when there is no value, which the query layer treats as "no data".
+ */
+function serializeJsonRaw(data: unknown): string {
+  if (data === undefined) {
+    return "";
+  }
+
+  try {
+    return JSON.stringify(data) ?? "";
+  } catch {
+    return "";
+  }
+}

apps/webapp/app/v3/querySchemas.ts

@@ -353,9 +353,11 @@ export const runsSchema: TableSchema = {
         description: "The data you returned from the task.",
         example: '{"result": "success"}',
       }),
-      nullValue: "'{}'", // Transform NULL checks to compare against empty object
-      textColumn: "output_text", // Use output_text for full JSON value queries
-      dataPrefix: "data", // Internal data is wrapped in {"data": ...}
+      // Stored as serialized JSON text in output_raw (a String) rather than the native JSON
+      // column, so reads/writes can't hit the JSON binary type-complexity ceiling.
+      nullValue: "''", // Empty raw string means "no output"
+      textColumn: "output_raw", // Full-value reads/search use the raw String column
+      rawColumn: "output_raw", // Path access (output.foo) compiles to JSON_VALUE(output_raw, '$.foo')
     },
     error: {
       name: "error",

apps/webapp/test/runsReplicationService.part3.test.ts

@@ -168,13 +168,17 @@ describe("RunsReplicationService (part 3/7)", () => {
             status: "COMPLETED_SUCCESSFULLY",
           })
         );
-        expect(found?.output).toBeDefined();
+        // Output is stored as serialized text in output_raw; the native JSON column stays empty
+        expect(found?.output).toStrictEqual({});
+        expect(found?.output_raw).toBe(`{"foo":"bar"}`);
       }
 
-      // Check the run with the bad JSON
+      // The run with the bad JSON lands with its output blanked (output_raw empty) rather than
+      // being dropped, so its terminal status is still recorded.
       const foundBad = result?.find((r: any) => r.span_id === "bulk-10");
       expect(foundBad).toBeDefined();
       expect(foundBad?.output).toStrictEqual({});
+      expect(foundBad?.output_raw).toBe("");
 
       await runsReplicationService.stop();
     }

apps/webapp/test/runsReplicationService.part6.test.ts

@@ -473,8 +473,9 @@ describe("RunsReplicationService (part 6/7)", () => {
       expect(parseClickhouseTimestamp(clickhouseRun.queued_at)).toBe(queuedAt.getTime());
       expect(parseClickhouseTimestamp(clickhouseRun.expired_at)).toBeNull();
 
-      // Output (parsed JSON)
-      expect(clickhouseRun.output).toEqual({ data: { result: "test-output" } });
+      // Output is stored as serialized JSON text in output_raw; the native JSON column is empty
+      expect(clickhouseRun.output).toEqual({});
+      expect(clickhouseRun.output_raw).toEqual(JSON.stringify({ result: "test-output" }));
 
       // Error
       expect(clickhouseRun.error).toEqual({

internal-packages/clickhouse/schema/035_add_task_runs_v2_output_raw.sql

@@ -0,0 +1,17 @@
+-- +goose Up
+-- Store task output as a serialized JSON String alongside the native JSON `output` column.
+-- A String has constant binary type complexity regardless of payload depth/width, so writes
+-- and reads can never hit input_format_binary_max_type_complexity the way the JSON type can.
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD COLUMN IF NOT EXISTS output_raw String DEFAULT '';
+
+-- Keep full-text search on output fast now that reads come from output_raw instead of output_text.
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD INDEX IF NOT EXISTS idx_output_raw output_raw TYPE ngrambf_v1 (3, 131072, 3, 0) GRANULARITY 4;
+
+-- +goose Down
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP INDEX IF EXISTS idx_output_raw;
+
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP COLUMN IF EXISTS output_raw;

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

@@ -90,6 +90,7 @@ describe("Task Runs V2", () => {
       "", // root_trigger_source
       "", // task_kind
       null, // is_warm_start
+      "", // output_raw
     ];
 
     const [insertError, insertResult] = await insert([taskRunData]);
@@ -224,6 +225,7 @@ describe("Task Runs V2", () => {
       "", // root_trigger_source
       "", // task_kind
       null, // is_warm_start
+      "", // output_raw
     ];
 
     const run2: TaskRunInsertArray = [
@@ -281,6 +283,7 @@ describe("Task Runs V2", () => {
       "", // root_trigger_source
       "", // task_kind
       null, // is_warm_start
+      "", // output_raw
     ];
 
     const [insertError, insertResult] = await insert([run1, run2]);
@@ -385,6 +388,7 @@ describe("Task Runs V2", () => {
         "", // root_trigger_source
         "", // task_kind
         null, // is_warm_start
+        "", // output_raw
       ];
 
       const [insertError, insertResult] = await insert([taskRun]);
@@ -497,6 +501,7 @@ describe("Task Runs V2", () => {
         "", // root_trigger_source
         "", // task_kind
         null, // is_warm_start
+        "", // output_raw
       ];
 
       const childA_v1: TaskRunInsertArray = [
@@ -554,6 +559,7 @@ describe("Task Runs V2", () => {
         "",
         "",
         null,
+        "", // output_raw
       ];
 
       const childA_v2: TaskRunInsertArray = [...childA_v1];
@@ -615,6 +621,7 @@ describe("Task Runs V2", () => {
         "",
         "",
         null,
+        "", // output_raw
       ];
 
       const childDeleted_v1: TaskRunInsertArray = [
@@ -672,6 +679,7 @@ describe("Task Runs V2", () => {
         "",
         "",
         null,
+        "", // output_raw
       ];
 
       const childDeleted_v2: TaskRunInsertArray = [...childDeleted_v1];

internal-packages/clickhouse/src/taskRuns.ts

@@ -28,6 +28,7 @@ export const TaskRunV2 = z.object({
   cost_in_cents: z.number().default(0),
   base_cost_in_cents: z.number().default(0),
   output: z.unknown(),
+  output_raw: z.string().default(""),
   error: z.unknown(),
   error_fingerprint: z.string().default(""),
   tags: z.array(z.string()).default([]),
@@ -117,6 +118,7 @@ export const TASK_RUN_COLUMNS = [
   "root_trigger_source",
   "task_kind",
   "is_warm_start",
+  "output_raw",
 ] as const;
 
 export type TaskRunColumnName = (typeof TASK_RUN_COLUMNS)[number];
@@ -186,6 +188,7 @@ export type TaskRunFieldTypes = {
   root_trigger_source: string;
   task_kind: string;
   is_warm_start: boolean | null;
+  output_raw: string;
 };
 
 /**
@@ -326,6 +329,7 @@ export type TaskRunInsertArray = [
   root_trigger_source: string,
   task_kind: string,
   is_warm_start: boolean | null,
+  output_raw: string,
 ];
 
 /**

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

@@ -1608,4 +1608,134 @@ describe("Field Mapping Tests", () => {
     expect(result?.rows).toHaveLength(2);
     expect(result?.rows?.map((r) => r.run_id).sort()).toEqual(["run_fm_in1", "run_fm_in2"]);
   });
+
+  describe("output stored as serialized text (rawColumn bridge)", () => {
+    // `output` is a JSON column physically stored as serialized JSON text in the String
+    // column output_raw. Path access bridges to JSON_VALUE; bare access reads the String.
+    const outputSchema: TableSchema = {
+      ...taskRunsSchema,
+      columns: {
+        ...taskRunsSchema.columns,
+        output: {
+          name: "output",
+          ...column("JSON"),
+          nullValue: "''",
+          textColumn: "output_raw",
+          rawColumn: "output_raw",
+        },
+      },
+    };
+
+    const tenant = {
+      organization_id: { op: "eq", value: "org_tenant1" },
+      project_id: { op: "eq", value: "proj_tenant1" },
+      environment_id: { op: "eq", value: "env_tenant1" },
+    } as const;
+
+    clickhouseTest(
+      "filters and selects JSON paths via JSON_VALUE over output_raw",
+      async ({ clickhouseContainer }) => {
+        const client = new ClickhouseClient({
+          name: "test",
+          url: clickhouseContainer.getConnectionUrl(),
+        });
+
+        const insert = insertTaskRuns(client, { async_insert: 0 });
+        const [insertError] = await insert([
+          createTaskRun({ run_id: "run_o1", output_raw: JSON.stringify({ foo: "bar", n: 42 }) }),
+          createTaskRun({ run_id: "run_o2", output_raw: JSON.stringify({ foo: "baz" }) }),
+          createTaskRun({ run_id: "run_o3", output_raw: "" }),
+        ]);
+        expect(insertError).toBeNull();
+
+        // Path access in SELECT + WHERE both compile to JSON_VALUE(output_raw, '$.foo')
+        const [error, result] = await executeTSQL(client, {
+          name: "test-output-path",
+          query: "SELECT run_id, output.foo AS foo FROM task_runs WHERE output.foo = 'bar'",
+          schema: z.object({ run_id: z.string(), foo: z.string() }),
+          enforcedWhereClause: tenant,
+          tableSchema: [outputSchema],
+        });
+
+        expect(error).toBeNull();
+        expect(result?.rows).toEqual([{ run_id: "run_o1", foo: "bar" }]);
+
+        // A nested scalar path works too, returning the value as a string
+        const [nError, nResult] = await executeTSQL(client, {
+          name: "test-output-path-number",
+          query: "SELECT output.n AS n FROM task_runs WHERE run_id = 'run_o1'",
+          schema: z.object({ n: z.string() }),
+          enforcedWhereClause: tenant,
+          tableSchema: [outputSchema],
+        });
+        expect(nError).toBeNull();
+        expect(nResult?.rows).toEqual([{ n: "42" }]);
+      }
+    );
+
+    clickhouseTest("reads the bare value from output_raw", async ({ clickhouseContainer }) => {
+      const client = new ClickhouseClient({
+        name: "test",
+        url: clickhouseContainer.getConnectionUrl(),
+      });
+
+      const insert = insertTaskRuns(client, { async_insert: 0 });
+      const [insertError] = await insert([
+        createTaskRun({ run_id: "run_b1", output_raw: JSON.stringify({ foo: "baz" }) }),
+        createTaskRun({ run_id: "run_b2", output_raw: "" }),
+      ]);
+      expect(insertError).toBeNull();
+
+      // Bare selection reads the raw String column directly
+      const [error, result] = await executeTSQL(client, {
+        name: "test-output-bare",
+        query: "SELECT run_id, output FROM task_runs WHERE run_id = 'run_b1'",
+        schema: z.object({ run_id: z.string(), output: z.string() }),
+        enforcedWhereClause: tenant,
+        tableSchema: [outputSchema],
+      });
+      expect(error).toBeNull();
+      expect(result?.rows).toEqual([{ run_id: "run_b1", output: JSON.stringify({ foo: "baz" }) }]);
+
+      // An empty output_raw is treated as "no output" by IS NULL
+      const [nullError, nullResult] = await executeTSQL(client, {
+        name: "test-output-is-null",
+        query: "SELECT run_id FROM task_runs WHERE output IS NULL",
+        schema: z.object({ run_id: z.string() }),
+        enforcedWhereClause: tenant,
+        tableSchema: [outputSchema],
+      });
+      expect(nullError).toBeNull();
+      expect(nullResult?.rows).toEqual([{ run_id: "run_b2" }]);
+    });
+
+    clickhouseTest("full-text search matches against output_raw", async ({ clickhouseContainer }) => {
+      const client = new ClickhouseClient({
+        name: "test",
+        url: clickhouseContainer.getConnectionUrl(),
+      });
+
+      const insert = insertTaskRuns(client, { async_insert: 0 });
+      const [insertError] = await insert([
+        createTaskRun({
+          run_id: "run_s1",
+          output_raw: JSON.stringify({ message: "boom happened here" }),
+        }),
+        createTaskRun({ run_id: "run_s2", output_raw: JSON.stringify({ message: "all good" }) }),
+        createTaskRun({ run_id: "run_s3", output_raw: "" }),
+      ]);
+      expect(insertError).toBeNull();
+
+      // Bare LIKE compiles to like(output_raw, ...) and matches the serialized text
+      const [error, result] = await executeTSQL(client, {
+        name: "test-output-like",
+        query: "SELECT run_id FROM task_runs WHERE output LIKE '%boom%'",
+        schema: z.object({ run_id: z.string() }),
+        enforcedWhereClause: tenant,
+        tableSchema: [outputSchema],
+      });
+      expect(error).toBeNull();
+      expect(result?.rows).toEqual([{ run_id: "run_s1" }]);
+    });
+  });
 });