fix: allow GC to run in between batches

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,7 +84,7 @@ export class TransformerConsumer {
       const platform = job.platform as PlatformType
       const source = getDataSource(platform, job.sourceName)
 
-      const rows = await this.s3Service.readParquetRows(job.s3Path)
+      let rows: Record<string, unknown>[] | null = await this.s3Service.readParquetRows(job.s3Path)
 
       let transformedCount = 0
       let transformSkippedCount = 0
@@ -110,6 +112,8 @@ export class TransformerConsumer {
         transformedCount++
       }
 
+      rows = null
+
       const skippedCount = transformSkippedCount + resolveSkippedCount
       const processingMetrics = {
         transformedCount,

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

@@ -42,8 +42,9 @@ export class S3Service {
   }
 
   async readParquetRows(s3Uri: string): Promise<Record<string, unknown>[]> {
-    const buffer = await this.downloadFile(s3Uri)
+    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