improv(webapp): Add new table for optimized logs search

apps/webapp/app/presenters/v3/LogsListPresenter.server.ts

@@ -84,13 +84,13 @@ export type LogsListAppliedFilters = LogsList["filters"];
 // Cursor is a base64 encoded JSON of the pagination keys
 type LogCursor = {
   environmentId: string;
-  unixTimestamp: number;
+  triggeredTimestamp: string; // DateTime64(9) string
   traceId: string;
 };
 
 const LogCursorSchema = z.object({
   environmentId: z.string(),
-  unixTimestamp: z.number(),
+  triggeredTimestamp: z.string(),
   traceId: z.string(),
 });
 
@@ -252,7 +252,7 @@ export class LogsListPresenter extends BasePresenter {
       );
     }
 
-    const queryBuilder = this.clickhouse.taskEventsV2.logsListQueryBuilder();
+    const queryBuilder = this.clickhouse.taskEventsSearch.logsListQueryBuilder();
 
     queryBuilder.where("environment_id = {environmentId: String}", {
       environmentId,
@@ -350,38 +350,30 @@ export class LogsListPresenter extends BasePresenter {
     }
 
     // Debug logs are available only to admins
-    if (includeDebugLogs === false) {
-      queryBuilder.where("kind NOT IN {debugKinds: Array(String)}", {
-        debugKinds: ["DEBUG_EVENT"],
-      });
-
-      queryBuilder.where("NOT ((kind = 'LOG_INFO') AND (attributes_text = '{}'))");
-    }
-
-    queryBuilder.where("kind NOT IN {debugSpans: Array(String)}", {
-      debugSpans: ["SPAN", "ANCESTOR_OVERRIDE", "SPAN_EVENT"],
-    });
-
-    // kindCondition += ` `;
-    // params["excluded_statuses"] = ["SPAN", "ANCESTOR_OVERRIDE", "SPAN_EVENT"];
-
+    // if (includeDebugLogs === false) {
+    //   queryBuilder.where("kind NOT IN {debugKinds: Array(String)}", {
+    //     debugKinds: ["DEBUG_EVENT"],
+    //   });
+    //
+    //   queryBuilder.where("NOT ((kind = 'LOG_INFO') AND (attributes_text = '{}'))");
+    // }
 
-    queryBuilder.where("NOT (kind = 'SPAN' AND status = 'PARTIAL')");
+    // SPAN, ANCESTOR_OVERRIDE, SPAN_EVENT kinds are already filtered out by the materialized view
 
     // Cursor pagination
     const decodedCursor = cursor ? decodeCursor(cursor) : null;
     if (decodedCursor) {
       queryBuilder.where(
-        "(environment_id, toUnixTimestamp(start_time), trace_id) < ({cursorEnvId: String}, {cursorUnixTimestamp: Int64}, {cursorTraceId: String})",
+        "(environment_id, triggered_timestamp, trace_id) < ({cursorEnvId: String}, {cursorTriggeredTimestamp: String}, {cursorTraceId: String})",
         {
           cursorEnvId: decodedCursor.environmentId,
-          cursorUnixTimestamp: decodedCursor.unixTimestamp,
+          cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp,
           cursorTraceId: decodedCursor.traceId,
         }
       );
     }
 
-    queryBuilder.orderBy("environment_id DESC, toUnixTimestamp(start_time) DESC, trace_id DESC");
+    queryBuilder.orderBy("environment_id DESC, triggered_timestamp DESC, trace_id DESC");
     // Limit + 1 to check if there are more results
     queryBuilder.limit(pageSize + 1);
 
@@ -399,10 +391,9 @@ export class LogsListPresenter extends BasePresenter {
     let nextCursor: string | undefined;
     if (hasMore && logs.length > 0) {
       const lastLog = logs[logs.length - 1];
-      const unixTimestamp = Math.floor(new Date(lastLog.start_time).getTime() / 1000);
       nextCursor = encodeCursor({
         environmentId,
-        unixTimestamp,
+        triggeredTimestamp: lastLog.triggered_timestamp,
         traceId: lastLog.trace_id,
       });
     }

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.spans.tsx

@@ -63,7 +63,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   }
 
   // Query ClickHouse for related spans in the same trace
-  const queryBuilder = clickhouseClient.taskEventsV2.logsListQueryBuilder();
+  const queryBuilder = clickhouseClient.taskEventsSearch.logsListQueryBuilder();
 
   queryBuilder.where("environment_id = {environmentId: String}", {
     environmentId: environment.id,

internal-packages/clickhouse/schema/016_add_task_events_search_v1.sql

@@ -0,0 +1,62 @@
+-- +goose Up
+CREATE TABLE IF NOT EXISTS trigger_dev.task_events_search_v1
+(
+  environment_id String,
+  organization_id String,
+  project_id String,
+  triggered_timestamp DateTime64(9) CODEC(Delta(8), ZSTD(1)),
+  trace_id String CODEC(ZSTD(1)),
+  span_id String CODEC(ZSTD(1)),
+  run_id String CODEC(ZSTD(1)),
+  task_identifier String CODEC(ZSTD(1)),
+  start_time DateTime64(9) CODEC(Delta(8), ZSTD(1)),
+  inserted_at DateTime64(3),
+  message String CODEC(ZSTD(1)),
+  kind LowCardinality(String) CODEC(ZSTD(1)),
+  status LowCardinality(String) CODEC(ZSTD(1)),
+  duration UInt64 CODEC(ZSTD(1)),
+  parent_span_id String CODEC(ZSTD(1)),
+  attributes_text String CODEC(ZSTD(1)),
+
+  INDEX idx_run_id run_id TYPE bloom_filter(0.001) GRANULARITY 1,
+  INDEX idx_message_text_search lower(message) TYPE ngrambf_v1(3, 32768, 2, 0) GRANULARITY 1,
+  INDEX idx_attributes_text_search lower(attributes_text) TYPE ngrambf_v1(3, 32768, 2, 0) GRANULARITY 1
+)
+ENGINE = MergeTree
+PARTITION BY toDate(triggered_timestamp)
+ORDER BY (organization_id, environment_id, triggered_timestamp)
+--Right now we have maximum retention of up to 30 days based on plan.
+--We put a logical limit for now, the 90 DAY TTL is just a backup
+--This might need to be updated for longer retention periods
+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_v1
+TO trigger_dev.task_events_search_v1 AS
+SELECT
+  environment_id,
+  organization_id,
+  project_id,
+  trace_id,
+  span_id,
+  run_id,
+  task_identifier,
+  start_time,
+  inserted_at,
+  message,
+  kind,
+  status,
+  duration,
+  parent_span_id,
+  toJSONString(attributes) AS attributes_text,
+  fromUnixTimestamp64Nano(toUnixTimestamp64Nano(start_time) + toInt64(duration)) AS triggered_timestamp
+FROM trigger_dev.task_events_v2
+WHERE
+    kind != 'DEBUG_EVENT'
+    AND status != 'PARTIAL'
+    AND NOT (kind = 'SPAN_EVENT' AND attributes_text = '{}')
+    AND message != 'trigger.dev/start';
+
+-- +goose Down
+DROP VIEW IF EXISTS trigger_dev.task_events_search_mv_v1;
+DROP TABLE IF EXISTS trigger_dev.task_events_search_v1;

internal-packages/clickhouse/src/index.ts

@@ -23,8 +23,8 @@ import {
   getTraceSummaryQueryBuilderV2,
   insertTaskEvents,
   insertTaskEventsV2,
-  getLogsListQueryBuilderV2,
   getLogDetailQueryBuilderV2,
+  getLogsSearchListQueryBuilder,
 } from "./taskEvents.js";
 import { Logger, type LogLevel } from "@trigger.dev/core/logger";
 import type { Agent as HttpAgent } from "http";
@@ -220,8 +220,13 @@ export class ClickHouse {
       traceSummaryQueryBuilder: getTraceSummaryQueryBuilderV2(this.reader),
       traceDetailedSummaryQueryBuilder: getTraceDetailedSummaryQueryBuilderV2(this.reader),
       spanDetailsQueryBuilder: getSpanDetailsQueryBuilderV2(this.reader),
-      logsListQueryBuilder: getLogsListQueryBuilderV2(this.reader, this.logsQuerySettings?.list),
       logDetailQueryBuilder: getLogDetailQueryBuilderV2(this.reader, this.logsQuerySettings?.detail),
     };
   }
+
+  get taskEventsSearch() {
+    return {
+      logsListQueryBuilder: getLogsSearchListQueryBuilder(this.reader, this.logsQuerySettings?.list),
+    };
+  }
 }

internal-packages/clickhouse/src/taskEvents.ts

@@ -231,11 +231,12 @@ export function getSpanDetailsQueryBuilderV2(
   });
 }
 
+
 // ============================================================================
-// Logs List Query Builders (for aggregated logs page)
+// Search Table Query Builders (for logs page, using task_events_search_v1)
 // ============================================================================
 
-export const LogsListResult = z.object({
+export const LogsSearchListResult = z.object({
   environment_id: z.string(),
   organization_id: z.string(),
   project_id: z.string(),
@@ -250,14 +251,18 @@ export const LogsListResult = z.object({
   status: z.string(),
   duration: z.number().or(z.string()),
   attributes_text: z.string(),
+  triggered_timestamp: z.string(),
 });
 
-export type LogsListResult = z.output<typeof LogsListResult>;
+export type LogsSearchListResult = z.output<typeof LogsSearchListResult>;
 
-export function getLogsListQueryBuilderV2(ch: ClickhouseReader, settings?: ClickHouseSettings) {
-  return ch.queryBuilderFast<LogsListResult>({
-    name: "getLogsList",
-    table: "trigger_dev.task_events_v2",
+export function getLogsSearchListQueryBuilder(
+  ch: ClickhouseReader,
+  settings?: ClickHouseSettings
+) {
+  return ch.queryBuilderFast<LogsSearchListResult>({
+    name: "getLogsSearchList",
+    table: "trigger_dev.task_events_search_v1",
     columns: [
       "environment_id",
       "organization_id",
@@ -268,11 +273,12 @@ export function getLogsListQueryBuilderV2(ch: ClickhouseReader, settings?: Click
       "trace_id",
       "span_id",
       "parent_span_id",
-      { name: "message", expression: "LEFT(message, 512)" },
+      { name: "message", expression: "LEFT(message, 256)" },
       "kind",
       "status",
       "duration",
-      "attributes_text"
+      "attributes_text",
+      "triggered_timestamp",
     ],
     settings,
   });

references/seed/src/trigger/spanSpammer.ts

@@ -1,4 +1,4 @@
-import { logger, task, wait } from "@trigger.dev/sdk/v3";
+import { logger, task, wait, metadata } from "@trigger.dev/sdk/v3";
 
 const CONFIG = {
   delayBetweenBatchesSeconds: 0.2,
@@ -14,6 +14,22 @@ export const SpanSpammerTask = task({
     const context = { payload, ctx };
     let logCount = 0;
 
+    // 30s trace with events every 5s
+    await logger.trace("10s-span", async () => {
+      const totalSeconds = 10;
+      const intervalSeconds = 2;
+      const totalEvents = totalSeconds / intervalSeconds;
+
+      logger.info("Starting 30s span", context);
+
+      for (let i = 1; i <= totalEvents; i++) {
+        await wait.for({ seconds: intervalSeconds });
+        logger.info(`Inner event ${i}/${totalEvents} at ${i * intervalSeconds}s`, context);
+      }
+
+      logger.info("Completed 30s span", context);
+    });
+
     logger.info("Starting span spammer task", context);
     logger.warn("This will generate a lot of logs", context);
 
@@ -36,6 +52,14 @@ export const SpanSpammerTask = task({
       emitBatch("This is a test log!!! Log number: ");
     }
 
+    metadata.parent.set("childStatus", "running");
+    metadata.parent.increment("completedChildren", 1);
+
+    // Update the root run's metadata (top-level run in the chain)
+    metadata.root.set("deepChildStatus", "done");
+    metadata.root.append("completedTasks", "child-task");
+
+
     logger.info("Completed span spammer task", context);
     return { message: `Created ${logCount} logs` };
   },