refactor(clickhouse): consolidate log search v2 schema migration

Create the scheduled-projector schema directly in migration 038 and remove the intermediate migration.

Author: carderne

internal-packages/clickhouse/schema/038_add_task_events_search_v2.sql

@@ -1,8 +1,9 @@
 -- +goose Up
--- Search v2 keeps the dedicated event-time search boundary, but stores only bounded,
--- normalized searchable text and fields needed by the list. The materialized view is
--- intentionally forward-only. Any historical backfill must be run as a separate,
--- throttled operation.
+-- Search v2 stores bounded normalized text outside the task_events_v2 insert path.
+-- The source index supports closed projector windows on newly written parts.
+ALTER TABLE trigger_dev.task_events_v2
+  ADD INDEX IF NOT EXISTS idx_inserted_at_projector inserted_at TYPE minmax GRANULARITY 1;
+
 CREATE TABLE IF NOT EXISTS trigger_dev.task_events_search_v2
 (
   environment_id String,
@@ -22,63 +23,29 @@ CREATE TABLE IF NOT EXISTS trigger_dev.task_events_search_v2
   status LowCardinality(String) CODEC(ZSTD(1)),
   duration UInt64 CODEC(ZSTD(1)),
   parent_span_id String CODEC(ZSTD(1)),
+  projection_fingerprint UInt128 DEFAULT reinterpretAsUInt128(
+    sipHash128(trace_id, span_id, run_id, start_time)
+  ),
 
   INDEX idx_run_id run_id TYPE bloom_filter(0.001) GRANULARITY 1,
   INDEX idx_search_text search_text
     TYPE text(tokenizer = 'ngrams', preprocessor = lowerUTF8(search_text))
 )
-ENGINE = MergeTree
+ENGINE = ReplacingMergeTree
 PARTITION BY toDate(triggered_timestamp)
-ORDER BY (organization_id, environment_id, triggered_timestamp, trace_id, span_id)
-TTL toDateTime(triggered_timestamp) + INTERVAL 90 DAY
-SETTINGS ttl_only_drop_parts = 1;
-
-CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.task_events_search_mv_v2
-TO trigger_dev.task_events_search_v2 AS
-SELECT
-  environment_id,
+ORDER BY (
   organization_id,
-  project_id,
-  least(
-    fromUnixTimestamp64Nano(toUnixTimestamp64Nano(start_time) + toInt64(duration)),
-    now64(9) + INTERVAL 5 MINUTE
-  ) AS triggered_timestamp,
+  environment_id,
+  triggered_timestamp,
   trace_id,
   span_id,
-  run_id,
-  task_identifier,
-  start_time,
-  inserted_at,
-  message,
-  substring(JSONExtractString(attributes_text, 'error', 'message'), 1, 2048) AS error_message,
-  replaceRegexpAll(
-    lowerUTF8(
-      substring(
-        concat(
-          substring(message, 1, 2048),
-          ' ',
-          replaceAll(substring(attributes_text, 1, 6144), '\\/', '/')
-        ),
-        1,
-        8192
-      )
-    ),
-    '[^\\p{L}\\p{N}_./:@+-]+',
-    ' '
-  ) AS search_text,
-  kind,
-  status,
-  duration,
-  parent_span_id
-FROM trigger_dev.task_events_v2
-WHERE
-  trace_id != ''
-  AND kind != 'DEBUG_EVENT'
-  AND status != 'PARTIAL'
-  AND NOT (kind = 'SPAN_EVENT' AND attributes_text = '{}')
-  AND kind != 'ANCESTOR_OVERRIDE'
-  AND message != 'trigger.dev/start';
+  projection_fingerprint
+)
+TTL toDateTime(triggered_timestamp) + INTERVAL 90 DAY
+SETTINGS ttl_only_drop_parts = 1;
 
 -- +goose Down
-DROP VIEW IF EXISTS trigger_dev.task_events_search_mv_v2;
 DROP TABLE IF EXISTS trigger_dev.task_events_search_v2;
+
+ALTER TABLE trigger_dev.task_events_v2
+  DROP INDEX IF EXISTS idx_inserted_at_projector;

internal-packages/clickhouse/schema/039_schedule_task_events_search_v2.sql

@@ -93,20 +93,13 @@ describe("task events search v2", () => {
         name: "read-search-v2-table-engine",
         query: `SELECT name, engine FROM system.tables
           WHERE database = 'trigger_dev'
-            AND name IN (
-              'task_events_search_mv_v2',
-              'task_events_search_v2',
-              'task_events_search_v2_insert_triggered'
-            )
+            AND name IN ('task_events_search_mv_v2', 'task_events_search_v2')
           ORDER BY name`,
         schema: z.object({ name: z.string(), engine: z.string() }),
       });
       const [tableError, tables] = await tableQuery({});
       expect(tableError).toBeNull();
-      expect(tables).toEqual([
-        { name: "task_events_search_v2", engine: "ReplacingMergeTree" },
-        { name: "task_events_search_v2_insert_triggered", engine: "MergeTree" },
-      ]);
+      expect(tables).toEqual([{ name: "task_events_search_v2", engine: "ReplacingMergeTree" }]);
 
       const firstProjection = await project(ch, start, end);
       const retryProjection = await project(ch, start, end);