fix: simplify event buffer

Author: lindesvard

packages/db/src/buffers/event-buffer.test.ts

@@ -56,14 +56,15 @@ export const TABLE_NAMES = {
   event_property_values_mv: 'event_property_values_mv',
   cohort_events_mv: 'cohort_events_mv',
   sessions: 'sessions',
+  events_imports: 'events_imports',
 };
 
 export const CLICKHOUSE_OPTIONS: NodeClickHouseClientConfigOptions = {
   max_open_connections: 30,
-  request_timeout: 60000,
+  request_timeout: 300000,
   keep_alive: {
     enabled: true,
-    idle_socket_ttl: 8000,
+    idle_socket_ttl: 60000,
   },
   compression: {
     request: true,
@@ -132,7 +133,27 @@ export const ch = new Proxy(originalCh, {
     const value = Reflect.get(target, property, receiver);
 
     if (property === 'insert') {
-      return (...args: any[]) => withRetry(() => value.apply(target, args));
+      return (...args: any[]) =>
+        withRetry(() => {
+          args[0].clickhouse_settings = {
+            // Allow bigger HTTP payloads/time to stream rows
+            async_insert: 1,
+            wait_for_async_insert: 1,
+            // Increase insert timeouts and buffer sizes for large batches
+            max_execution_time: 300,
+            max_insert_block_size: '500000',
+            max_http_get_redirects: '0',
+            // Ensure JSONEachRow stays efficient
+            input_format_parallel_parsing: 1,
+            // Keep long-running inserts/queries from idling out at proxies by sending progress headers
+            send_progress_in_http_headers: 1,
+            http_headers_progress_interval_ms: '50000',
+            // Ensure server holds the connection until the query is finished
+            wait_end_of_query: 1,
+            ...args[0].clickhouse_settings,
+          };
+          return value.apply(target, args);
+        });
     }
 
     return value;

packages/db/src/buffers/event-buffer.ts

@@ -510,7 +510,7 @@ export async function getEventList(options: GetEventListOptions) {
     sb.select.model = 'model';
   }
   if (select.duration) {
-    sb.select.duration = 'duration';
+    sb.select.duration = `${getDurationSql()} as duration`;
   }
   if (select.path) {
     sb.select.path = 'path';
@@ -570,7 +570,7 @@ export async function getEventList(options: GetEventListOptions) {
     custom(sb);
   }
 
-  console.log('getSql()', getSql());
+  console.log('getSql() ----> ', getSql());
 
   const data = await getEvents(getSql(), {
     profile: select.profile ?? true,
@@ -1016,3 +1016,6 @@ class EventService {
 }
 
 export const eventService = new EventService(ch);
+
+export const getDurationSql = (field = 'created_at') =>
+  `dateDiff('millisecond', ${field}, leadInFrame(toNullable(${field}), 1, NULL) OVER (PARTITION BY session_id ORDER BY ${field} ASC ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING))`;