perf(clickhouse): stop storing native event attributes

Author: carderne

internal-packages/clickhouse/schema/042_reduce_task_events_v2_storage_overhead.sql

@@ -0,0 +1,35 @@
+-- +goose Up
+
+-- Full-text search is served outside the source event table. Keeping these
+-- indexes here adds work to every event insert and merge without serving reads.
+ALTER TABLE trigger_dev.task_events_v2
+    DROP INDEX IF EXISTS idx_attributes_text_search;
+
+ALTER TABLE trigger_dev.task_events_v2
+    DROP INDEX IF EXISTS idx_message_text_search;
+
+-- attributes remains an insert input for attributes_text, but is no longer
+-- stored. Writers must include attributes in an explicit insert column list
+-- because implicit INSERT column lists exclude EPHEMERAL columns.
+ALTER TABLE trigger_dev.task_events_v2
+    MODIFY COLUMN attributes JSON EPHEMERAL;
+
+-- +goose Down
+
+-- ClickHouse 26.2 preserves EPHEMERAL when only the type or codec is modified.
+-- Replace the input column while keeping the stored attributes_text values.
+ALTER TABLE trigger_dev.task_events_v2
+    RENAME COLUMN attributes TO attributes_ephemeral,
+    ADD COLUMN attributes JSON CODEC(ZSTD(1)) AFTER status,
+    MODIFY COLUMN attributes_text String MATERIALIZED toJSONString(attributes),
+    DROP COLUMN attributes_ephemeral;
+
+ALTER TABLE trigger_dev.task_events_v2
+    ADD INDEX IF NOT EXISTS idx_attributes_text_search lower(attributes_text)
+    TYPE ngrambf_v1(3, 32768, 2, 0)
+    GRANULARITY 1;
+
+ALTER TABLE trigger_dev.task_events_v2
+    ADD INDEX IF NOT EXISTS idx_message_text_search lower(message)
+    TYPE ngrambf_v1(3, 32768, 2, 0)
+    GRANULARITY 1;

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

@@ -62,6 +62,121 @@ describe("task events v2", () => {
         },
       ]);
 
+      const readColumnKinds = ch.reader.query({
+        name: "read-task-event-attribute-column-kinds",
+        query: `SELECT name, default_kind, default_expression
+        FROM system.columns
+        WHERE database = 'trigger_dev'
+          AND table = 'task_events_v2'
+          AND name IN ('attributes', 'attributes_text')
+        ORDER BY name`,
+        schema: z.object({
+          name: z.string(),
+          default_kind: z.string(),
+          default_expression: z.string(),
+        }),
+      });
+      const [columnError, columns] = await readColumnKinds({});
+      expect(columnError).toBeNull();
+      expect(columns).toEqual([
+        {
+          name: "attributes",
+          default_kind: "EPHEMERAL",
+          default_expression: "defaultValueOfTypeName('JSON')",
+        },
+        {
+          name: "attributes_text",
+          default_kind: "MATERIALIZED",
+          default_expression: "toJSONString(attributes)",
+        },
+      ]);
+
+      const readRemovedIndexes = ch.reader.query({
+        name: "read-removed-task-event-text-indexes",
+        query: `SELECT name
+        FROM system.data_skipping_indices
+        WHERE database = 'trigger_dev'
+          AND table = 'task_events_v2'
+          AND name IN ('idx_attributes_text_search', 'idx_message_text_search')
+        ORDER BY name`,
+        schema: z.object({ name: z.string() }),
+      });
+      const [indexError, indexes] = await readRemovedIndexes({});
+      expect(indexError).toBeNull();
+      expect(indexes).toEqual([]);
+    }
+  );
+
+  clickhouseTest(
+    "restores stored attributes for implicit inserts on rollback",
+    async ({ clickhouseClient }) => {
+      const table = "trigger_dev.task_events_ephemeral_rollback_test";
+      await clickhouseClient.command({
+        query: `CREATE TABLE ${table}
+          (
+            id UInt64,
+            attributes JSON EPHEMERAL,
+            attributes_text String MATERIALIZED toJSONString(attributes)
+          )
+          ENGINE = MergeTree
+          ORDER BY id`,
+      });
+
+      try {
+        await clickhouseClient.insert({
+          table,
+          columns: ["id", "attributes"],
+          format: "JSONEachRow",
+          values: [{ id: 1, attributes: { before: "rollback" } }],
+        });
+
+        await clickhouseClient.command({
+          query: `ALTER TABLE ${table}
+            RENAME COLUMN attributes TO attributes_ephemeral,
+            ADD COLUMN attributes JSON CODEC(ZSTD(1)) AFTER id,
+            MODIFY COLUMN attributes_text String MATERIALIZED toJSONString(attributes),
+            DROP COLUMN attributes_ephemeral`,
+        });
+
+        await clickhouseClient.insert({
+          table,
+          format: "JSONEachRow",
+          values: [{ id: 2, attributes: { after: "rollback" } }],
+        });
+
+        const columnsResult = await clickhouseClient.query({
+          query: `SELECT default_kind, compression_codec
+            FROM system.columns
+            WHERE database = 'trigger_dev'
+              AND table = 'task_events_ephemeral_rollback_test'
+              AND name = 'attributes'`,
+          format: "JSONEachRow",
+        });
+        expect(await columnsResult.json()).toEqual([
+          { default_kind: "", compression_codec: "CODEC(ZSTD(1))" },
+        ]);
+
+        const rowsResult = await clickhouseClient.query({
+          query: `SELECT id, attributes_text, toJSONString(attributes) AS attributes_json
+            FROM ${table}
+            ORDER BY id`,
+          format: "JSONEachRow",
+        });
+        expect(await rowsResult.json()).toEqual([
+          {
+            id: 1,
+            attributes_text: '{"before":"rollback"}',
+            attributes_json: "{}",
+          },
+          {
+            id: 2,
+            attributes_text: '{"after":"rollback"}',
+            attributes_json: '{"after":"rollback"}',
+          },
+        ]);
+      } finally {
+        await clickhouseClient.command({ query: `DROP TABLE IF EXISTS ${table}` });
+      }
     }
   );
 });