From fca018dd874e09a3cb89ea7dfe4355787c3c15cf Mon Sep 17 00:00:00 2001
From: Ardis Lu <ardisglu@gmail.com>
Date: Sun, 19 Oct 2025 21:43:40 -0700
Subject: [PATCH] Fix bug where batch is cleared before transform function can
 use it

---
 src/core.js       |  6 ++++--
 test/core.test.js | 13 +++++++++++++
 2 files changed, 17 insertions(+), 2 deletions(-)

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([
-- 
2.47.3