additional debug info

Author: ptiurin

test/integration/v2/stream.test.ts

@@ -1,4 +1,3 @@
-import exp from "node:constants";
 import { Firebolt } from "../../../src";
 import stream, { TransformCallback } from "node:stream";
 import BigNumber from "bignumber.js";
@@ -259,10 +258,20 @@ describe("streams", () => {
     console.log(`Pipeline test: processed ${processedCount} rows successfully`);
   });
   it("stream with different data types and memory management", async () => {
+    console.log("[DEBUG] Starting data types streaming test");
+    
+    // Add timeout tracking to identify where test hangs
+    const timeoutId = setTimeout(() => {
+      console.error(
+        "[DEBUG] Test timeout warning - test has been running for 5 minutes"
+      );
+    }, 300000); // 5 minutes
+    
     const firebolt = Firebolt({
       apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
     });
     const connection = await firebolt.connect(connectionParams);
+    console.log("[DEBUG] Connected to Firebolt");
 
     // Generate a query with various data types
     const seriesNum = 100000;
@@ -279,6 +288,7 @@ describe("streams", () => {
       FROM generate_series(1, ${rows}) as i
     `;
 
+    console.log(`[DEBUG] Executing stream query for ${seriesNum} rows`);
     const statement = await connection.executeStream(
       generateLargeResultQuery(seriesNum),
       {
@@ -288,11 +298,14 @@ describe("streams", () => {
         }
       }
     );
+    console.log("[DEBUG] Query execution started, getting stream result");
 
     const { data } = await statement.streamResult();
+    console.log("[DEBUG] Stream result obtained, setting up event handlers");
 
     // Add meta event handler to verify column metadata
     data.on("meta", m => {
+      console.log("[DEBUG] Received meta event with", m.length, "columns");
       try {
         expect(m).toEqual([
           { name: "id", type: "int" },
@@ -304,14 +317,17 @@ describe("streams", () => {
           { name: "score", type: "double" },
           { name: "description", type: "text" }
         ]);
+        console.log("[DEBUG] Meta validation passed");
       } catch (err) {
+        console.error("[DEBUG] Meta validation failed:", err);
         // Re-emit error so test fails properly
         data.destroy(err as Error);
       }
     });
 
     // Add error handler to ensure any stream errors are caught
     data.on("error", err => {
+      console.error("[DEBUG] Stream error occurred:", err.message);
       throw err;
     });
 
@@ -324,8 +340,14 @@ describe("streams", () => {
     const initialMemory = process.memoryUsage();
     let maxMemoryUsed = initialMemory.heapUsed;
     let rowCount = 0;
+    console.log(
+      "[DEBUG] Initial memory:",
+      (initialMemory.heapUsed / (1024 * 1024)).toFixed(2),
+      "MB"
+    );
 
     // Create a JSON transform stream with minimal allocation
+    console.log("[DEBUG] Creating JSON transform stream");
     const jsonTransform = new stream.Transform({
       objectMode: true,
       highWaterMark: 1, // Limit buffering - critical for memory
@@ -337,13 +359,26 @@ describe("streams", () => {
         try {
           rowCount++;
 
+          // More frequent progress logging for debugging
+          if (rowCount % 1000 === 0) {
+            const currentMemory = process.memoryUsage();
+            maxMemoryUsed = Math.max(maxMemoryUsed, currentMemory.heapUsed);
+            console.log(
+              `[DEBUG] Processed ${rowCount} rows, memory: ${(
+                currentMemory.heapUsed /
+                (1024 * 1024)
+              ).toFixed(2)} MB`
+            );
+          }
+
           if (rowCount % 5000 === 0) {
             const currentMemory = process.memoryUsage();
             maxMemoryUsed = Math.max(maxMemoryUsed, currentMemory.heapUsed);
           }
 
           // Verify data types are correct for normalized data on first row
           if (rowCount === 1) {
+            console.log("[DEBUG] Validating first row data types");
             const typedRow = row as Record<string, unknown>;
 
             // Verify actual values for first row
@@ -358,10 +393,12 @@ describe("streams", () => {
             expect(typedRow.description).toBe(
               "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua."
             );
+            console.log("[DEBUG] First row validation passed");
           }
 
           // Verify data types are correct for normalized data on last row
           if (rowCount === seriesNum) {
+            console.log("[DEBUG] Validating last row data types");
             const typedRow = row as Record<string, unknown>;
 
             // Verify actual values for last row
@@ -376,6 +413,7 @@ describe("streams", () => {
             expect(typedRow.description).toBe(
               "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua."
             );
+            console.log("[DEBUG] Last row validation passed");
           }
 
           const json = JSON.stringify(row);
@@ -405,12 +443,20 @@ describe("streams", () => {
 
     // Create a moderate backpressure stream
     let processedChunks = 0;
+    console.log("[DEBUG] Creating output transform stream");
     const outputStream = new stream.Transform({
       highWaterMark: 1,
       transform(chunk, encoding, callback) {
         try {
           processedChunks++;
 
+          // Log progress for debugging
+          if (processedChunks % 10000 === 0) {
+            console.log(
+              `[DEBUG] Output stream processed ${processedChunks} chunks`
+            );
+          }
+
           // Simulate occasional slow processing with minimal delays
           if (processedChunks % 1000 === 0) {
             // Use setImmediate instead of setTimeout to avoid potential timing issues
@@ -421,21 +467,27 @@ describe("streams", () => {
             callback();
           }
         } catch (err) {
+          console.error("[DEBUG] Error in output stream:", err);
           callback(err as Error);
         }
       }
     });
 
+    console.log("[DEBUG] Starting pipeline");
     // Use pipeline for proper backpressure handling
     await stream.promises.pipeline(data, jsonTransform, outputStream);
+    console.log("[DEBUG] Pipeline completed successfully");
 
     // Verify everything worked correctly
+    console.log(`[DEBUG] Final row count: ${rowCount}, expected: ${seriesNum}`);
     expect(rowCount).toBe(seriesNum);
+    console.log(`[DEBUG] Final processed chunks: ${processedChunks}`);
     expect(processedChunks).toBeGreaterThan(0);
 
     // Memory usage should remain reasonable with proper streaming
     const memoryGrowth =
       (maxMemoryUsed - initialMemory.heapUsed) / (1024 * 1024);
+    console.log(`[DEBUG] Memory growth: ${memoryGrowth.toFixed(2)} MB`);
     expect(memoryGrowth).toBeLessThan(120); // Allow reasonable memory for complex data types with various field types
 
     console.log(
@@ -444,5 +496,7 @@ describe("streams", () => {
           2
         )} MB, processed chunks: ${processedChunks}`
     );
+    console.log("[DEBUG] Test completed successfully");
+    clearTimeout(timeoutId);
   });
 });