fix: stream row by row

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

Author: mbani01

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

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

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

@@ -41,18 +41,19 @@ export class S3Service {
     return Buffer.from(byteArray)
   }
 
-  async readParquetRows(s3Uri: string): Promise<Record<string, unknown>[]> {
+  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> {