add: max_bytes_before_external_group_by and max_bytes_before_external_sort and csv logging per byte

Author: ayushjhanwar-png

apps/worker/src/jobs/import.ts

@@ -173,7 +173,18 @@ export async function importJob(job: Job<ImportQueuePayload>) {
             ) => providerInstance!.transformEvent(event),
           );
 
-          await insertImportBatch(transformedEvents, importId);
+          const result = await insertImportBatch(transformedEvents, importId);
+
+          // Log actual CSV data size sent to ClickHouse
+          const csvSizeMB = result.csvDataSizeBytes
+            ? (result.csvDataSizeBytes / (1024 * 1024)).toFixed(2)
+            : 'N/A';
+
+          jobLogger.info('Inserted batch into ClickHouse', {
+            batchSize: result.insertedEvents,
+            csvDataSizeMB: csvSizeMB,
+            csvDataSizeBytes: result.csvDataSizeBytes || 0,
+          });
 
           processedEvents += eventBatch.length;
           eventBatch = [];
@@ -211,7 +222,18 @@ export async function importJob(job: Job<ImportQueuePayload>) {
           ) => providerInstance!.transformEvent(event),
         );
 
-        await insertImportBatch(transformedEvents, importId);
+        const result = await insertImportBatch(transformedEvents, importId);
+
+        // Log actual CSV data size sent to ClickHouse
+        const csvSizeMB = result.csvDataSizeBytes
+          ? (result.csvDataSizeBytes / (1024 * 1024)).toFixed(2)
+          : 'N/A';
+
+        jobLogger.info('Inserted final batch into ClickHouse', {
+          batchSize: result.insertedEvents,
+          csvDataSizeMB: csvSizeMB,
+          csvDataSizeBytes: result.csvDataSizeBytes || 0,
+        });
 
         processedEvents += eventBatch.length;
         eventBatch = [];

packages/db/src/clickhouse/client.ts

@@ -288,6 +288,10 @@ export async function chInsertCSV(tableName: string, rows: string[]) {
         clickhouse_settings: {
           format_csv_allow_double_quotes: 1,
           format_csv_allow_single_quotes: 0,
+          // Memory optimization: prevent OOM by spilling to disk
+          max_memory_usage: 40000000000, // 40GB max per query
+          max_bytes_before_external_group_by: 20000000000, // 20GB - spill GROUP BY to disk
+          max_bytes_before_external_sort: 20000000000, // 20GB - spill ORDER BY to disk
         },
       });
     }

packages/db/src/services/import.service.ts

@@ -16,6 +16,7 @@ export interface ImportStageResult {
   importId: string;
   totalEvents: number;
   insertedEvents: number;
+  csvDataSizeBytes?: number;
 }
 
 export interface ImportProgress {
@@ -78,6 +79,9 @@ export async function insertImportBatch(
     return fields.join(',');
   });
 
+  // Calculate actual CSV data size being sent to ClickHouse
+  const csvDataSize = csvRows.reduce((sum, row) => sum + row.length + 1, 0);
+
   await chInsertCSV(TABLE_NAMES.events_imports, csvRows);
 
   // Explicitly release memory
@@ -87,6 +91,7 @@ export async function insertImportBatch(
     importId,
     totalEvents: events.length,
     insertedEvents: events.length,
+    csvDataSizeBytes: csvDataSize,
   };
 }
 
@@ -209,6 +214,12 @@ export async function createSessionsStartEndEvents(
     query: sessionEventsQuery,
     query_params: { importId, from },
     format: 'JSONEachRow',
+    clickhouse_settings: {
+      // Memory optimization: prevent OOM on large GROUP BY operations
+      max_memory_usage: 40000000000, // 40GB max per query
+      max_bytes_before_external_group_by: 20000000000, // 20GB - spill GROUP BY to disk
+      max_bytes_before_external_sort: 20000000000, // 20GB - spill ORDER BY to disk
+    },
   });
 
   const sessionData = (await sessionEventsResult.json()) as Array<{
@@ -508,6 +519,10 @@ export async function moveImportsToProduction(
       send_progress_in_http_headers: 1,
       // The interval of sending these progress headers. Here it is less than 60s,
       http_headers_progress_interval_ms: '50000',
+      // Memory optimization: prevent OOM by spilling to disk
+      max_memory_usage: 40000000000, // 40GB max per query
+      max_bytes_before_external_group_by: 20000000000, // 20GB - spill GROUP BY to disk
+      max_bytes_before_external_sort: 20000000000, // 20GB - spill ORDER BY to disk
     },
   });
   console.log('[Phase 4] Migration completed successfully');
@@ -633,6 +648,10 @@ export async function backfillSessionsToProduction(
       send_progress_in_http_headers: 1,
       // The interval of sending these progress headers. Here it is less than 60s,
       http_headers_progress_interval_ms: '50000',
+      // Memory optimization: prevent OOM by spilling to disk
+      max_memory_usage: 40000000000, // 40GB max per query
+      max_bytes_before_external_group_by: 20000000000, // 20GB - spill GROUP BY to disk
+      max_bytes_before_external_sort: 20000000000, // 20GB - spill ORDER BY to disk
     },
   });
   console.log('[Phase 5] Sessions backfill completed successfully');