From: Ardis Lu <ardisglu@gmail.com>
Date: Mon, 20 Oct 2025 04:43:40 +0000 (-0700)
Subject: Fix bug where batch is cleared before transform function can use it
X-Git-Tag: v0.1.8~4
X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=fca018dd874e09a3cb89ea7dfe4355787c3c15cf;p=csvio

Fix bug where batch is cleared before transform function can use it
---

diff --git a/src/core.js b/src/core.js
index ef97463..4e49125 100644
--- a/src/core.js
+++ b/src/core.js
@@ -377,7 +377,6 @@ export class CSVTransformer extends TransformStream {
 
   async #enqueueConcurrent(controller) {
     const results = await Promise.allSettled(this.#concurrent);
-    this.#concurrent.length = 0;
     for (const r of results) {
       if (r.status === 'rejected') {
         if (r.reason instanceof Error) {
@@ -391,6 +390,10 @@ export class CSVTransformer extends TransformStream {
         this.#enqueueRow(r.value, controller);
       }
     }
+    // Only clear these arrays AFTER all transform processing is complete. Otherwise downstream functions
+    // may not be able to access the row data.
+    this.#concurrent.length = 0;
+    this.#batch.length = 0;
   }
 
   /** @type {TransformerTransformCallback<Array<string>,Array<string>>} */
@@ -413,7 +416,6 @@ export class CSVTransformer extends TransformStream {
       this.#batch.push(chunk);
       if (this.#batch.length === this.#maxBatchSize) {
         this.#concurrent.push(this.#wrappedFn(this.#batch));
-        this.#batch.length = 0;
       }
     }
     else {
diff --git a/test/core.test.js b/test/core.test.js
index fb7a4af..cead01f 100644
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -482,6 +482,19 @@ suite('CSVTransformer', { concurrency: true }, () => {
       { name: 'Error', message: 'Not an error type' }
     );
   });
+  test('can batch row data', { concurrency: true }, async (t) => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2']
+    ])
+      .pipeThrough(new CSVTransformer(r => r, { maxBatchSize: 2 }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1'],
+        ['2']
+      ]));
+  });
   test('creates batch of 1 when maxBatchSize is equal to 1', { concurrency: true }, async (t) => {
     const fn = t.mock.fn(b => b.map(r => r.map(f => Number(f) * 2)));
     await createCSVMockStream([