fix: oom in snowflake worker [CM-1067] (#3948)

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/snowflake_connectors/src/consumer/transformerConsumer.ts

@@ -47,6 +47,8 @@ export class TransformerConsumer {
           log.info({ jobId: job.id, platform: job.platform, s3Path: job.s3Path }, 'Processing job')
           this.currentPollingIntervalMs = this.pollingIntervalMs
           await this.processJob(job)
+          // yield to the event loop so GC can collect the previous batch before the next one starts
+          await new Promise<void>((resolve) => setImmediate(resolve))
           continue
         }
       } catch (err) {
@@ -82,14 +84,11 @@ export class TransformerConsumer {
       const platform = job.platform as PlatformType
       const source = getDataSource(platform, job.sourceName)
 
-      const rows = await this.s3Service.readParquetRows(job.s3Path)
-
       let transformedCount = 0
       let transformSkippedCount = 0
       let resolveSkippedCount = 0
 
-      for (let i = 0; i < rows.length; i++) {
-        const row = rows[i]
+      for await (const row of this.s3Service.streamParquetRows(job.s3Path)) {
         const result = source.transformer.safeTransformRow(row)
         if (!result) {
           transformSkippedCount++

services/apps/snowflake_connectors/src/core/s3Service.ts

@@ -41,17 +41,19 @@ export class S3Service {
     return Buffer.from(byteArray)
   }
 
-  async readParquetRows(s3Uri: string): Promise<Record<string, unknown>[]> {
-    const buffer = await this.downloadFile(s3Uri)
+  async *streamParquetRows(s3Uri: string): AsyncGenerator<Record<string, unknown>> {
+    let buffer: Buffer | null = await this.downloadFile(s3Uri)
     const reader = await ParquetReader.openBuffer(buffer)
+    buffer = null
     const cursor = reader.getCursor()
-    const rows: Record<string, unknown>[] = []
-    let row: Record<string, unknown> | null = null
-    while ((row = (await cursor.next()) as Record<string, unknown> | null) !== null) {
-      rows.push(row)
+    try {
+      let row: Record<string, unknown> | null = null
+      while ((row = (await cursor.next()) as Record<string, unknown> | null) !== null) {
+        yield row
+      }
+    } finally {
+      await reader.close()
     }
-    await reader.close()
-    return rows
   }
 
   async deleteFile(s3Uri: string): Promise<void> {

services/apps/snowflake_connectors/src/integrations/tnc/courses/buildSourceQuery.ts

@@ -34,7 +34,21 @@ export const buildSourceQuery = (sinceTimestamp?: string): string => {
   let select = `
   SELECT
     ca.*,
-    co.*,
+    co.title,
+    co.course_group_id,
+    co.client_id,
+    co.sku,
+    co.status,
+    co.slug,
+    co.created_ts,
+    co.start_date,
+    co.end_date,
+    co.is_training,
+    co.is_certification,
+    co.instruction_type,
+    co.product_type,
+    co.location,
+    co.is_test_or_archived,
     tu.user_email,
     tu.lfid,
     tu.learner_name,