fix: include column context in bulk load validation errors

src/bulk-load-payload.ts

@@ -117,6 +117,7 @@ export class BulkLoadPayload implements AsyncIterable<Buffer> {
 
     let done = false;
     let closed = false;
+    let rowIndex = 0;
     try {
       buffer.writeBuffer(bulkLoad.getColMetaData());
 
@@ -139,32 +140,38 @@ export class BulkLoadPayload implements AsyncIterable<Buffer> {
           row = await row;
         }
 
+        if (row === null || typeof row !== 'object') {
+          throw new InputError(`Bulk load row at index ${rowIndex} is not an array or object`);
+        }
+
         buffer.writeBuffer(rowTokenBuffer);
 
         const isArray = Array.isArray(row);
         for (let i = 0; i < columns.length; i++) {
           const c = columns[i];
-          const value = c.type.validate(isArray ? (row as unknown[])[i] : (row as { [colName: string]: unknown })[c.objName], c.collation);
-
-          const parameter = {
-            length: c.length,
-            scale: c.scale,
-            precision: c.precision,
-            value: value
-          };
-
-          if (isTextType[i] && value == null) {
-            buffer.writeBuffer(textPointerNullBuffer);
-          } else {
-            if (isTextType[i]) {
-              buffer.writeBuffer(textPointerAndTimestampBuffer);
-            }
-
-            try {
+          const input = isArray ? (row as unknown[])[i] : (row as { [colName: string]: unknown })[c.objName];
+          let step = 'validated';
+          try {
+            const value = c.type.validate(input, c.collation);
+            const parameter = {
+              length: c.length,
+              scale: c.scale,
+              precision: c.precision,
+              value: value
+            };
+
+            if (isTextType[i] && value == null) {
+              buffer.writeBuffer(textPointerNullBuffer);
+            } else {
+              if (isTextType[i]) {
+                buffer.writeBuffer(textPointerAndTimestampBuffer);
+              }
+
+              step = 'serialized';
               c.type.writeValue(buffer, parameter, options);
-            } catch (error) {
-              throw new InputError(`Column '${c.name}' could not be serialized`, { cause: error });
             }
+          } catch (error) {
+            throw new InputError(`Column '${c.name}' could not be ${step} at row index ${rowIndex}`, { cause: error });
           }
 
           // Checked after every cell, not only after every row, so that a
@@ -176,6 +183,8 @@ export class BulkLoadPayload implements AsyncIterable<Buffer> {
             buffer.consume(buffer.length);
           }
         }
+
+        rowIndex++;
       }
     } catch (err) {
       // A failed row closes the source as a `for await` would. A close

test/integration/bulk-load-test.ts

@@ -1648,8 +1648,10 @@ describe('BulkLoad', function() {
       connection.execBulkLoad(bulkLoad, rowSource);
 
       function completeBulkLoad(err: Error | undefined | null, rowCount: undefined | number) {
-        assert.instanceOf(err, TypeError);
-        assert.strictEqual(err.message, 'Invalid date.');
+        assert.instanceOf(err, InputError);
+        assert.strictEqual(err.message, "Column 'value' could not be validated at row index 0");
+        assert.instanceOf(err.cause, TypeError);
+        assert.strictEqual((err.cause as TypeError).message, 'Invalid date.');
 
         done();
       }
@@ -1664,8 +1666,10 @@ describe('BulkLoad', function() {
       connection.execBulkLoad(bulkLoad, rowSource);
 
       function completeBulkLoad(err: Error | undefined | null, rowCount: undefined | number) {
-        assert.instanceOf(err, TypeError);
-        assert.strictEqual(err.message, 'Invalid date.');
+        assert.instanceOf(err, InputError);
+        assert.strictEqual(err.message, "Column 'value' could not be validated at row index 0");
+        assert.instanceOf(err.cause, TypeError);
+        assert.strictEqual((err.cause as TypeError).message, 'Invalid date.');
 
         assert.strictEqual(rowCount, 0);
 

test/unit/bulk-load-test.ts

@@ -1,7 +1,7 @@
 import { assert } from 'chai';
 import { Readable } from 'stream';
 import BulkLoad from '../../src/bulk-load';
-import { BulkLoadPayload } from '../../src/bulk-load-payload';
+import { BulkLoadPayload, type Row } from '../../src/bulk-load-payload';
 import { InputError, RequestError } from '../../src/errors';
 import Connection, { type InternalConnectionOptions } from '../../src/connection';
 import { type Request } from '../../src/tedious';
@@ -39,14 +39,18 @@ describe('BulkLoad', function() {
       };
     }
 
-    async function writeRow(type: DataType, rows: Iterable<unknown[]> = [[null]]): Promise<Error> {
+    async function writeRow(
+      type: DataType,
+      rows: Iterable<Row> = [[null]],
+      name = 'foo',
+      options: Parameters<BulkLoad['addColumn']>[2] = { nullable: true }
+    ): Promise<Error> {
       const request = new BulkLoad('tablename', undefined, connectionOptions, { }, () => {});
-      request.addColumn('foo', type, { nullable: true });
+      request.addColumn(name, type, options);
 
-      const chunks: Buffer[] = [];
       try {
         for await (const chunk of new BulkLoadPayload(request, rows)) {
-          chunks.push(chunk);
+          assert.instanceOf(chunk, Buffer);
         }
       } catch (err) {
         return err as Error;
@@ -76,10 +80,49 @@ describe('BulkLoad', function() {
       }));
 
       assert.instanceOf(error, InputError);
-      assert.match(error.message, /Column 'foo' could not be serialized/);
+      assert.strictEqual(error.message, "Column 'foo' could not be serialized at row index 0");
       assert.strictEqual(error.cause, cause);
     });
 
+    for (const { shape, rows } of [
+      { shape: 'array', rows: [['valid value'], ['sensitive value']] },
+      { shape: 'object', rows: [{ inputName: 'valid value' }, { inputName: 'sensitive value' }] }
+    ]) {
+      it(`reports the failing column for ${shape} rows without exposing the value`, async function() {
+        const cause = new TypeError('Invalid value.');
+        const type = buildType({
+          validate(value) {
+            if (value === 'sensitive value') {
+              throw cause;
+            }
+            return value;
+          }
+        });
+        const error = await writeRow(
+          type,
+          rows,
+          'databaseColumn',
+          { objName: 'inputName', nullable: false }
+        );
+
+        assert.instanceOf(error, InputError);
+        assert.strictEqual(error.message, "Column 'databaseColumn' could not be validated at row index 1");
+        assert.notInclude((error as InputError).stack, 'sensitive value');
+        // The rejected value is not retained on the error.
+        assert.deepEqual(Object.keys(error as object), []);
+        assert.strictEqual((error as InputError).cause, cause);
+      });
+    }
+
+    for (const row of [null, undefined]) {
+      it(`reports the row index when a row is ${row}`, async function() {
+        const rows = [[null], row] as unknown as Iterable<Row>;
+        const error = await writeRow(buildType({}), rows);
+
+        assert.instanceOf(error, InputError);
+        assert.strictEqual(error.message, 'Bulk load row at index 1 is not an array or object');
+      });
+    }
 
     it('closes the row source when the column metadata cannot be written', async function() {
       let closed = false;
@@ -254,7 +297,9 @@ describe('BulkLoad', function() {
         error = err;
       }
 
-      assert.instanceOf(error, TypeError);
+      assert.instanceOf(error, InputError);
+      assert.strictEqual((error as InputError).message, "Column 'id' could not be validated at row index 1");
+      assert.instanceOf((error as InputError).cause, TypeError);
       assert.lengthOf(chunks, 0);
       assert.isTrue(closed);
     });