feat(Extract from File Node): Add Skip Records With Errors option (#21347)

Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
Co-authored-by: Michael Kret <michael.k@radency.com>

packages/nodes-base/nodes/Files/ExtractFromFile/ExtractFromFile.node.ts

@@ -17,7 +17,7 @@ export class ExtractFromFile implements INodeType {
 		name: 'extractFromFile',
 		icon: { light: 'file:extractFromFile.svg', dark: 'file:extractFromFile.dark.svg' },
 		group: ['input'],
-		version: 1,
+		version: [1, 1.1],
 		description: 'Convert binary data to JSON',
 		defaults: {
 			name: 'Extract from File',
@@ -114,12 +114,15 @@ export class ExtractFromFile implements INodeType {
 	};
 
 	async execute(this: IExecuteFunctions) {
+		const version = this.getNode().typeVersion;
 		const items = this.getInputData();
 		const operation = this.getNodeParameter('operation', 0);
 		let returnData: INodeExecutionData[] = [];
 
 		if (spreadsheet.operations.includes(operation)) {
-			returnData = await spreadsheet.execute.call(this, items, 'operation');
+			returnData = await spreadsheet.execute.call(this, items, 'operation', {
+				failOnCsvBufferError: version > 1,
+			});
 		}
 
 		if (['binaryToPropery', 'fromJson', 'text', 'fromIcs', 'xml'].includes(operation)) {

packages/nodes-base/nodes/Files/ExtractFromFile/actions/spreadsheet.operation.ts

@@ -18,9 +18,15 @@ export const description: INodeProperties[] = fromFile.description
 			newProperty.options = (newProperty.options as INodeProperties[]).map((option) => {
 				let newOption = option;
 				if (
-					['delimiter', 'encoding', 'fromLine', 'maxRowCount', 'enableBOM', 'relaxQuotes'].includes(
-						option.name,
-					)
+					[
+						'delimiter',
+						'encoding',
+						'fromLine',
+						'maxRowCount',
+						'enableBOM',
+						'relaxQuotes',
+						'skipRecordsWithErrors',
+					].includes(option.name)
 				) {
 					newOption = { ...option, displayOptions: { show: { '/operation': ['csv'] } } };
 				}
@@ -53,11 +59,13 @@ export async function execute(
 	this: IExecuteFunctions,
 	items: INodeExecutionData[],
 	fileFormatProperty: string,
+	options?: fromFile.FromFileOptions,
 ) {
 	const returnData: INodeExecutionData[] = await fromFile.execute.call(
 		this,
 		items,
 		fileFormatProperty,
+		options,
 	);
 	return returnData;
 }

packages/nodes-base/nodes/SpreadsheetFile/description.ts

@@ -298,5 +298,39 @@ export const fromFileOptions: INodeProperties = {
 			placeholder: 'e.g. 0',
 			description: 'Start handling records from the requested line number. Starts at 0.',
 		},
+		{
+			displayName: 'Skip Records With Errors',
+			name: 'skipRecordsWithErrors',
+			type: 'fixedCollection',
+			default: { value: { enabled: true, maxSkippedRecords: -1 } },
+			options: [
+				{
+					displayName: 'Value',
+					name: 'value',
+					values: [
+						{
+							displayName: 'Enabled',
+							name: 'enabled',
+							type: 'boolean',
+							default: false,
+							description: 'Whether to skip records with errors when reading from file',
+						},
+						{
+							displayName: 'Max Skipped Records',
+							name: 'maxSkippedRecords',
+							type: 'number',
+							default: -1,
+							description:
+								'The maximum number of records that can be skipped, will throw an error if exceeded. Set to -1 to remove limit.',
+						},
+					],
+				},
+			],
+			displayOptions: {
+				show: {
+					'/fileFormat': ['csv'],
+				},
+			},
+		},
 	],
 };

packages/nodes-base/nodes/SpreadsheetFile/test/fromFile.operation.test.ts

@@ -1,5 +1,5 @@
 import { mockDeep } from 'jest-mock-extended';
-import type { IExecuteFunctions, INodeExecutionData, IBinaryData, INode } from 'n8n-workflow';
+import type { IBinaryData, IExecuteFunctions, INode, INodeExecutionData } from 'n8n-workflow';
 import { BINARY_ENCODING, NodeOperationError } from 'n8n-workflow';
 import { Readable } from 'stream';
 
@@ -677,4 +677,124 @@ describe('fromFile.operation - xlsx parsing logic', () => {
 			});
 		});
 	});
+
+	describe('CSV parsing with skipRecordsWithErrors', () => {
+		const invalidCsvData = 'id,name\n3,"John"\n1,"Alice\n2,"Bob"';
+		const mockBinaryDataCSV: IBinaryData = {
+			data: Buffer.from(invalidCsvData, 'utf8').toString(BINARY_ENCODING),
+			mimeType: 'text/csv',
+			fileExtension: 'csv',
+			fileName: 'test.csv',
+		};
+
+		beforeEach(() => {
+			jest.clearAllMocks();
+			mockExecuteFunctions.getNodeParameter.mockImplementation(
+				(paramName: string, _itemIndex: number, defaultValue?: any) => {
+					switch (paramName) {
+						case 'fileFormat':
+							return 'csv';
+						case 'binaryPropertyName':
+							return 'data';
+						case 'options':
+							return {};
+						default:
+							return defaultValue;
+					}
+				},
+			);
+			mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataCSV);
+			mockExecuteFunctions.getNode.mockReturnValue({
+				name: 'SpreadsheetFile',
+				type: 'n8n-nodes-base.spreadsheetFile',
+				id: 'test-node-id',
+			} as INode);
+			mockExecuteFunctions.continueOnFail.mockReturnValue(false);
+		});
+
+		it('should skip records with errors when skipRecordsWithErrors is enabled with limit -1', async () => {
+			mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => {
+				if (paramName === 'fileFormat') return 'csv';
+				if (paramName === 'binaryPropertyName') return 'data';
+				if (paramName === 'options')
+					return {
+						skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: -1 } },
+						columns: true,
+					};
+				return undefined;
+			});
+
+			const items: INodeExecutionData[] = [{ json: {} }];
+
+			const result = await execute.call(mockExecuteFunctions, items);
+
+			// Should have 1 valid record (John), Bob and Alice is considered a single record with error
+			expect(result).toHaveLength(1);
+			expect(result[0].json).toEqual({ id: '3', name: 'John' });
+		});
+
+		it('should skip records with errors when skipRecordsWithErrors is enabled with limit 1', async () => {
+			mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => {
+				if (paramName === 'fileFormat') return 'csv';
+				if (paramName === 'binaryPropertyName') return 'data';
+				if (paramName === 'options')
+					return {
+						skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } },
+						columns: true,
+					};
+				return undefined;
+			});
+
+			const items: INodeExecutionData[] = [{ json: {} }];
+
+			const result = await execute.call(mockExecuteFunctions, items);
+
+			expect(result).toHaveLength(1);
+			expect(result[0].json).toEqual({ id: '3', name: 'John' });
+			expect(result[0].pairedItem).toEqual({ item: 0 });
+		});
+
+		it('should throw error when skipped records exceed maxSkippedRecords limit', async () => {
+			mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => {
+				if (paramName === 'fileFormat') return 'csv';
+				if (paramName === 'binaryPropertyName') return 'data';
+				if (paramName === 'options')
+					return {
+						skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } },
+						columns: true,
+					};
+				return undefined;
+			});
+
+			const csvWithThreeErrors =
+				'id,name\n3,"John"\n1,"Alice\n2,"Bob"\n4,"Charlie\n5,"Eve\n6,"David';
+			const mockBinaryDataThreeErrors: IBinaryData = {
+				data: Buffer.from(csvWithThreeErrors, 'utf8').toString(BINARY_ENCODING),
+				mimeType: 'text/csv',
+				fileExtension: 'csv',
+				fileName: 'test-three-errors.csv',
+			};
+
+			mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataThreeErrors);
+
+			const items: INodeExecutionData[] = [{ json: {} }];
+
+			mockExecuteFunctions.getNodeParameter.mockImplementation((paramName: string) => {
+				if (paramName === 'fileFormat') return 'csv';
+				if (paramName === 'binaryPropertyName') return 'data';
+				if (paramName === 'options')
+					return {
+						skipRecordsWithErrors: { value: { enabled: true, maxSkippedRecords: 1 } },
+						columns: true,
+					};
+				return undefined;
+			});
+
+			mockExecuteFunctions.helpers.assertBinaryData.mockReturnValue(mockBinaryDataThreeErrors);
+
+			await expect(execute.call(mockExecuteFunctions, items)).rejects.toThrow(
+				'Max number of skipped records exceeded',
+			);
+		});
+	});
 });

packages/nodes-base/nodes/SpreadsheetFile/v2/fromFile.operation.ts

@@ -1,4 +1,4 @@
-import { parse as createCSVParser } from 'csv-parse';
+import { parse as createCSVParser, type Options as CSVOptions } from 'csv-parse';
 import type { IExecuteFunctions, INodeExecutionData, INodeProperties } from 'n8n-workflow';
 import { BINARY_ENCODING, NodeOperationError } from 'n8n-workflow';
 import type { Sheet2JSONOpts, ParsingOptions } from 'xlsx';
@@ -6,6 +6,25 @@ import { read as xlsxRead, utils as xlsxUtils } from 'xlsx';
 
 import { binaryProperty, fromFileOptions } from '../description';
 
+interface Options {
+	maxRowCount?: number;
+	delimiter?: string;
+	fromLine?: number;
+	encoding?: BufferEncoding;
+	enableBOM?: boolean;
+	skipRecordsWithErrors?: {
+		value?: { enabled?: boolean; maxSkippedRecords?: number };
+	};
+	to?: number;
+	relaxQuotes?: boolean;
+	includeEmptyCells?: boolean;
+	rawData?: boolean;
+	readAsString?: boolean;
+	sheetName?: string;
+	range?: number | string;
+	headerRow?: boolean;
+}
+
 export const description: INodeProperties[] = [
 	binaryProperty,
 	{
@@ -59,10 +78,15 @@ export const description: INodeProperties[] = [
 	fromFileOptions,
 ];
 
+export interface FromFileOptions {
+	failOnCsvBufferError?: boolean;
+}
+
 export async function execute(
 	this: IExecuteFunctions,
 	items: INodeExecutionData[],
 	fileFormatProperty = 'fileFormat',
+	{ failOnCsvBufferError = false }: FromFileOptions = {},
 ) {
 	const returnData: INodeExecutionData[] = [];
 	let fileExtension;
@@ -70,7 +94,7 @@ export async function execute(
 
 	for (let i = 0; i < items.length; i++) {
 		try {
-			const options = this.getNodeParameter('options', i, {});
+			const options = this.getNodeParameter('options', i, {}) as Options;
 			fileFormat = this.getNodeParameter(fileFormatProperty, i, '');
 			const binaryPropertyName = this.getNodeParameter('binaryPropertyName', i);
 			const binaryData = this.helpers.assertBinaryData(i, binaryPropertyName);
@@ -88,14 +112,16 @@ export async function execute(
 
 			if (fileFormat === 'csv') {
 				const maxRowCount = options.maxRowCount as number;
-				const parser = createCSVParser({
-					delimiter: options.delimiter as string,
-					fromLine: options.fromLine as number,
-					encoding: options.encoding as BufferEncoding,
-					bom: options.enableBOM as boolean,
+				const skipRecordsWithErrors = options.skipRecordsWithErrors?.value?.enabled;
+				const csvOptions: CSVOptions = {
+					delimiter: options.delimiter,
+					fromLine: options.fromLine,
+					encoding: options.encoding,
+					bom: options.enableBOM,
 					to: maxRowCount > -1 ? maxRowCount : undefined,
+					skip_records_with_error: skipRecordsWithErrors,
 					columns: options.headerRow !== false,
-					relax_quotes: options.relaxQuotes as boolean,
+					relax_quotes: options.relaxQuotes,
 					onRecord: (record) => {
 						if (!options.includeEmptyCells) {
 							record = Object.fromEntries(
@@ -104,10 +130,17 @@ export async function execute(
 						}
 						rows.push(record);
 					},
+				};
+				const parser = createCSVParser(csvOptions);
+
+				let skippedRecords = 0;
+				parser.on('skip', (_err) => {
+					skippedRecords += 1;
 				});
+
 				if (binaryData.id) {
 					const stream = await this.helpers.getBinaryStream(binaryData.id);
-					await new Promise<void>(async (resolve, reject) => {
+					await new Promise<void>((resolve, reject) => {
 						parser.on('error', reject);
 						parser.on('readable', () => {
 							stream.unpipe(parser);
@@ -118,7 +151,26 @@ export async function execute(
 					});
 				} else {
 					parser.write(binaryData.data, BINARY_ENCODING);
-					parser.end();
+
+					if (failOnCsvBufferError) {
+						await new Promise<void>((resolve, reject) => {
+							parser.on('error', reject);
+							parser.on('readable', () => {
+								resolve();
+							});
+							parser.end();
+						});
+					} else {
+						// this ignores errors, but we keep it for backwards compatibility
+						parser.end();
+					}
+				}
+
+				const maxSkippedRecords = options.skipRecordsWithErrors?.value?.maxSkippedRecords ?? -1;
+				if (skipRecordsWithErrors && maxSkippedRecords > 0 && skippedRecords > maxSkippedRecords) {
+					throw new NodeOperationError(this.getNode(), 'Max number of skipped records exceeded', {
+						itemIndex: i,
+					});
 				}
 			} else {
 				const xlsxOptions: ParsingOptions = { raw: options.rawData as boolean };