Ignore `handleHeaders` if the row has been normalized

This behavior is desirable when `CSVTransformer` interacts with `CSVNormalizer` or any other transformation which converts the row from `Array<string>` into `Array<Object>`, because it avoids the need to pass `handleHeaders: true` to `CSVTransformer` (which is counterintuitive). It is more accurate to say the header row has been deleted by the transformation and must by re-inserted into the write stream later.

The loosening of the type checking is not a breaking change. Since the row is passed directly to `JSON.parse()`, the fields could have always been type `Object` instead of `string`.

diff --git a/src/core.js b/src/core.js
index 5b6f0248e714a1129e621533b2e66f930e15eadb..95948997eb17eddcfccc024303cad197af79fabc 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -202,7 +202,7 @@ export class CSVReader extends ReadableStream {
 /**
  * A function to process a row of CSV data from `CSVReader`.
  * @callback TransformationFunction
- * @param {Array<string>} row A row of input CSV data before transformation.
+ * @param {Array<any>} row A row of input CSV data before transformation.
  * @returns {TransformationOutput|Promise<TransformationOutput>}
  */
 
@@ -216,14 +216,14 @@ export class CSVReader extends ReadableStream {
 /**
  * A function to process multiple rows of CSV data from `CSVReader`.
  * @callback TransformationFunctionBatch
- * @param {Array<Array<string>>} rows Multiple rows of input CSV data.
+ * @param {Array<Array<any>>} rows Multiple rows of input CSV data.
  * @returns {TransformationOutput|Promise<TransformationOutput>}
  */
 
 /**
  * A function to handle errors thrown by a transformation function.
  * @callback TransformationErrorFunction
- * @param {Array<string>} row The row passed to the transformation function which threw the error.
+ * @param {Array<any>} row The row passed to the transformation function which threw the error.
  * @param {Error} error The error thrown by the transformation function.
  * @param {TransformationFunction} fn The transformation function itself. This argument can be used to retry a transformation.
  * @returns {TransformationOutput|Promise<TransformationOutput>}
@@ -241,7 +241,7 @@ export class CSVReader extends ReadableStream {
 /**
  * A function to handle errors thrown by a transformation function.
  * @callback TransformationErrorFunctionBatch
- * @param {Array<Array<string>>} rows The rows of input CSV data passed to the transformation function which threw the error.
+ * @param {Array<Array<any>>} rows The rows of input CSV data passed to the transformation function which threw the error.
  * @param {Error} error The error thrown by the transformation function.
  * @param {TransformationFunctionBatch} fn The transformation function itself. This argument can be used to retry a transformation.
  * @returns {TransformationOutput|Promise<TransformationOutput>}
@@ -433,6 +433,13 @@ export class CSVTransformer extends TransformStream {
   async #transform(chunk, controller) {
     const row = this.#rawInput ? chunk : JSON.parse(chunk);
 
+    // If row is an array of objects, then the transformation has "normalized" the data and there is no header row.
+    // Instead, the header data has been incorporated into the object keys. The transformation is responsible
+    // for "denormalizing" the object back into an (extra) header row in the output. Ignore `handleHeaders`.
+    if (typeof row[0] === 'object') {
+      this.#firstChunk = false;
+    }
+
     if (this.#firstChunk) {
       this.#firstChunk = false;
       const out = await this.#wrappedHeaderFn(row);

diff --git a/test/core.test.js b/test/core.test.js
index 3e0180ee7ffad1b14938a261d2b8e5e06c02a1fc..e5100d009ee30d64fa5cbfc83faad7cc4c23e1f5 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -400,6 +400,17 @@ suite('CSVTransformer', { concurrency: true }, () => {
         ['1', '2']
       ]));
   });
+  test('ignores handleHeaders when row is an array of objects (normalized)', { concurrency: true, only: true }, async () => {
+    await createCSVMockStream([
+      [{ value: 'a1' }, { value: 'b1' }],
+      [{ value: 'a2' }, { value: 'b2' }],
+    ])
+      .pipeThrough(new CSVTransformer(r => r.map(f => f.value), { handleHeaders: ['updatedA', 'updatedB'] }))
+      .pipeTo(csvStreamEqualWritable([
+        ['a1', 'b1'],
+        ['a2', 'b2']
+      ]));
+  });
   test('enqueues multiple rows when Array<Array<any>> is returned', { concurrency: true }, async () => {
     await createCSVMockStream([
       ['columnA', 'columnB'],

diff --git a/test/normalization.test.js b/test/normalization.test.js
index c12577c28c0c81551f15a53a92f5c6615570e75c..3ee5ecf5e55a14ac96caa4e5d1d920d73f01007e 100644 (file)
--- a/test/normalization.test.js
+++ b/test/normalization.test.js
@@ -2,7 +2,7 @@ import { suite, test } from 'node:test';
 import { ok, deepStrictEqual, throws } from 'node:assert/strict';
 
 import { csvStreamEqualWritable, createCSVMockStream, assertConsole } from './utils.js';
-import { CSVReader } from '../src/core.js';
+import { CSVReader, CSVTransformer } from '../src/core.js';
 import { getDecimalSeparator, toCamelCase, expandScientificNotation, CSVNormalizer, CSVDenormalizer } from '../src/normalization.js';
 
 suite('getDecimalSeparator', { concurrency: true }, () => {
@@ -397,4 +397,23 @@ suite('CSVNormalizer and CSVDenormalizer end-to-end', { concurrency: true }, ()
         ['N/A', 123100, '1000000000000000000', '2024-12-31T08:00:00.000Z']
       ]));
   });
+  test('can normalize with transformation', { concurrency: true }, async () => {
+    const headers = [
+      /** @type {const} */({ name: 'columnA' }),
+      /** @type {const} */({ name: 'columnB' })
+    ]
+    await createCSVMockStream([
+      ['columnA', '', 'columnB', '', ''],
+      ['a', '', 'b', '', ''],
+      ['', '', '', '', ''],
+      ['', '', '', '', '']
+    ])
+      .pipeThrough(new CSVNormalizer(headers))
+      .pipeThrough(new CSVTransformer(r => r.map(f => ({ name: `${f.name} updated`, value: `${f.value} updated` }))))
+      .pipeThrough(new CSVDenormalizer())
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA updated', 'columnB updated'],
+        ['a updated', 'b updated']
+      ]));
+  });
 });
\ No newline at end of file