try this approach to testing

Author: ptiurin

test/integration/v2/stream.test.ts

@@ -259,10 +259,14 @@ describe("streams", () => {
     console.log(`Pipeline test: processed ${processedCount} rows successfully`);
   });
   it("stream with different data types and memory management", async () => {
+    console.log("Starting data types streaming test...");
+    const startTime = Date.now();
+
     const firebolt = Firebolt({
       apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
     });
     const connection = await firebolt.connect(connectionParams);
+    console.log("Connected to Firebolt");
 
     // Generate a query with various data types
     const seriesNum = 100000;
@@ -279,6 +283,7 @@ describe("streams", () => {
       FROM generate_series(1, ${rows}) as i
     `;
 
+    console.log(`Executing query for ${seriesNum} rows...`);
     const statement = await connection.executeStream(
       generateLargeResultQuery(seriesNum),
       {
@@ -289,10 +294,13 @@ describe("streams", () => {
       }
     );
 
+    console.log("Getting stream result...");
     const { data } = await statement.streamResult();
+    console.log("Stream result obtained, setting up listeners...");
 
     // Add meta event handler to verify column metadata
     data.on("meta", m => {
+      console.log("Meta received:", m.length, "columns");
       expect(m).toEqual([
         { name: "id", type: "int" },
         { name: "username", type: "text" },
@@ -305,15 +313,17 @@ describe("streams", () => {
       ]);
     });
 
-    // Buffer pool configuration
-    const poolSize = 8192; // 8KB
-    const poolBuffer = Buffer.allocUnsafe(poolSize);
-    const newlineCode = 0x0a; // '\n' character code
+    // Add error handler to the data stream
+    data.on("error", error => {
+      console.error("Data stream error:", error);
+      throw error;
+    });
 
     // Track memory usage
     const initialMemory = process.memoryUsage();
     let maxMemoryUsed = initialMemory.heapUsed;
     let rowCount = 0;
+    let transformStart = 0;
 
     // Create a JSON transform stream with minimal allocation
     const jsonTransform = new stream.Transform({
@@ -327,9 +337,20 @@ describe("streams", () => {
         try {
           rowCount++;
 
-          if (rowCount % 5000 === 0) {
+          if (rowCount === 1) {
+            transformStart = Date.now();
+            console.log("Transform started, first row received");
+          }
+
+          if (rowCount % 10000 === 0) {
             const currentMemory = process.memoryUsage();
             maxMemoryUsed = Math.max(maxMemoryUsed, currentMemory.heapUsed);
+            const elapsed = Date.now() - transformStart;
+            const memoryMB = (currentMemory.heapUsed / 1024 / 1024).toFixed(2);
+            console.log(
+              `Transform progress: ${rowCount} rows processed in ${elapsed}ms, ` +
+                `memory: ${memoryMB}MB`
+            );
           }
 
           // Verify data types are correct for normalized data on first row
@@ -369,65 +390,120 @@ describe("streams", () => {
           }
 
           const json = JSON.stringify(row);
-          const jsonLen = Buffer.byteLength(json);
-          const totalLen = jsonLen + 1;
-
-          let buffer: Buffer;
-          if (totalLen <= poolSize) {
-            // Use pool for small rows - no allocation
-            poolBuffer.write(json, 0, jsonLen);
-            poolBuffer[jsonLen] = newlineCode;
-            buffer = poolBuffer.subarray(0, totalLen);
-          } else {
-            // Allocate for large rows
-            buffer = Buffer.allocUnsafe(totalLen);
-            buffer.write(json, 0, jsonLen);
-            buffer[jsonLen] = newlineCode;
+          const jsonBuffer = Buffer.from(json + "\n", "utf8");
+
+          this.push(jsonBuffer);
+
+          if (rowCount === seriesNum) {
+            const transformEnd = Date.now();
+            const elapsed = transformEnd - transformStart;
+            console.log(
+              `Transform completed: ${seriesNum} rows processed in ${elapsed}ms`
+            );
           }
 
-          this.push(buffer);
           callback();
         } catch (err) {
+          console.error("Transform error:", err);
           callback(err as Error);
         }
       }
     });
 
+    // Add error handler to transform stream
+    jsonTransform.on("error", error => {
+      console.error("JSON transform stream error:", error);
+    });
+
     // Create a moderate backpressure stream
     let processedChunks = 0;
+    let outputStart = 0;
     const outputStream = new stream.Writable({
       highWaterMark: 1,
       write(chunk, encoding, callback) {
+        if (processedChunks === 0) {
+          outputStart = Date.now();
+          console.log("Output stream started processing chunks");
+        }
+
         processedChunks++;
 
-        // Simulate occasional slow processing with minimal delays
-        if (processedChunks % 1000 === 0) {
-          setTimeout(() => {
-            callback();
-          }, 1); // 1ms delay
-        } else {
+        // Log progress periodically
+        if (processedChunks % 10000 === 0) {
+          const elapsed = Date.now() - outputStart;
+          console.log(
+            `Output progress: ${processedChunks} chunks processed in ${elapsed}ms`
+          );
+        }
+
+        // Use synchronous processing instead of setTimeout to avoid timing issues in CI
+        try {
+          // Simulate some minimal processing without async delay
+          if (processedChunks % 1000 === 0) {
+            // Just add a tiny synchronous delay instead of async setTimeout
+            const start = Date.now();
+            while (Date.now() - start < 0.1) {
+              // Busy wait for 0.1ms
+            }
+          }
           callback();
+        } catch (err) {
+          callback(err);
         }
       }
     });
 
+    // Add error handler to output stream
+    outputStream.on("error", error => {
+      console.error("Output stream error:", error);
+    });
+
+    outputStream.on("finish", () => {
+      const elapsed = Date.now() - outputStart;
+      console.log(
+        `Output stream finished: ${processedChunks} total chunks in ${elapsed}ms`
+      );
+    });
+
     // Use pipeline for proper backpressure handling
-    await stream.promises.pipeline(data, jsonTransform, outputStream);
+    console.log("Starting pipeline...");
+    const pipelineStart = Date.now();
+
+    // Create the main pipeline promise
+    const pipelinePromise = stream.promises.pipeline(
+      data,
+      jsonTransform,
+      outputStream
+    );
+
+    try {
+      // Execute the pipeline
+      await pipelinePromise;
+      const pipelineEnd = Date.now();
+      console.log(`Pipeline completed in ${pipelineEnd - pipelineStart}ms`);
+    } catch (error) {
+      console.error("Pipeline error:", error);
+      throw error;
+    }
 
     // Verify everything worked correctly
+    console.log(
+      `Final verification: rowCount=${rowCount}, processedChunks=${processedChunks}`
+    );
     expect(rowCount).toBe(seriesNum);
     expect(processedChunks).toBeGreaterThan(0);
 
     // Memory usage should remain reasonable with proper streaming
     const memoryGrowth =
       (maxMemoryUsed - initialMemory.heapUsed) / (1024 * 1024);
-    expect(memoryGrowth).toBeLessThan(120); // Allow reasonable memory for complex data types with various field types
+    expect(memoryGrowth).toBeLessThan(150); // Allow reasonable memory for complex data types with various field types
 
+    const totalElapsed = Date.now() - startTime;
     console.log(
-      `Data types streaming test: processed ${rowCount} rows with various data types, ` +
+      `Data types streaming test completed: processed ${rowCount} rows with various data types, ` +
         `memory growth: ${memoryGrowth.toFixed(
           2
-        )} MB, processed chunks: ${processedChunks}`
+        )} MB, processed chunks: ${processedChunks}, total time: ${totalElapsed}ms`
     );
   });
 });