From 71b1de4bf1f1840ba0fbe7f6b252e3cfcf9d86f1 Mon Sep 17 00:00:00 2001 From: Artem Niehrieiev Date: Wed, 14 Oct 2026 11:48:10 +0000 Subject: [PATCH] feat: implement bulk deletion for tables with composite primary keys - Added end-to-end tests for bulk deletion of rows in PostgreSQL, handling cases where the number of keys exceeds statement limits and ensuring proper logging of deletions. - Introduced unit tests for primary key normalization and filtering, ensuring correct handling of composite keys and null values. - Refactored data access objects for ClickHouse, MSSQL, MySQL, Oracle, and PostgreSQL to utilize new bulk primary key handling logic, improving efficiency and correctness in bulk delete operations. - Created helper functions for normalizing primary keys, chunking them according to database limits, and applying filters in queries. - Updated error messages for clarity regarding primary key requirements. --- .../entities/table-logs/table-logs.service.ts | 21 +- .../pure-delete-row-from-table.use.case.ts | 2 +- .../pure-read-row-from-table.use.case.ts | 2 +- .../pure-update-row-in-table.use.case.ts | 2 +- .../src/entities/table/table.controller.ts | 1 + .../delete-row-from-table.use.case.ts | 2 +- .../delete-rows-from-table.use.case.ts | 24 +- .../get-row-by-primary-key.use.case.ts | 2 +- .../use-cases/update-row-in-table.use.case.ts | 2 +- .../add-display-names-for-tables.util.ts | 2 +- .../build-table-settings-for-response.util.ts | 8 +- .../utils/find-objects-with-properties.ts | 6 - ...n-saas-table-mysql-bulk-delete-e2e.test.ts | 222 ++++++++++++++++++ ...aas-table-postgres-bulk-delete-e2e.test.ts | 220 +++++++++++++++++ .../bulk-primary-keys-filter.test.ts | 144 ++++++++++++ .../data-access-object-clickhouse.ts | 35 ++- .../data-access-object-mssql.ts | 58 ++--- .../data-access-object-mysql.ts | 51 ++-- .../data-access-object-oracle.ts | 59 ++--- .../data-access-object-postgres.ts | 59 ++--- shared-code/src/helpers/bulk-primary-keys.ts | 120 ++++++++++ .../src/helpers/errors/error-messages.ts | 2 + 22 files changed, 906 insertions(+), 138 deletions(-) delete mode 100644 backend/src/entities/table/utils/find-objects-with-properties.ts create mode 100644 backend/test/ava-tests/non-saas-tests/non-saas-table-mysql-bulk-delete-e2e.test.ts create mode 100644 backend/test/ava-tests/non-saas-tests/non-saas-table-postgres-bulk-delete-e2e.test.ts create mode 100644 backend/test/ava-tests/unit-tests/bulk-primary-keys-filter.test.ts create mode 100644 shared-code/src/helpers/bulk-primary-keys.ts diff --git a/backend/src/entities/table-logs/table-logs.service.ts b/backend/src/entities/table-logs/table-logs.service.ts index 108bf4e7c..f26a970de 100644 --- a/backend/src/entities/table-logs/table-logs.service.ts +++ b/backend/src/entities/table-logs/table-logs.service.ts @@ -1,7 +1,6 @@ import { Injectable } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { findSensitiveValues, scrub } from '@zapier/secret-scrubber'; -import PQueue from 'p-queue'; import { Repository } from 'typeorm'; import { LogOperationTypeEnum } from '../../enums/log-operation-type.enum.js'; import { OperationResultStatusEnum } from '../../enums/operation-result-status.enum.js'; @@ -17,6 +16,9 @@ import { TableLogsEntity } from './table-logs.entity.js'; import { buildCreatedLogRecord } from './utils/build-created-log-record.js'; import { buildTableLogsEntity } from './utils/build-table-logs-entity.js'; +/** Rows per INSERT when a bulk operation writes its audit log; keeps one statement well under Postgres' bind limit. */ +const TABLE_LOGS_INSERT_CHUNK_SIZE = 500; + @Injectable() export class TableLogsService { constructor( @@ -220,16 +222,13 @@ export class TableLogsService { const newLogRecord = buildTableLogsEntity(loggedData, email); tableLogsEntities.push(newLogRecord); } - const queue = new PQueue({ concurrency: 2 }); - const createdLogs: Array = await Promise.all( - tableLogsEntities.map(async (newLogRecord) => { - return await queue.add(async () => { - const savedLogRecord = await this.tableLogsRepository.save(newLogRecord); - return buildCreatedLogRecord(savedLogRecord); - }); - }), - ); - return createdLogs.filter((log) => log) as Array; + if (tableLogsEntities.length === 0) { + return []; + } + const savedLogRecords = await this.tableLogsRepository.save(tableLogsEntities, { + chunk: TABLE_LOGS_INSERT_CHUNK_SIZE, + }); + return savedLogRecords.map((savedLogRecord) => buildCreatedLogRecord(savedLogRecord)); } private async isTableAuditEnabledInConnectionProperties(connectionId: string): Promise { diff --git a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-delete-row-from-table.use.case.ts b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-delete-row-from-table.use.case.ts index 695039fac..b7974abd2 100644 --- a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-delete-row-from-table.use.case.ts +++ b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-delete-row-from-table.use.case.ts @@ -64,7 +64,7 @@ export class PureDeleteRowFromTableUseCase const availablePrimaryColumns = primaryColumns.map((column) => column.column_name); for (const key in primaryKey) { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') delete primaryKey[key]; + if (primaryKey[key] === undefined) delete primaryKey[key]; } const receivedPrimaryColumns = Object.keys(primaryKey); if (!compareArrayElements(availablePrimaryColumns, receivedPrimaryColumns)) { diff --git a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-read-row-from-table.use.case.ts b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-read-row-from-table.use.case.ts index 9f53ce25d..9b875b76c 100644 --- a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-read-row-from-table.use.case.ts +++ b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-read-row-from-table.use.case.ts @@ -64,7 +64,7 @@ export class PureReadRowFromTableUseCase const availablePrimaryColumns = primaryColumns.map((column) => column.column_name); for (const key in primaryKey) { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') delete primaryKey[key]; + if (primaryKey[key] === undefined) delete primaryKey[key]; } const receivedPrimaryColumns = Object.keys(primaryKey); if (!compareArrayElements(availablePrimaryColumns, receivedPrimaryColumns)) { diff --git a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-update-row-in-table.use.case.ts b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-update-row-in-table.use.case.ts index 00a0ec500..8eb50a311 100644 --- a/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-update-row-in-table.use.case.ts +++ b/backend/src/entities/table/table-pure-crud-operations/use-cases/pure-update-row-in-table.use.case.ts @@ -67,7 +67,7 @@ export class PureUpdateRowInTableUseCase const availablePrimaryColumns = tablePrimaryKeys.map((key) => key.column_name); for (const key in primaryKey) { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') delete primaryKey[key]; + if (primaryKey[key] === undefined) delete primaryKey[key]; } const receivedPrimaryColumns = Object.keys(primaryKey); if (!compareArrayElements(availablePrimaryColumns, receivedPrimaryColumns)) { diff --git a/backend/src/entities/table/table.controller.ts b/backend/src/entities/table/table.controller.ts index 0b9c4ab33..68162588d 100644 --- a/backend/src/entities/table/table.controller.ts +++ b/backend/src/entities/table/table.controller.ts @@ -565,6 +565,7 @@ export class TableController { }) @ApiQuery({ name: 'tableName', required: true }) @UseGuards(TableDeleteGuard) + @Timeout(TimeoutDefaults.EXTENDED) @Put('/table/rows/delete/:connectionId') async deleteRowsInTable( @MasterPassword() masterPwd: string, diff --git a/backend/src/entities/table/use-cases/delete-row-from-table.use.case.ts b/backend/src/entities/table/use-cases/delete-row-from-table.use.case.ts index b73aef65d..933499d34 100644 --- a/backend/src/entities/table/use-cases/delete-row-from-table.use.case.ts +++ b/backend/src/entities/table/use-cases/delete-row-from-table.use.case.ts @@ -80,7 +80,7 @@ export class DeleteRowFromTableUseCase Object.keys(primaryKey).forEach((key) => { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') { + if (primaryKey[key] === undefined) { // eslint-disable-next-line security/detect-object-injection delete primaryKey[key]; } diff --git a/backend/src/entities/table/use-cases/delete-rows-from-table.use.case.ts b/backend/src/entities/table/use-cases/delete-rows-from-table.use.case.ts index 1049677fc..bf91ea61c 100644 --- a/backend/src/entities/table/use-cases/delete-rows-from-table.use.case.ts +++ b/backend/src/entities/table/use-cases/delete-rows-from-table.use.case.ts @@ -17,7 +17,6 @@ import { TableLogsService } from '../../table-logs/table-logs.service.js'; import { DeleteRowsFromTableDs } from '../application/data-structures/delete-row-from-table.ds.js'; import { buildCommonTableSettingsInput } from '../utils/build-common-table-settings-input.util.js'; import { convertHexDataInPrimaryKeyUtil } from '../utils/convert-hex-data-in-primary-key.util.js'; -import { findObjectsWithProperties } from '../utils/find-objects-with-properties.js'; import { getUserEmailForAgent, validateConnection } from '../utils/validate-connection.util.js'; import { IDeleteRowsFromTable } from './table-use-cases.interface.js'; @@ -87,7 +86,7 @@ export class DeleteRowsFromTableUseCase primaryKeys.forEach((primaryKey) => { Object.keys(primaryKey).forEach((key) => { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') { + if (primaryKey[key] === undefined) { // eslint-disable-next-line security/detect-object-injection delete primaryKey[key]; } @@ -124,6 +123,7 @@ export class DeleteRowsFromTableUseCase ); } + const oldRowsByPrimaryKey = indexRowsByPrimaryKey(oldRowsData, availablePrimaryColumns); const deleteOperationsResults: Array = []; try { @@ -132,7 +132,7 @@ export class DeleteRowsFromTableUseCase deleteOperationsResults.push({ operationStatusResult: OperationResultStatusEnum.successfully, row: primaryKey, - old_data: findObjectsWithProperties(oldRowsData, primaryKey).at(0) ?? {}, + old_data: oldRowsByPrimaryKey.get(primaryKeyIndexKey(primaryKey, availablePrimaryColumns)) ?? {}, error: null, affected_primary_key: primaryKey as unknown as string, }); @@ -143,7 +143,7 @@ export class DeleteRowsFromTableUseCase deleteOperationsResults.push({ operationStatusResult: OperationResultStatusEnum.unsuccessfully, row: primaryKey, - old_data: findObjectsWithProperties(oldRowsData, primaryKey).at(0) ?? {}, + old_data: oldRowsByPrimaryKey.get(primaryKeyIndexKey(primaryKey, availablePrimaryColumns)) ?? {}, error: getErrorMessage(error), affected_primary_key: primaryKey as unknown as string, }); @@ -166,3 +166,19 @@ export class DeleteRowsFromTableUseCase } } } + +function primaryKeyIndexKey(row: Record, primaryColumns: Array): string { + // eslint-disable-next-line security/detect-object-injection + return primaryColumns.map((column) => String(row[column])).join('\u0000'); +} + +function indexRowsByPrimaryKey( + rows: Array>, + primaryColumns: Array, +): Map> { + const index = new Map>(); + for (const row of rows) { + index.set(primaryKeyIndexKey(row, primaryColumns), row); + } + return index; +} diff --git a/backend/src/entities/table/use-cases/get-row-by-primary-key.use.case.ts b/backend/src/entities/table/use-cases/get-row-by-primary-key.use.case.ts index c72be407f..cd4f35711 100644 --- a/backend/src/entities/table/use-cases/get-row-by-primary-key.use.case.ts +++ b/backend/src/entities/table/use-cases/get-row-by-primary-key.use.case.ts @@ -100,7 +100,7 @@ export class GetRowByPrimaryKeyUseCase const availablePrimaryColumns: Array = tablePrimaryKeys.map((column) => column.column_name); for (const key in primaryKey) { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') delete primaryKey[key]; + if (primaryKey[key] === undefined) delete primaryKey[key]; } const receivedPrimaryColumns = Object.keys(primaryKey); if (!compareArrayElements(availablePrimaryColumns, receivedPrimaryColumns)) { diff --git a/backend/src/entities/table/use-cases/update-row-in-table.use.case.ts b/backend/src/entities/table/use-cases/update-row-in-table.use.case.ts index 62b52fcb4..77dfaf65d 100644 --- a/backend/src/entities/table/use-cases/update-row-in-table.use.case.ts +++ b/backend/src/entities/table/use-cases/update-row-in-table.use.case.ts @@ -172,7 +172,7 @@ export class UpdateRowInTableUseCase }); for (const key in primaryKey) { // eslint-disable-next-line security/detect-object-injection - if (!primaryKey[key] && primaryKey[key] !== '') delete primaryKey[key]; + if (primaryKey[key] === undefined) delete primaryKey[key]; } const receivedPrimaryColumns = Object.keys(primaryKey); diff --git a/backend/src/entities/table/utils/add-display-names-for-tables.util.ts b/backend/src/entities/table/utils/add-display-names-for-tables.util.ts index 8a671064e..c710a763a 100644 --- a/backend/src/entities/table/utils/add-display-names-for-tables.util.ts +++ b/backend/src/entities/table/utils/add-display-names-for-tables.util.ts @@ -1,5 +1,5 @@ -import { TableSettingsEntity } from '../../table-settings/common-table-settings/table-settings.entity.js'; import { ITableAndViewPermissionData } from '../../permission/permission.interface.js'; +import { TableSettingsEntity } from '../../table-settings/common-table-settings/table-settings.entity.js'; import { FoundTableDs } from '../application/data-structures/found-table.ds.js'; export function addDisplayNamesForTables( diff --git a/backend/src/entities/table/utils/build-table-settings-for-response.util.ts b/backend/src/entities/table/utils/build-table-settings-for-response.util.ts index 3188b0b5e..d18b7b97f 100644 --- a/backend/src/entities/table/utils/build-table-settings-for-response.util.ts +++ b/backend/src/entities/table/utils/build-table-settings-for-response.util.ts @@ -3,7 +3,13 @@ import { TableSettingsInRowsDS } from '../application/data-structures/found-tabl export function buildTableSettingsForResponse( builtDAOsTableSettings: TableSettingsDS, - tableSettings?: { allow_csv_export?: boolean; allow_csv_import?: boolean; can_delete?: boolean; can_update?: boolean; can_add?: boolean } | null, + tableSettings?: { + allow_csv_export?: boolean; + allow_csv_import?: boolean; + can_delete?: boolean; + can_update?: boolean; + can_add?: boolean; + } | null, ): TableSettingsInRowsDS { return { sortable_by: builtDAOsTableSettings?.sortable_by?.length > 0 ? builtDAOsTableSettings.sortable_by : [], diff --git a/backend/src/entities/table/utils/find-objects-with-properties.ts b/backend/src/entities/table/utils/find-objects-with-properties.ts deleted file mode 100644 index 78c96cf6d..000000000 --- a/backend/src/entities/table/utils/find-objects-with-properties.ts +++ /dev/null @@ -1,6 +0,0 @@ -export function findObjectsWithProperties(objectsArray: T[], properties: Pick): T[] { - return objectsArray.filter((obj) => { - // eslint-disable-next-line security/detect-object-injection - return Object.entries(properties).every(([key, value]) => (obj as Record)[key] === value); - }); -} diff --git a/backend/test/ava-tests/non-saas-tests/non-saas-table-mysql-bulk-delete-e2e.test.ts b/backend/test/ava-tests/non-saas-tests/non-saas-table-mysql-bulk-delete-e2e.test.ts new file mode 100644 index 000000000..dbe7d8f4c --- /dev/null +++ b/backend/test/ava-tests/non-saas-tests/non-saas-table-mysql-bulk-delete-e2e.test.ts @@ -0,0 +1,222 @@ +import { INestApplication, ValidationPipe } from '@nestjs/common'; +import { Test } from '@nestjs/testing'; +import test from 'ava'; +import { ValidationError } from 'class-validator'; +import cookieParser from 'cookie-parser'; +import { nanoid } from 'nanoid'; +import request from 'supertest'; +import { ApplicationModule } from '../../../src/app.module.js'; +import { WinstonLogger } from '../../../src/entities/logging/winston-logger.js'; +import { LogOperationTypeEnum } from '../../../src/enums/log-operation-type.enum.js'; +import { AllExceptionsFilter } from '../../../src/exceptions/all-exceptions.filter.js'; +import { ValidationException } from '../../../src/exceptions/custom-exceptions/validation-exception.js'; +import { Cacher } from '../../../src/helpers/cache/cacher.js'; +import { DatabaseModule } from '../../../src/shared/database/database.module.js'; +import { DatabaseService } from '../../../src/shared/database/database.service.js'; +import { MockFactory } from '../../mock.factory.js'; +import { createTestTable } from '../../utils/create-test-table.js'; +import { getTestData } from '../../utils/get-test-data.js'; +import { getTestKnex } from '../../utils/get-test-knex.js'; +import { + createInitialTestUser, + registerUserAndReturnUserInfo, +} from '../../utils/register-user-and-return-user-info.js'; +import { setSaasEnvVariable } from '../../utils/set-saas-env-variable.js'; +import { TestUtils } from '../../utils/test.utils.js'; + +// Bulk row deletion against a real MySQL: more keys than fit one statement (the DAO chunks at +// 1000 keys and runs the chunks in one transaction), a composite primary key (row-value IN list), +// and the audit log, which is now written with one batched INSERT per chunk instead of one +// INSERT per deleted row. + +const mockFactory = new MockFactory(); +// The test knex speaks the mysql2 driver; the connection the backend stores keeps type "mysql". +const mysqlKnexParams = () => ({ ...getTestData(mockFactory).connectionToMySQL, type: 'mysql2' }); +let app: INestApplication; +let _testUtils: TestUtils; +let currentTest: string; + +const suffix = nanoid(6).replace(/[-_]/g, 'x').toLowerCase(); +const compositeTableName = `orders_${suffix}`; +const COMPOSITE_ROWS = 1200; +const COMPOSITE_KEYS_TO_DELETE = 1100; + +test.before(async () => { + setSaasEnvVariable(); + const moduleFixture = await Test.createTestingModule({ + imports: [ApplicationModule, DatabaseModule], + providers: [DatabaseService, TestUtils], + }).compile(); + app = moduleFixture.createNestApplication(); + _testUtils = moduleFixture.get(TestUtils); + + app.use(cookieParser()); + app.useGlobalFilters(new AllExceptionsFilter(app.get(WinstonLogger))); + app.useGlobalPipes( + new ValidationPipe({ + exceptionFactory(validationErrors: ValidationError[] = []) { + return new ValidationException(validationErrors); + }, + }), + ); + await app.init(); + await createInitialTestUser(app); + app.getHttpServer().listen(0); + + const knex = getTestKnex(mysqlKnexParams()); + await knex.raw( + `CREATE TABLE \`${compositeTableName}\` ( + region integer NOT NULL, + code varchar(16) NOT NULL, + note text, + PRIMARY KEY (region, code) + )`, + ); + await knex.batchInsert( + compositeTableName, + Array.from({ length: COMPOSITE_ROWS }, (_, i) => ({ region: i % 10, code: `c${i}`, note: `row ${i}` })), + 200, + ); +}); + +test.after(async () => { + try { + const knex = getTestKnex(mysqlKnexParams()); + await knex.raw(`DROP TABLE IF EXISTS \`${compositeTableName}\``); + await Cacher.clearAllCache(); + await app.close(); + } catch (e) { + console.error('After tests error ' + e); + } +}); + +async function createConnection(token: string): Promise { + const createConnectionResponse = await request(app.getHttpServer()) + .post('/connection') + .send(getTestData(mockFactory).connectionToMySQL) + .set('Cookie', token) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + if (createConnectionResponse.status !== 201) { + throw new Error(`connection not created: ${createConnectionResponse.text}`); + } + return JSON.parse(createConnectionResponse.text).id; +} + +currentTest = 'PUT /table/rows/delete/:slug'; + +test.serial(`${currentTest} should delete more rows than fit one statement and log every row`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToMySQL; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const seedsCount = 2500; + const { testTableName } = await createTestTable(connectionToTestDB, seedsCount); + const connectionId = await createConnection(firstUserToken); + + const keysToDelete = 1500; + const primaryKeysForDeletion = Array.from({ length: keysToDelete }, (_, i) => ({ id: i + 1 })); + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send(primaryKeysForDeletion) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + t.is(JSON.parse(deleteRowsInTableResponse.text), true); + + const knex = getTestKnex(mysqlKnexParams()); + const remaining = await knex(testTableName).select('id').orderBy('id', 'asc'); + t.is(remaining.length, seedsCount - keysToDelete); + t.is(remaining[0].id, keysToDelete + 1); + t.is(remaining.at(-1).id, seedsCount); + + const tableLogsResponse = await request(app.getHttpServer()) + .get(`/logs/${connectionId}?tableName=${testTableName}&page=1&perPage=10`) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(tableLogsResponse.status, 200); + const tableLogsRO = JSON.parse(tableLogsResponse.text); + t.is(tableLogsRO.pagination.total, keysToDelete); + t.is(tableLogsRO.logs.length, 10); + for (const log of tableLogsRO.logs) { + t.is(log.operationType, LogOperationTypeEnum.deleteRow); + t.is(typeof log.received_data.id, 'number'); + t.is(log.old_data.id, log.received_data.id); + } +}); + +test.serial(`${currentTest} should delete rows of a table with a composite primary key`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToMySQL; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const connectionId = await createConnection(firstUserToken); + + const primaryKeysForDeletion = Array.from({ length: COMPOSITE_KEYS_TO_DELETE }, (_, i) => ({ + code: `c${i}`, + region: i % 10, + })); + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${compositeTableName}`) + .send(primaryKeysForDeletion) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + + const knex = getTestKnex(mysqlKnexParams()); + const remaining = await knex(compositeTableName).select('region', 'code'); + t.is(remaining.length, COMPOSITE_ROWS - COMPOSITE_KEYS_TO_DELETE); + t.true(remaining.every((row) => Number(row.code.slice(1)) >= COMPOSITE_KEYS_TO_DELETE)); + + const tableLogsResponse = await request(app.getHttpServer()) + .get(`/logs/${connectionId}?tableName=${compositeTableName}&page=1&perPage=5`) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(tableLogsResponse.status, 200); + const tableLogsRO = JSON.parse(tableLogsResponse.text); + t.is(tableLogsRO.pagination.total, COMPOSITE_KEYS_TO_DELETE); + for (const log of tableLogsRO.logs) { + t.is(log.old_data.note, `row ${log.received_data.code.slice(1)}`); + } +}); + +test.serial(`${currentTest} should reject a key that is not the table primary key`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToMySQL; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const { testTableName } = await createTestTable(connectionToTestDB, 5); + const connectionId = await createConnection(firstUserToken); + + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send([{ id: 1 }, {}]) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 400); + + const knex = getTestKnex(mysqlKnexParams()); + const remaining = await knex(testTableName).count('* as count'); + t.is(Number(remaining[0].count), 5); +}); + +test.serial(`${currentTest} should accept a null key value and delete nothing for it`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToMySQL; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const { testTableName } = await createTestTable(connectionToTestDB, 5); + const connectionId = await createConnection(firstUserToken); + + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send([{ id: 2 }, { id: null }]) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + + const knex = getTestKnex(mysqlKnexParams()); + const remaining = await knex(testTableName).select('id').orderBy('id', 'asc'); + t.deepEqual( + remaining.map((row) => row.id), + [1, 3, 4, 5], + ); +}); diff --git a/backend/test/ava-tests/non-saas-tests/non-saas-table-postgres-bulk-delete-e2e.test.ts b/backend/test/ava-tests/non-saas-tests/non-saas-table-postgres-bulk-delete-e2e.test.ts new file mode 100644 index 000000000..cfba1fcf5 --- /dev/null +++ b/backend/test/ava-tests/non-saas-tests/non-saas-table-postgres-bulk-delete-e2e.test.ts @@ -0,0 +1,220 @@ +import { INestApplication, ValidationPipe } from '@nestjs/common'; +import { Test } from '@nestjs/testing'; +import test from 'ava'; +import { ValidationError } from 'class-validator'; +import cookieParser from 'cookie-parser'; +import { nanoid } from 'nanoid'; +import request from 'supertest'; +import { ApplicationModule } from '../../../src/app.module.js'; +import { WinstonLogger } from '../../../src/entities/logging/winston-logger.js'; +import { LogOperationTypeEnum } from '../../../src/enums/log-operation-type.enum.js'; +import { AllExceptionsFilter } from '../../../src/exceptions/all-exceptions.filter.js'; +import { ValidationException } from '../../../src/exceptions/custom-exceptions/validation-exception.js'; +import { Cacher } from '../../../src/helpers/cache/cacher.js'; +import { DatabaseModule } from '../../../src/shared/database/database.module.js'; +import { DatabaseService } from '../../../src/shared/database/database.service.js'; +import { MockFactory } from '../../mock.factory.js'; +import { createTestTable } from '../../utils/create-test-table.js'; +import { getTestData } from '../../utils/get-test-data.js'; +import { getTestKnex } from '../../utils/get-test-knex.js'; +import { + createInitialTestUser, + registerUserAndReturnUserInfo, +} from '../../utils/register-user-and-return-user-info.js'; +import { setSaasEnvVariable } from '../../utils/set-saas-env-variable.js'; +import { TestUtils } from '../../utils/test.utils.js'; + +// Bulk row deletion against a real Postgres: more keys than fit one statement (the DAO chunks at +// 1000 keys and runs the chunks in one transaction), a composite primary key (row-value IN list), +// and the audit log, which is now written with one batched INSERT per chunk instead of one +// INSERT per deleted row. + +const mockFactory = new MockFactory(); +let app: INestApplication; +let _testUtils: TestUtils; +let currentTest: string; + +const suffix = nanoid(6).replace(/[-_]/g, 'x').toLowerCase(); +const compositeTableName = `orders_${suffix}`; +const COMPOSITE_ROWS = 1200; +const COMPOSITE_KEYS_TO_DELETE = 1100; + +test.before(async () => { + setSaasEnvVariable(); + const moduleFixture = await Test.createTestingModule({ + imports: [ApplicationModule, DatabaseModule], + providers: [DatabaseService, TestUtils], + }).compile(); + app = moduleFixture.createNestApplication(); + _testUtils = moduleFixture.get(TestUtils); + + app.use(cookieParser()); + app.useGlobalFilters(new AllExceptionsFilter(app.get(WinstonLogger))); + app.useGlobalPipes( + new ValidationPipe({ + exceptionFactory(validationErrors: ValidationError[] = []) { + return new ValidationException(validationErrors); + }, + }), + ); + await app.init(); + await createInitialTestUser(app); + app.getHttpServer().listen(0); + + const knex = getTestKnex(getTestData(mockFactory).connectionToPostgres); + await knex.raw( + `CREATE TABLE "${compositeTableName}" ( + region integer NOT NULL, + code varchar(16) NOT NULL, + note text, + PRIMARY KEY (region, code) + )`, + ); + await knex.batchInsert( + compositeTableName, + Array.from({ length: COMPOSITE_ROWS }, (_, i) => ({ region: i % 10, code: `c${i}`, note: `row ${i}` })), + 200, + ); +}); + +test.after(async () => { + try { + const knex = getTestKnex(getTestData(mockFactory).connectionToPostgres); + await knex.raw(`DROP TABLE IF EXISTS "${compositeTableName}"`); + await Cacher.clearAllCache(); + await app.close(); + } catch (e) { + console.error('After tests error ' + e); + } +}); + +async function createConnection(token: string): Promise { + const createConnectionResponse = await request(app.getHttpServer()) + .post('/connection') + .send(getTestData(mockFactory).connectionToPostgres) + .set('Cookie', token) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + if (createConnectionResponse.status !== 201) { + throw new Error(`connection not created: ${createConnectionResponse.text}`); + } + return JSON.parse(createConnectionResponse.text).id; +} + +currentTest = 'PUT /table/rows/delete/:slug'; + +test.serial(`${currentTest} should delete more rows than fit one statement and log every row`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToPostgres; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const seedsCount = 2500; + const { testTableName } = await createTestTable(connectionToTestDB, seedsCount); + const connectionId = await createConnection(firstUserToken); + + const keysToDelete = 1500; + const primaryKeysForDeletion = Array.from({ length: keysToDelete }, (_, i) => ({ id: i + 1 })); + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send(primaryKeysForDeletion) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + t.is(JSON.parse(deleteRowsInTableResponse.text), true); + + const knex = getTestKnex(connectionToTestDB); + const remaining = await knex(testTableName).select('id').orderBy('id', 'asc'); + t.is(remaining.length, seedsCount - keysToDelete); + t.is(remaining[0].id, keysToDelete + 1); + t.is(remaining.at(-1).id, seedsCount); + + const tableLogsResponse = await request(app.getHttpServer()) + .get(`/logs/${connectionId}?tableName=${testTableName}&page=1&perPage=10`) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(tableLogsResponse.status, 200); + const tableLogsRO = JSON.parse(tableLogsResponse.text); + t.is(tableLogsRO.pagination.total, keysToDelete); + t.is(tableLogsRO.logs.length, 10); + for (const log of tableLogsRO.logs) { + t.is(log.operationType, LogOperationTypeEnum.deleteRow); + t.is(typeof log.received_data.id, 'number'); + t.is(log.old_data.id, log.received_data.id); + } +}); + +test.serial(`${currentTest} should delete rows of a table with a composite primary key`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToPostgres; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const connectionId = await createConnection(firstUserToken); + + const primaryKeysForDeletion = Array.from({ length: COMPOSITE_KEYS_TO_DELETE }, (_, i) => ({ + code: `c${i}`, + region: i % 10, + })); + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${compositeTableName}`) + .send(primaryKeysForDeletion) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + + const knex = getTestKnex(connectionToTestDB); + const remaining = await knex(compositeTableName).select('region', 'code'); + t.is(remaining.length, COMPOSITE_ROWS - COMPOSITE_KEYS_TO_DELETE); + t.true(remaining.every((row) => Number(row.code.slice(1)) >= COMPOSITE_KEYS_TO_DELETE)); + + const tableLogsResponse = await request(app.getHttpServer()) + .get(`/logs/${connectionId}?tableName=${compositeTableName}&page=1&perPage=5`) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(tableLogsResponse.status, 200); + const tableLogsRO = JSON.parse(tableLogsResponse.text); + t.is(tableLogsRO.pagination.total, COMPOSITE_KEYS_TO_DELETE); + for (const log of tableLogsRO.logs) { + t.is(log.old_data.note, `row ${log.received_data.code.slice(1)}`); + } +}); + +test.serial(`${currentTest} should reject a key that is not the table primary key`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToPostgres; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const { testTableName } = await createTestTable(connectionToTestDB, 5); + const connectionId = await createConnection(firstUserToken); + + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send([{ id: 1 }, {}]) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 400); + + const knex = getTestKnex(connectionToTestDB); + const remaining = await knex(testTableName).count('* as count'); + t.is(Number(remaining[0].count), 5); +}); + +test.serial(`${currentTest} should accept a null key value and delete nothing for it`, async (t) => { + const connectionToTestDB = getTestData(mockFactory).connectionToPostgres; + const firstUserToken = (await registerUserAndReturnUserInfo(app)).token; + const { testTableName } = await createTestTable(connectionToTestDB, 5); + const connectionId = await createConnection(firstUserToken); + + const deleteRowsInTableResponse = await request(app.getHttpServer()) + .put(`/table/rows/delete/${connectionId}?tableName=${testTableName}`) + .send([{ id: 2 }, { id: null }]) + .set('Cookie', firstUserToken) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); + t.is(deleteRowsInTableResponse.status, 200, deleteRowsInTableResponse.text); + + const knex = getTestKnex(connectionToTestDB); + const remaining = await knex(testTableName).select('id').orderBy('id', 'asc'); + t.deepEqual( + remaining.map((row) => row.id), + [1, 3, 4, 5], + ); +}); diff --git a/backend/test/ava-tests/unit-tests/bulk-primary-keys-filter.test.ts b/backend/test/ava-tests/unit-tests/bulk-primary-keys-filter.test.ts new file mode 100644 index 000000000..316f2e88f --- /dev/null +++ b/backend/test/ava-tests/unit-tests/bulk-primary-keys-filter.test.ts @@ -0,0 +1,144 @@ +import { + applyPrimaryKeysFilter, + BULK_PRIMARY_KEYS_LIMITS, + chunkPrimaryKeyRows, + normalisePrimaryKeys, +} from '@rocketadmin/shared-code/dist/src/helpers/bulk-primary-keys.js'; +import { ERROR_MESSAGES } from '@rocketadmin/shared-code/dist/src/helpers/errors/error-messages.js'; +import test from 'ava'; +import knex from 'knex'; + +// Bulk row deletion (and the bulk "old rows" read that precedes it) used to filter with one +// `OR (col = ? AND …)` group per primary key, all in a single statement. The SQL DAOs now +// normalise the keys once, chunk them to each dialect's statement limits and filter with one +// IN list per chunk. These checks compile the filter with knex's own query compilers (no +// database) and pin the exact SQL each dialect gets. + +const compile = (client: string) => knex({ client }); + +test('normalisePrimaryKeys keeps the column order of the first key and reorders the rest', (t) => { + const { columns, rows } = normalisePrimaryKeys([ + { a: 1, b: 'x' }, + { b: 'y', a: 2 }, + ]); + t.deepEqual(columns, ['a', 'b']); + t.deepEqual(rows, [ + [1, 'x'], + [2, 'y'], + ]); +}); + +test('normalisePrimaryKeys returns nothing for no keys and refuses an empty key', (t) => { + t.deepEqual(normalisePrimaryKeys([]), { columns: [], rows: [] }); + t.throws(() => normalisePrimaryKeys([{}]), { message: ERROR_MESSAGES.PRIMARY_KEY_EMPTY }); +}); + +test('normalisePrimaryKeys refuses keys whose columns differ', (t) => { + t.throws(() => normalisePrimaryKeys([{ a: 1 }, { a: 2, b: 3 }]), { + message: ERROR_MESSAGES.PRIMARY_KEYS_COLUMNS_MISMATCH, + }); + t.throws(() => normalisePrimaryKeys([{ a: 1 }, { b: 2 }]), { + message: ERROR_MESSAGES.PRIMARY_KEYS_COLUMNS_MISMATCH, + }); +}); + +test('chunkPrimaryKeyRows splits by keys per statement and by bound values per statement', (t) => { + const singleColumn = normalisePrimaryKeys(Array.from({ length: 2500 }, (_, i) => ({ id: i }))); + t.deepEqual( + chunkPrimaryKeyRows(singleColumn, BULK_PRIMARY_KEYS_LIMITS.postgres).map((chunk) => chunk.length), + [1000, 1000, 500], + ); + + const threeColumns = normalisePrimaryKeys(Array.from({ length: 700 }, (_, i) => ({ a: i, b: i, c: i }))); + // MSSQL: 2000 bindings / 3 columns = 666 keys per statement. + t.deepEqual( + chunkPrimaryKeyRows(threeColumns, BULK_PRIMARY_KEYS_LIMITS.mssql).map((chunk) => chunk.length), + [666, 34], + ); + + const twoColumns = normalisePrimaryKeys(Array.from({ length: 1001 }, (_, i) => ({ a: i, b: i }))); + // Oracle: 1000 expressions per IN list / 2 columns = 500 keys per statement. + t.deepEqual( + chunkPrimaryKeyRows(twoColumns, BULK_PRIMARY_KEYS_LIMITS.oracle).map((chunk) => chunk.length), + [500, 500, 1], + ); + + t.deepEqual(chunkPrimaryKeyRows({ columns: [], rows: [] }, BULK_PRIMARY_KEYS_LIMITS.postgres), []); +}); + +test('applyPrimaryKeysFilter compiles a single-column key to one IN list', (t) => { + const { columns, rows } = normalisePrimaryKeys([{ id: 1 }, { id: 2 }, { id: 3 }]); + const pg = applyPrimaryKeysFilter(compile('pg')('t').delete(), columns, rows, true).toSQL().toNative(); + t.is(pg.sql, 'delete from "t" where "id" in ($1, $2, $3)'); + t.deepEqual(pg.bindings, [1, 2, 3]); + + const mysql = applyPrimaryKeysFilter(compile('mysql2')('t').select('*'), columns, rows, true).toSQL().toNative(); + t.is(mysql.sql, 'select * from `t` where `id` in (?, ?, ?)'); + + const mssql = applyPrimaryKeysFilter(compile('mssql')('t').select('*'), columns, rows, false).toSQL().toNative(); + t.is(mssql.sql, 'select * from [t] where [id] in (@p0, @p1, @p2)'); +}); + +test('applyPrimaryKeysFilter compiles a composite key to a row-value IN list where supported', (t) => { + const { columns, rows } = normalisePrimaryKeys([ + { a: 1, b: 'x' }, + { a: 2, b: 'y' }, + ]); + const pg = applyPrimaryKeysFilter(compile('pg')('t').delete(), columns, rows, true).toSQL().toNative(); + t.is(pg.sql, 'delete from "t" where ("a", "b") in (($1, $2), ($3, $4))'); + t.deepEqual(pg.bindings, [1, 'x', 2, 'y']); + + const oracle = applyPrimaryKeysFilter(compile('oracledb')('t').delete(), columns, rows, true).toSQL().toNative(); + t.is(oracle.sql, 'delete from "t" where ("a", "b") in ((:1, :2), (:3, :4))'); +}); + +test('applyPrimaryKeysFilter falls back to a grouped OR of ANDs for composite keys without row values', (t) => { + const { columns, rows } = normalisePrimaryKeys([ + { a: 1, b: 'x' }, + { a: 2, b: 'y' }, + ]); + const mssql = applyPrimaryKeysFilter(compile('mssql')('t').delete(), columns, rows, false).toSQL().toNative(); + t.is(mssql.sql, 'delete from [t] where (([a] = @p0 and [b] = @p1) or ([a] = @p2 and [b] = @p3));select @@rowcount'); + t.deepEqual(mssql.bindings, [1, 'x', 2, 'y']); +}); + +test('applyPrimaryKeysFilter keeps the key group isolated from other predicates', (t) => { + const { columns, rows } = normalisePrimaryKeys([ + { a: 1, b: 'x' }, + { a: 2, b: 'y' }, + ]); + const sql = applyPrimaryKeysFilter(compile('pg')('t').select('*').where('deleted', false), columns, rows, false) + .toSQL() + .toNative().sql; + t.is(sql, 'select * from "t" where "deleted" = $1 and (("a" = $2 and "b" = $3) or ("a" = $4 and "b" = $5))'); +}); + +test('applyPrimaryKeysFilter matches a null key component with IS NULL instead of IN', (t) => { + const single = normalisePrimaryKeys([{ id: 1 }, { id: null }]); + const pgSingle = applyPrimaryKeysFilter(compile('pg')('t').delete(), single.columns, single.rows, true) + .toSQL() + .toNative(); + t.is(pgSingle.sql, 'delete from "t" where ("id" in ($1) or ("id" is null))'); + t.deepEqual(pgSingle.bindings, [1]); + + const onlyNull = normalisePrimaryKeys([{ id: null }]); + t.is( + applyPrimaryKeysFilter(compile('pg')('t').delete(), onlyNull.columns, onlyNull.rows, true).toSQL().toNative().sql, + 'delete from "t" where (("id" is null))', + ); + + const composite = normalisePrimaryKeys([ + { a: 1, b: 'x' }, + { a: 2, b: null }, + ]); + const pgComposite = applyPrimaryKeysFilter(compile('pg')('t').delete(), composite.columns, composite.rows, true) + .toSQL() + .toNative(); + t.is(pgComposite.sql, 'delete from "t" where (("a", "b") in (($1, $2)) or ("a" = $3 and "b" is null))'); + t.deepEqual(pgComposite.bindings, [1, 'x', 2]); + + const mssql = applyPrimaryKeysFilter(compile('mssql')('t').delete(), composite.columns, composite.rows, false) + .toSQL() + .toNative(); + t.is(mssql.sql, 'delete from [t] where (([a] = @p0 and [b] = @p1) or ([a] = @p2 and [b] is null));select @@rowcount'); +}); diff --git a/shared-code/src/data-access-layer/data-access-objects/data-access-object-clickhouse.ts b/shared-code/src/data-access-layer/data-access-objects/data-access-object-clickhouse.ts index 3f5437a01..50f3afd16 100644 --- a/shared-code/src/data-access-layer/data-access-objects/data-access-object-clickhouse.ts +++ b/shared-code/src/data-access-layer/data-access-objects/data-access-object-clickhouse.ts @@ -5,6 +5,7 @@ import * as csv from 'csv'; import getPort from 'get-port'; import { Readable, Stream } from 'stream'; import { LRUStorage } from '../../caching/lru-storage.js'; +import { normalisePrimaryKeys } from '../../helpers/bulk-primary-keys.js'; import { DAO_CONSTANTS } from '../../helpers/data-access-objects-constants.js'; import { ERROR_MESSAGES } from '../../helpers/errors/error-messages.js'; import { getTunnel } from '../../helpers/get-ssh-tunnel.js'; @@ -136,7 +137,7 @@ export class DataAccessObjectClickHouse extends BasicDataAccessObject implements const selectFields = availableFields.length > 0 ? availableFields.map((f) => this.escapeIdentifier(f)).join(', ') : '*'; - const whereConditions = primaryKeys.map((pk) => `(${this.buildWhereClause(pk)})`).join(' OR '); + const whereConditions = this.buildPrimaryKeysInClause(primaryKeys); const query = `SELECT ${selectFields} FROM ${this.escapeIdentifier(tableName)} WHERE ${whereConditions}`; @@ -537,7 +538,7 @@ export class DataAccessObjectClickHouse extends BasicDataAccessObject implements const client = await this.getClickHouseClient(); try { - const whereConditions = primaryKeys.map((pk) => `(${this.buildWhereClause(pk)})`).join(' OR '); + const whereConditions = this.buildPrimaryKeysInClause(primaryKeys); const query = `ALTER TABLE ${this.escapeIdentifier(tableName)} DELETE WHERE ${whereConditions}`; await client.command({ query }); @@ -781,6 +782,36 @@ export class DataAccessObjectClickHouse extends BasicDataAccessObject implements throw new Error(`Unsupported value type: ${typeof value}`); } + /** + * `col IN (…)` for one key column, `(a, b) IN ((…), (…))` for composite keys; a key with a null + * component cannot match through IN and is OR-ed in as `(a = … AND b IS NULL)`. + */ + private buildPrimaryKeysInClause(primaryKeys: Array>): string { + const { columns, rows } = normalisePrimaryKeys(primaryKeys); + const escapedColumns = columns.map((column) => this.escapeIdentifier(column)); + const listRows = rows.filter((row) => !row.some((value) => value === null || value === undefined)); + const nullRows = rows.filter((row) => row.some((value) => value === null || value === undefined)); + const predicates: Array = []; + if (listRows.length > 0) { + predicates.push( + columns.length === 1 + ? `${escapedColumns[0]} IN (${listRows.map((row) => this.sanitizeValue(row[0])).join(', ')})` + : `(${escapedColumns.join(', ')}) IN (${listRows + .map((row) => `(${row.map((value) => this.sanitizeValue(value)).join(', ')})`) + .join(', ')})`, + ); + } + for (const row of nullRows) { + const conditions = row.map((value, index) => + value === null || value === undefined + ? `${escapedColumns[index]} IS NULL` + : `${escapedColumns[index]} = ${this.sanitizeValue(value)}`, + ); + predicates.push(`(${conditions.join(' AND ')})`); + } + return predicates.length === 1 ? predicates[0] : `(${predicates.join(' OR ')})`; + } + private buildWhereClause(conditions: Record): string { if (!conditions || typeof conditions !== 'object') { throw new Error('Invalid conditions: must be an object'); diff --git a/shared-code/src/data-access-layer/data-access-objects/data-access-object-mssql.ts b/shared-code/src/data-access-layer/data-access-objects/data-access-object-mssql.ts index b8aea4dd0..35f8b99e6 100644 --- a/shared-code/src/data-access-layer/data-access-objects/data-access-object-mssql.ts +++ b/shared-code/src/data-access-layer/data-access-objects/data-access-object-mssql.ts @@ -6,6 +6,12 @@ import * as csv from 'csv'; import { Knex } from 'knex'; import { nanoid } from 'nanoid'; import { LRUStorage } from '../../caching/lru-storage.js'; +import { + applyPrimaryKeysFilter, + BULK_PRIMARY_KEYS_LIMITS, + chunkPrimaryKeyRows, + normalisePrimaryKeys, +} from '../../helpers/bulk-primary-keys.js'; import { DAO_CONSTANTS } from '../../helpers/data-access-objects-constants.js'; import { ERROR_MESSAGES } from '../../helpers/errors/error-messages.js'; import { isMSSQLDateOrTimeType, isMSSQLDateStringByRegexp } from '../../helpers/is-database-date.js'; @@ -90,6 +96,9 @@ export class DataAccessObjectMssql extends BasicDataAccessObject implements IDat primaryKeys: Array>, settings: TableSettingsDS, ): Promise>> { + if (primaryKeys.length === 0) { + return []; + } const knex = await this.configureKnex(); const schemaName = await this.getSchemaName(tableName); tableName = `${schemaName}.[${tableName}]`; @@ -102,16 +111,13 @@ export class DataAccessObjectMssql extends BasicDataAccessObject implements IDat const query = knex(tableName).select(availableFields.length ? availableFields : '*'); - primaryKeys.forEach((primaryKey) => { - query.orWhere((builder) => { - Object.entries(primaryKey).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - - const results = await query; - return results as Array>; + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const results: Array> = []; + for (const rows of chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.mssql)) { + const chunkRows = await applyPrimaryKeysFilter(query.clone(), normalisedKeys.columns, rows, false); + results.push(...(chunkRows as Array>)); + } + return results; } public async getRowsFromTable( @@ -450,28 +456,26 @@ WHERE TABLE_TYPE = 'VIEW' } public async bulkDeleteRowsInTable(tableName: string, primaryKeys: Array>): Promise { - const [knex, schemaName] = await Promise.all([this.configureKnex(), this.getSchemaName(tableName)]); - const tableWithSchema = `${schemaName}.[${tableName}]`; - if (primaryKeys.length === 0) { return 0; } + const [knex, schemaName] = await Promise.all([this.configureKnex(), this.getSchemaName(tableName)]); + const tableWithSchema = `${schemaName}.[${tableName}]`; + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const chunks = chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.mssql); - await knex.transaction(async (trx) => { - await trx(tableWithSchema) - .delete() - .modify((queryBuilder) => { - primaryKeys.forEach((key) => { - queryBuilder.orWhere((builder) => { - Object.entries(key).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - }); + return await knex.transaction(async (trx) => { + let deletedRowsCount = 0; + for (const rows of chunks) { + deletedRowsCount += await applyPrimaryKeysFilter( + trx(tableWithSchema).delete(), + normalisedKeys.columns, + rows, + false, + ); + } + return deletedRowsCount; }); - - return primaryKeys.length; } public async validateSettings(settings: ValidateTableSettingsDS, tableName: string): Promise { diff --git a/shared-code/src/data-access-layer/data-access-objects/data-access-object-mysql.ts b/shared-code/src/data-access-layer/data-access-objects/data-access-object-mysql.ts index 27f925155..e81e961b7 100644 --- a/shared-code/src/data-access-layer/data-access-objects/data-access-object-mysql.ts +++ b/shared-code/src/data-access-layer/data-access-objects/data-access-object-mysql.ts @@ -6,6 +6,12 @@ import * as csv from 'csv'; import { Knex } from 'knex'; import { nanoid } from 'nanoid'; import { LRUStorage } from '../../caching/lru-storage.js'; +import { + applyPrimaryKeysFilter, + BULK_PRIMARY_KEYS_LIMITS, + chunkPrimaryKeyRows, + normalisePrimaryKeys, +} from '../../helpers/bulk-primary-keys.js'; import { checkFieldAutoincrement } from '../../helpers/check-field-autoincrement.js'; import { DAO_CONSTANTS } from '../../helpers/data-access-objects-constants.js'; import { ERROR_MESSAGES } from '../../helpers/errors/error-messages.js'; @@ -143,6 +149,9 @@ export class DataAccessObjectMysql extends BasicDataAccessObject implements IDat primaryKeys: Array>, settings: TableSettingsDS, ): Promise>> { + if (primaryKeys.length === 0) { + return []; + } const knex = await this.configureKnex(); let availableFields: string[] = []; @@ -153,16 +162,13 @@ export class DataAccessObjectMysql extends BasicDataAccessObject implements IDat const query = knex(tableName).select(availableFields.length ? availableFields : '*'); - primaryKeys.forEach((primaryKey) => { - query.orWhere((builder) => { - Object.entries(primaryKey).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - - const results = await query; - return results as Array>; + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const results: Array> = []; + for (const rows of chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.mysql)) { + const chunkRows = await applyPrimaryKeysFilter(query.clone(), normalisedKeys.columns, rows, true); + results.push(...(chunkRows as Array>)); + } + return results; } public async getRowsFromTable( @@ -520,27 +526,20 @@ export class DataAccessObjectMysql extends BasicDataAccessObject implements IDat } public async bulkDeleteRowsInTable(tableName: string, primaryKeys: Array>): Promise { - const knex = await this.configureKnex(); - if (primaryKeys.length === 0) { return 0; } + const knex = await this.configureKnex(); + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const chunks = chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.mysql); - await knex.transaction(async (trx) => { - await trx(tableName) - .delete() - .modify((queryBuilder) => { - primaryKeys.forEach((key) => { - queryBuilder.orWhere((builder) => { - Object.entries(key).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - }); + return await knex.transaction(async (trx) => { + let deletedRowsCount = 0; + for (const rows of chunks) { + deletedRowsCount += await applyPrimaryKeysFilter(trx(tableName).delete(), normalisedKeys.columns, rows, true); + } + return deletedRowsCount; }); - - return primaryKeys.length; } public async validateSettings(settings: ValidateTableSettingsDS, tableName: string): Promise { diff --git a/shared-code/src/data-access-layer/data-access-objects/data-access-object-oracle.ts b/shared-code/src/data-access-layer/data-access-objects/data-access-object-oracle.ts index 2bf0e7308..eb245c4d1 100644 --- a/shared-code/src/data-access-layer/data-access-objects/data-access-object-oracle.ts +++ b/shared-code/src/data-access-layer/data-access-objects/data-access-object-oracle.ts @@ -6,6 +6,12 @@ import * as csv from 'csv'; import { Knex } from 'knex'; import { nanoid } from 'nanoid'; import { LRUStorage } from '../../caching/lru-storage.js'; +import { + applyPrimaryKeysFilter, + BULK_PRIMARY_KEYS_LIMITS, + chunkPrimaryKeyRows, + normalisePrimaryKeys, +} from '../../helpers/bulk-primary-keys.js'; import { DAO_CONSTANTS } from '../../helpers/data-access-objects-constants.js'; import { ERROR_MESSAGES } from '../../helpers/errors/error-messages.js'; import { @@ -151,6 +157,9 @@ export class DataAccessObjectOracle extends BasicDataAccessObject implements IDa primaryKeys: Array>, settings: TableSettingsDS, ): Promise>> { + if (primaryKeys.length === 0) { + return []; + } const knex = await this.configureKnex(); const schema = this.connection.schema ?? this.connection.username.toUpperCase(); let availableFields: string[] = []; @@ -164,16 +173,13 @@ export class DataAccessObjectOracle extends BasicDataAccessObject implements IDa .withSchema(schema) .select(availableFields.length ? availableFields : '*'); - primaryKeys.forEach((primaryKey) => { - query.orWhere((builder) => { - Object.entries(primaryKey).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - - const results = await query; - return results as Array>; + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const results: Array> = []; + for (const rows of chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.oracle)) { + const chunkRows = await applyPrimaryKeysFilter(query.clone(), normalisedKeys.columns, rows, true); + results.push(...(chunkRows as Array>)); + } + return results; } public async getRowsFromTable( @@ -638,28 +644,27 @@ export class DataAccessObjectOracle extends BasicDataAccessObject implements IDa } public async bulkDeleteRowsInTable(tableName: string, primaryKeys: Array>): Promise { - const knex = await this.configureKnex(); - if (primaryKeys.length === 0) { return 0; } + const knex = await this.configureKnex(); + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const chunks = chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.oracle); - await knex.transaction(async (trx) => { - await trx(tableName) - .withSchema(this.connection.schema ?? this.connection.username.toUpperCase()) - .delete() - .modify((queryBuilder) => { - primaryKeys.forEach((key) => { - queryBuilder.orWhere((builder) => { - Object.entries(key).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - }); + return await knex.transaction(async (trx) => { + let deletedRowsCount = 0; + for (const rows of chunks) { + deletedRowsCount += await applyPrimaryKeysFilter( + trx(tableName) + .withSchema(this.connection.schema ?? this.connection.username.toUpperCase()) + .delete(), + normalisedKeys.columns, + rows, + true, + ); + } + return deletedRowsCount; }); - - return primaryKeys.length; } public async validateSettings(settings: ValidateTableSettingsDS, tableName: string): Promise { diff --git a/shared-code/src/data-access-layer/data-access-objects/data-access-object-postgres.ts b/shared-code/src/data-access-layer/data-access-objects/data-access-object-postgres.ts index cbc31c7b2..c24d183d3 100644 --- a/shared-code/src/data-access-layer/data-access-objects/data-access-object-postgres.ts +++ b/shared-code/src/data-access-layer/data-access-objects/data-access-object-postgres.ts @@ -6,6 +6,12 @@ import * as csv from 'csv'; import { Knex } from 'knex'; import { nanoid } from 'nanoid'; import { LRUStorage } from '../../caching/lru-storage.js'; +import { + applyPrimaryKeysFilter, + BULK_PRIMARY_KEYS_LIMITS, + chunkPrimaryKeyRows, + normalisePrimaryKeys, +} from '../../helpers/bulk-primary-keys.js'; import { DAO_CONSTANTS } from '../../helpers/data-access-objects-constants.js'; import { ERROR_MESSAGES } from '../../helpers/errors/error-messages.js'; import { isPostgresDateOrTimeType, isPostgresDateStringByRegexp } from '../../helpers/is-database-date.js'; @@ -119,6 +125,9 @@ export class DataAccessObjectPostgres extends BasicDataAccessObject implements I primaryKeys: Array>, settings: TableSettingsDS, ): Promise>> { + if (primaryKeys.length === 0) { + return []; + } const knex: Knex = await this.configureKnex(); let availableFields: string[] = []; if (settings) { @@ -130,16 +139,13 @@ export class DataAccessObjectPostgres extends BasicDataAccessObject implements I .withSchema(this.connection.schema ?? 'public') .select(availableFields.length ? availableFields : '*'); - primaryKeys.forEach((primaryKey) => { - query.orWhere((builder) => { - Object.entries(primaryKey).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - - const results = await query; - return results as Array>; + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const results: Array> = []; + for (const rows of chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.postgres)) { + const chunkRows = await applyPrimaryKeysFilter(query.clone(), normalisedKeys.columns, rows, true); + results.push(...(chunkRows as Array>)); + } + return results; } public async getRowsFromTable( @@ -526,28 +532,27 @@ export class DataAccessObjectPostgres extends BasicDataAccessObject implements I } public async bulkDeleteRowsInTable(tableName: string, primaryKeys: Array>): Promise { - const knex = await this.configureKnex(); - if (primaryKeys.length === 0) { return 0; } + const knex = await this.configureKnex(); + const normalisedKeys = normalisePrimaryKeys(primaryKeys); + const chunks = chunkPrimaryKeyRows(normalisedKeys, BULK_PRIMARY_KEYS_LIMITS.postgres); - await knex.transaction(async (trx) => { - await trx(tableName) - .withSchema(this.connection.schema ?? 'public') - .delete() - .modify((queryBuilder) => { - primaryKeys.forEach((key) => { - queryBuilder.orWhere((builder) => { - Object.entries(key).forEach(([column, value]) => { - builder.andWhere(column, value); - }); - }); - }); - }); + return await knex.transaction(async (trx) => { + let deletedRowsCount = 0; + for (const rows of chunks) { + deletedRowsCount += await applyPrimaryKeysFilter( + trx(tableName) + .withSchema(this.connection.schema ?? 'public') + .delete(), + normalisedKeys.columns, + rows, + true, + ); + } + return deletedRowsCount; }); - - return primaryKeys.length; } public async validateSettings(settings: ValidateTableSettingsDS, tableName: string): Promise { diff --git a/shared-code/src/helpers/bulk-primary-keys.ts b/shared-code/src/helpers/bulk-primary-keys.ts new file mode 100644 index 000000000..324f2c60c --- /dev/null +++ b/shared-code/src/helpers/bulk-primary-keys.ts @@ -0,0 +1,120 @@ +import { Knex } from 'knex'; +import { ERROR_MESSAGES } from './errors/error-messages.js'; + +/** + * Primary keys of a bulk operation, normalised to one column list and one value row per key, in + * the column order of the first key. Every key must carry exactly the same columns. + */ +export type NormalisedPrimaryKeys = { + columns: Array; + rows: Array>; +}; + +/** Per-dialect limits that bound how many keys a single statement may carry. */ +export type BulkPrimaryKeysLimits = { + /** Cap on keys per statement regardless of the number of key columns. */ + maxKeysPerStatement: number; + /** Cap on bound values per statement (every key column of every key binds one value). */ + maxBindingsPerStatement: number; +}; + +export const BULK_PRIMARY_KEYS_LIMITS = { + /** The extended-query protocol caps bind parameters at 65535. */ + postgres: { maxKeysPerStatement: 1000, maxBindingsPerStatement: 60000 }, + /** Prepared-statement placeholders cap at 65535. */ + mysql: { maxKeysPerStatement: 1000, maxBindingsPerStatement: 60000 }, + /** A request takes at most 2100 parameters. */ + mssql: { maxKeysPerStatement: 1000, maxBindingsPerStatement: 2000 }, + /** ORA-01795: an IN list takes at most 1000 expressions. */ + oracle: { maxKeysPerStatement: 1000, maxBindingsPerStatement: 1000 }, +} as const satisfies Record; + +export function normalisePrimaryKeys(primaryKeys: Array>): NormalisedPrimaryKeys { + if (primaryKeys.length === 0) { + return { columns: [], rows: [] }; + } + const columns = Object.keys(primaryKeys[0]); + if (columns.length === 0) { + throw new Error(ERROR_MESSAGES.PRIMARY_KEY_EMPTY); + } + const rows = primaryKeys.map((primaryKey) => { + const keyColumns = Object.keys(primaryKey); + if (keyColumns.length !== columns.length || !columns.every((column) => keyColumns.includes(column))) { + throw new Error(ERROR_MESSAGES.PRIMARY_KEYS_COLUMNS_MISMATCH); + } + // eslint-disable-next-line security/detect-object-injection + return columns.map((column) => primaryKey[column]); + }); + return { columns, rows }; +} + +/** Splits normalised key rows into statement-sized chunks for the given dialect limits. */ +export function chunkPrimaryKeyRows( + { columns, rows }: NormalisedPrimaryKeys, + limits: BulkPrimaryKeysLimits, +): Array>> { + if (rows.length === 0) { + return []; + } + const chunkSize = Math.max( + 1, + Math.min(limits.maxKeysPerStatement, Math.floor(limits.maxBindingsPerStatement / columns.length)), + ); + const chunks: Array>> = []; + for (let i = 0; i < rows.length; i += chunkSize) { + chunks.push(rows.slice(i, i + chunkSize)); + } + return chunks; +} + +/** + * Restricts the query to the given keys with one list predicate instead of one OR-ed group per key: + * `col IN (…)` for a single key column, `(a, b) IN ((…), (…))` for composite keys where the dialect + * has row-value constructors, and a grouped `(a = ? AND b = ?) OR …` where it has not (MSSQL). + * A key with a `null` component never matches through IN (`NULL IN (NULL)` is unknown), so such + * keys are always emitted as a group with `IS NULL` for the null columns. + */ +export function applyPrimaryKeysFilter( + queryBuilder: T, + columns: Array, + rows: Array>, + rowValueIn: boolean, +): T { + const listRows = rows.filter((row) => !hasNullComponent(row)); + const nullRows = rows.filter((row) => hasNullComponent(row)); + const useList = listRows.length > 0 && (columns.length === 1 || rowValueIn); + if (useList && nullRows.length === 0) { + return whereInKeys(queryBuilder, columns, listRows) as T; + } + return queryBuilder.where((group) => { + if (useList) { + whereInKeys(group, columns, listRows); + } + for (const row of useList ? nullRows : rows) { + group.orWhere((key) => { + columns.forEach((column, index) => { + // eslint-disable-next-line security/detect-object-injection + key.andWhere(column, row[index] as Knex.Value); + }); + }); + } + }) as T; +} + +function hasNullComponent(row: Array): boolean { + return row.some((value) => value === null); +} + +function whereInKeys( + queryBuilder: Knex.QueryBuilder, + columns: Array, + rows: Array>, +): Knex.QueryBuilder { + if (columns.length === 1) { + return queryBuilder.whereIn( + columns[0], + rows.map((row) => row[0] as Knex.Value), + ); + } + return queryBuilder.whereIn(columns, rows as Array>); +} diff --git a/shared-code/src/helpers/errors/error-messages.ts b/shared-code/src/helpers/errors/error-messages.ts index 9bbcde337..47cb3d9c4 100644 --- a/shared-code/src/helpers/errors/error-messages.ts +++ b/shared-code/src/helpers/errors/error-messages.ts @@ -11,4 +11,6 @@ export const ERROR_MESSAGES = { NO_DATA_RETURNED_FROM_AGENT: `No data returned from agent`, INVALID_OBJECT_ID_FORMAT: `Invalid object id format`, CLIENT_NOT_CONNECTED: `Client is not connected`, + PRIMARY_KEY_EMPTY: `Primary key must contain at least one column`, + PRIMARY_KEYS_COLUMNS_MISMATCH: `All primary keys in a bulk operation must contain the same columns`, };