From d2eae41dd6e46d2cb2d430d9df661c4ea8afd4da Mon Sep 17 00:00:00 2001 From: Arvin A <51036481+DeveloperTheExplorer@users.noreply.github.com> Date: Tue, 29 Sep 2026 09:25:02 +0000 Subject: [PATCH] fix(DataTable Node): Continue paginated reads when row counts change (#39764) --- .../nodes/DataTable/common/selectMany.ts | 16 +- .../DataTable/test/common/selectMany.test.ts | 249 +++++++++++++++++- .../DataTable/test/row/get.operation.test.ts | 1 + 3 files changed, 243 insertions(+), 23 deletions(-) diff --git a/packages/nodes-base/nodes/DataTable/common/selectMany.ts b/packages/nodes-base/nodes/DataTable/common/selectMany.ts index f88e89ec27d..3877aa13080 100644 --- a/packages/nodes-base/nodes/DataTable/common/selectMany.ts +++ b/packages/nodes-base/nodes/DataTable/common/selectMany.ts @@ -197,11 +197,11 @@ export async function executeSelectMany( const nodeVersion = ctx.getNode().typeVersion; const shouldConvertDates = nodeVersion >= 1.1; - let expectedTotal: number | undefined; let skip = 0; let take = PAGE_SIZE; + const abortSignal = ctx.getExecutionCancelSignal(); - while (true) { + while (!abortSignal?.aborted) { const { data, count } = await dataTableProxy.getManyRowsAndCount({ skip, take: limit ? Math.min(take, limit - result.length) : take, @@ -217,21 +217,17 @@ export async function executeSelectMany( return wrapped; } - // Ensure the total doesn't change mid-pagination - if (expectedTotal !== undefined && count !== expectedTotal) { - throw new NodeOperationError( - ctx.getNode(), - 'synchronization error: result count changed during pagination', - ); + // Concurrent changes can leave an empty page even when the count suggests more rows. + if (data.length === 0) { + break; } - expectedTotal = count; result.push.apply(result, wrapped); // Stop if we've hit the limit if (limit && result.length >= limit) break; - // Stop if we've collected everything + // Stop at the latest count if (result.length >= count) break; skip = result.length; diff --git a/packages/nodes-base/nodes/DataTable/test/common/selectMany.test.ts b/packages/nodes-base/nodes/DataTable/test/common/selectMany.test.ts index 1e6800bb834..df4e5979452 100644 --- a/packages/nodes-base/nodes/DataTable/test/common/selectMany.test.ts +++ b/packages/nodes-base/nodes/DataTable/test/common/selectMany.test.ts @@ -18,9 +18,14 @@ describe('selectMany utils', () => { } as unknown as IDataTableProjectService); const dataTableId = 2345; let filters: FieldEntry[]; + let returnAll: boolean; + let rowLimit: number; const node = { id: 1 } as unknown as INode; beforeEach(() => { + getManyRowsAndCount.mockReset(); + returnAll = true; + rowLimit = 50; filters = [ { condition: 'eq', @@ -39,6 +44,7 @@ describe('selectMany utils', () => { mockExecuteFunctions = { getNode: vi.fn().mockReturnValue(node), + getExecutionCancelSignal: vi.fn().mockReturnValue(undefined), getNodeParameter: vi.fn().mockImplementation((field) => { switch (field) { case DATA_TABLE_ID_FIELD: @@ -47,6 +53,10 @@ describe('selectMany utils', () => { return filters; case 'matchType': return ANY_CONDITION; + case 'returnAll': + return returnAll; + case 'limit': + return rowLimit; } }), helpers: { @@ -107,25 +117,238 @@ describe('selectMany utils', () => { expect(result).toEqual([{ json: { id: 1, colA: null } }]); }); - it('should panic if pagination gets out of sync', async () => { - // ARRANGE + it.each([2344, 2346])( + 'should continue when the matching count changes from 2345 to %i', + async (count) => { + filters = [{ condition: 'eq', keyName: 'status', keyValue: 'active' }]; + const rows = Array.from({ length: count }, (_, id) => ({ id, status: 'active' })); + getManyRowsAndCount + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 2345 }) + .mockResolvedValueOnce({ data: rows.slice(1000, 2000), count }) + .mockResolvedValueOnce({ data: rows.slice(2000), count }); + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(3); + const filter = { + type: 'or', + filters: [{ columnName: 'status', condition: 'eq', value: 'active' }], + }; + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ skip: 0, take: 1000, filter }), + ); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ skip: 1000, take: 1000, filter }), + ); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 3, + expect.objectContaining({ skip: 2000, take: count - 2000, filter }), + ); + }, + ); + + it('should finish when repeated count increases stop', async () => { + const rows = Array.from({ length: 1003 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 1001 }) + .mockResolvedValueOnce({ data: rows.slice(1000, 1001), count: 1002 }) + .mockResolvedValueOnce({ data: rows.slice(1001, 1002), count: 1003 }) + .mockResolvedValueOnce({ data: rows.slice(1002), count: 1003 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(4); + for (const [index, skip] of [0, 1000, 1001, 1002].entries()) { + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + index + 1, + expect.objectContaining({ skip, take: index === 0 ? 1000 : 1 }), + ); + } + }); + + it('should stop requesting pages after cancellation while the count keeps increasing', async () => { + const controller = new AbortController(); + vi.mocked(mockExecuteFunctions.getExecutionCancelSignal).mockReturnValue(controller.signal); + const rows = Array.from({ length: 1002 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 1001 }) + .mockResolvedValueOnce({ data: rows.slice(1000, 1001), count: 1002 }) + .mockImplementationOnce(async () => { + controller.abort(); + return { data: rows.slice(1001), count: 1003 }; + }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(3); + }); + + it('should not request rows when execution is already cancelled', async () => { + const controller = new AbortController(); + controller.abort(); + vi.mocked(mockExecuteFunctions.getExecutionCancelSignal).mockReturnValue(controller.signal); + getManyRowsAndCount.mockRejectedValue(new Error('Unexpected page request')); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual([]); + expect(getManyRowsAndCount).not.toHaveBeenCalled(); + }); + + it('should stop when the count falls below the number of collected rows', async () => { + const rows = Array.from({ length: 2000 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 2345 }) + .mockResolvedValueOnce({ data: rows.slice(1000), count: 1500 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); + }); + + it('should continue when a short first page reports more matching rows', async () => { + const rows = [{ id: 1 }, { id: 2 }, { id: 3 }]; + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows.slice(0, 2), count: 3 }) + .mockResolvedValueOnce({ data: rows.slice(2), count: 3 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ skip: 2, take: 1 }), + ); + }); + + it('should return a nonempty final page when the count decreases', async () => { + const rows = Array.from({ length: 1499 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 1500 }) + .mockResolvedValueOnce({ data: rows.slice(1000), count: 1499 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ skip: 1000, take: 500 }), + ); + }); + + it.each([1000, 1001])( + 'should return the collected rows when an empty page reports a count of %i', + async (count) => { + const rows = Array.from({ length: 1000 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows, count: 1001 }) + .mockResolvedValueOnce({ data: [], count }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ skip: 1000, take: 1 }), + ); + }, + ); + + it('should stop when the first page is empty but its count is positive', async () => { + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: [], count: 1 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual([]); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(1); + }); + + it('should respect the configured limit when the count changes', async () => { + returnAll = false; + rowLimit = 1500; + const rows = Array.from({ length: 1500 }, (_, id) => ({ id })); + getManyRowsAndCount + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count: 2345 }) + .mockResolvedValueOnce({ data: rows.slice(1000), count: 2346 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); + expect(getManyRowsAndCount).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ skip: 1000, take: 500 }), + ); + }); + + it('should respect an explicit limit of one when returnAll is enabled', async () => { + getManyRowsAndCount.mockResolvedValueOnce({ data: [{ id: 1 }], count: 2345 }); + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy, false, 1); + + expect(result).toEqual([{ json: { id: 1 } }]); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(1); + expect(getManyRowsAndCount).toHaveBeenCalledWith( + expect.objectContaining({ skip: 0, take: 1 }), + ); + }); + + it.each([1000, 2000])( + 'should finish a stable read of %i rows without an extra request', + async (count) => { + const rows = Array.from({ length: count }, (_, id) => ({ id })); + getManyRowsAndCount + .mockRejectedValue(new Error('Unexpected additional page')) + .mockResolvedValueOnce({ data: rows.slice(0, 1000), count }); + if (count > 1000) { + getManyRowsAndCount.mockResolvedValueOnce({ data: rows.slice(1000), count }); + } + filters = []; + + const result = await executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); + + expect(result).toEqual(rows.map((json) => ({ json }))); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(count / 1000); + }, + ); + + it('should propagate a database error from a later page', async () => { + const error = new Error('Database connection failed'); getManyRowsAndCount.mockReturnValueOnce({ data: Array.from({ length: 1000 }, (_, k) => ({ id: k })), count: 2345, }); - getManyRowsAndCount.mockReturnValueOnce({ - data: Array.from({ length: 1000 }, (_, k) => ({ id: k + 1000 })), - count: 2344, - }); - + getManyRowsAndCount.mockRejectedValueOnce(error); filters = []; - // ACT ASSERT - const execution = executeSelectMany(mockExecuteFunctions, 0, dataTableProxy); - await expect(execution).rejects.toThrow(NodeOperationError); - await expect(execution).rejects.toThrow( - 'synchronization error: result count changed during pagination', - ); + await expect(executeSelectMany(mockExecuteFunctions, 0, dataTableProxy)).rejects.toBe(error); + expect(getManyRowsAndCount).toHaveBeenCalledTimes(2); }); describe('filter conditions', () => { diff --git a/packages/nodes-base/nodes/DataTable/test/row/get.operation.test.ts b/packages/nodes-base/nodes/DataTable/test/row/get.operation.test.ts index b3a2ebae82d..7341ac55583 100644 --- a/packages/nodes-base/nodes/DataTable/test/row/get.operation.test.ts +++ b/packages/nodes-base/nodes/DataTable/test/row/get.operation.test.ts @@ -29,6 +29,7 @@ describe('DataTable Get Operation - Sort Feature', () => { mockExecuteFunctions = { getNode: vi.fn().mockReturnValue(node), + getExecutionCancelSignal: vi.fn().mockReturnValue(undefined), getNodeParameter: vi.fn(), helpers: { getDataTableProxy: vi.fn().mockResolvedValue(mockDataTableProxy),