file-based backpressure

Author: ptiurin

test/integration/v2/stream.test.ts

@@ -2,6 +2,9 @@ import exp from "node:constants";
 import { Firebolt } from "../../../src";
 import stream, { TransformCallback } from "node:stream";
 import BigNumber from "bignumber.js";
+import fs from "node:fs";
+import path from "node:path";
+import os from "node:os";
 
 const connectionParams = {
   auth: {
@@ -397,21 +400,71 @@ describe("streams", () => {
       }
     });
 
-    // Create a moderate backpressure stream
+    // Create a batch processing stream that writes to a temporary file
     let processedChunks = 0;
+    const batchSize = 100;
+    let batchBuffer: string[] = [];
+
+    // Create a temporary file for writing
+    const tempDir = os.tmpdir();
+    const tempFilePath = path.join(
+      tempDir,
+      `firebolt-stream-test-${Date.now()}.jsonl`
+    );
+    const fileWriteStream = fs.createWriteStream(tempFilePath, { flags: "w" });
+
     const outputStream = new stream.Writable({
-      highWaterMark: 1,
-      write(chunk, encoding, callback) {
+      highWaterMark: 1, // Reasonable buffer size
+      write(chunk: Buffer, encoding, callback) {
         processedChunks++;
+        const chunkStr = chunk.toString();
+        batchBuffer.push(chunkStr);
+
+        // Process in batches to create natural backpressure patterns
+        if (batchBuffer.length >= batchSize) {
+          // Process the batch synchronously (simulate some work)
+          const batchData = batchBuffer.join("");
+          const lines = batchData.split("\n").filter(line => line.trim());
+
+          // Write valid JSON lines to the file
+          for (const line of lines) {
+            try {
+              JSON.parse(line); // Verify it's valid JSON
+              fileWriteStream.write(line + "\n");
+            } catch (e) {
+              // Skip invalid JSON lines
+            }
+          }
 
-        // Simulate occasional slow processing with event loop yielding
-        if (processedChunks % 1000 === 0) {
-          setImmediate(() => {
-            callback();
-          });
-        } else {
-          callback();
+          // Clear the batch
+          batchBuffer = [];
+        }
+
+        callback();
+      },
+
+      final(callback) {
+        // Process any remaining items in the final batch
+        if (batchBuffer.length > 0) {
+          const batchData = batchBuffer.join("");
+          const lines = batchData.split("\n").filter(line => line.trim());
+
+          // Write remaining valid JSON lines to the file
+          for (const line of lines) {
+            try {
+              JSON.parse(line); // Verify it's valid JSON
+              fileWriteStream.write(line + "\n");
+            } catch (e) {
+              // Skip invalid JSON lines
+            }
+          }
+          batchBuffer = [];
         }
+
+        // Close the file stream
+        fileWriteStream.end(() => {
+          callback();
+        });
       }
     });
 
@@ -422,6 +475,26 @@ describe("streams", () => {
     expect(rowCount).toBe(seriesNum);
     expect(processedChunks).toBeGreaterThan(0);
 
+    // Verify the file was created and has content
+    expect(fs.existsSync(tempFilePath)).toBe(true);
+    const fileStats = fs.statSync(tempFilePath);
+    expect(fileStats.size).toBeGreaterThan(0);
+
+    // Read a few lines from the file to verify JSON format
+    const fileContent = fs.readFileSync(tempFilePath, "utf-8");
+    const lines = fileContent.split("\n").filter(line => line.trim());
+    expect(lines.length).toBeGreaterThan(0);
+
+    // Verify first line is valid JSON
+    if (lines.length > 0) {
+      const firstRow = JSON.parse(lines[0]);
+      expect(firstRow).toHaveProperty("id");
+      expect(firstRow).toHaveProperty("username");
+    }
+
+    // Clean up the temporary file
+    fs.unlinkSync(tempFilePath);
+
     // Memory usage should remain reasonable with proper streaming
     const memoryGrowth =
       (maxMemoryUsed - initialMemory.heapUsed) / (1024 * 1024);
@@ -431,7 +504,11 @@ describe("streams", () => {
       `Data types streaming test: processed ${rowCount} rows with various data types, ` +
         `memory growth: ${memoryGrowth.toFixed(
           2
-        )} MB, processed chunks: ${processedChunks}`
+        )} MB, processed chunks: ${processedChunks}, file size: ${(
+          fileStats.size /
+          1024 /
+          1024
+        ).toFixed(2)} MB`
     );
   });
 });