From 32f0dfe7f8ebdc525bdd452427378911a76a5066 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 11:39:28 +0200 Subject: [PATCH 1/3] incremental-ingestion: use ANY(array) instead of IN(...) on Postgres MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The whereIn('ref', refs) calls on ingestion_mark_entities generated a unique prepared statement for every distinct array length, bloating the Postgres query plan cache. On Postgres, use = ANY($1) with a single array parameter instead. Falls back to regular whereIn on SQLite/MySQL. Signed-off-by: Fredrik Adelöw Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .changeset/incremental-ingestion-any-array.md | 5 + ...ncrementalIngestionDatabaseManager.test.ts | 121 +++++++++++++++++- .../IncrementalIngestionDatabaseManager.ts | 64 ++++++--- 3 files changed, 168 insertions(+), 22 deletions(-) create mode 100644 .changeset/incremental-ingestion-any-array.md diff --git a/.changeset/incremental-ingestion-any-array.md b/.changeset/incremental-ingestion-any-array.md new file mode 100644 index 0000000000..9c7379ecae --- /dev/null +++ b/.changeset/incremental-ingestion-any-array.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch +--- + +On PostgreSQL, `WHERE ref IN ($1, $2, ..., $N)` queries on the `ingestion_mark_entities` table now use `= ANY($1)` with a single array parameter instead. This reduces prepared statement bloat in the query plan cache when the number of entity refs varies between calls. diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts index 035fc8d419..155b45cacb 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts @@ -121,5 +121,124 @@ describe.each(databases.eachSupportedId())( expect(result.total).toBe(3); expect(typeof result.total).toBe('number'); }); - }, + + it('createMarkEntities handles existing and new refs correctly', async () => { + const knex = await databases.init(databaseId); + await knex.migrate.latest({ directory: migrationsDir }); + + const manager = new IncrementalIngestionDatabaseManager({ client: knex }); + const { ingestionId } = (await manager.createProviderIngestionRecord( + 'testProvider', + ))!; + + const markId1 = uuid(); + await manager.createMark({ + record: { + id: markId1, + ingestion_id: ingestionId, + sequence: 1, + cursor: { data: 1 }, + }, + }); + + const makeEntity = (name: string): DeferredEntity => ({ + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { namespace: 'default', name }, + }, + }); + + // First batch: create 3 entities + await manager.createMarkEntities(markId1, [ + makeEntity('a'), + makeEntity('b'), + makeEntity('c'), + ]); + + const rows1 = await knex('ingestion_mark_entities').select('ref'); + expect(rows1).toHaveLength(3); + + // Second batch with overlap: b and c already exist, d is new. + // Existing refs should be updated to the new mark, new refs inserted. + const markId2 = uuid(); + await manager.createMark({ + record: { + id: markId2, + ingestion_id: ingestionId, + sequence: 2, + cursor: { data: 2 }, + }, + }); + + await manager.createMarkEntities(markId2, [ + makeEntity('b'), + makeEntity('c'), + makeEntity('d'), + ]); + + const rows2 = await knex('ingestion_mark_entities') + .select('ref', 'ingestion_mark_id') + .orderBy('ref'); + expect(rows2).toHaveLength(4); + + // a stays on markId1, b and c moved to markId2, d is new on markId2 + expect( + rows2.find(r => r.ref === 'component:default/a')?.ingestion_mark_id, + ).toBe(markId1); + expect( + rows2.find(r => r.ref === 'component:default/b')?.ingestion_mark_id, + ).toBe(markId2); + expect( + rows2.find(r => r.ref === 'component:default/c')?.ingestion_mark_id, + ).toBe(markId2); + expect( + rows2.find(r => r.ref === 'component:default/d')?.ingestion_mark_id, + ).toBe(markId2); + }); + + it('deleteEntityRecordsByRef removes matching refs', async () => { + const knex = await databases.init(databaseId); + await knex.migrate.latest({ directory: migrationsDir }); + + const manager = new IncrementalIngestionDatabaseManager({ client: knex }); + const { ingestionId } = (await manager.createProviderIngestionRecord( + 'testProvider', + ))!; + + const markId = uuid(); + await manager.createMark({ + record: { + id: markId, + ingestion_id: ingestionId, + sequence: 1, + cursor: { data: 1 }, + }, + }); + + const makeEntity = (name: string): DeferredEntity => ({ + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { namespace: 'default', name }, + }, + }); + + await manager.createMarkEntities(markId, [ + makeEntity('x'), + makeEntity('y'), + makeEntity('z'), + ]); + + // Delete two of the three + await manager.deleteEntityRecordsByRef([ + { entityRef: 'component:default/x' }, + { entityRef: 'component:default/z' }, + ]); + + const remaining = await knex('ingestion_mark_entities').select('ref'); + expect(remaining).toHaveLength(1); + expect(remaining[0].ref).toBe('component:default/y'); + }); +}, ); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts index ff1b98508f..78dc646d72 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -34,6 +34,17 @@ export class IncrementalIngestionDatabaseManager { this.client = options.client; } + private whereInArray( + query: Knex.QueryBuilder, + column: string, + values: string[], + ): Knex.QueryBuilder { + if (this.client.client.config.client === 'pg') { + return query.whereRaw('?? = ANY(?)', [column, values]); + } + return query.whereIn(column, values); + } + /** * Performs an update to the ingestion record with matching `id`. * @param options - IngestionRecordUpdate @@ -76,24 +87,25 @@ export class IncrementalIngestionDatabaseManager { tx: Knex.Transaction, ids: { id: string }[], ) { - const chunks: { id: string }[][] = []; - for (let i = 0; i < ids.length; i += 100) { - const chunk = ids.slice(i, i + 100); - chunks.push(chunk); + if (ids.length === 0) { + return 0; + } + + const allIds = ids.map(entry => entry.id); + + if (this.client.client.config.client === 'pg') { + return await tx('ingestion_mark_entities') + .delete() + .whereRaw('?? = ANY(?)', ['id', allIds]); } let deleted = 0; - - for (const chunk of chunks) { - const chunkDeleted = await tx('ingestion_mark_entities') + for (let i = 0; i < allIds.length; i += 100) { + const chunk = allIds.slice(i, i + 100); + deleted += await tx('ingestion_mark_entities') .delete() - .whereIn( - 'id', - chunk.map(entry => entry.id), - ); - deleted += chunkDeleted; + .whereIn('id', chunk); } - return deleted; } @@ -276,7 +288,11 @@ export class IncrementalIngestionDatabaseManager { async deleteEntityRecordsByRef(entities: { entityRef: string }[]) { const refs = entities.map(e => e.entityRef); await this.client.transaction(async tx => { - await tx('ingestion_mark_entities').delete().whereIn('ref', refs); + await this.whereInArray( + tx('ingestion_mark_entities').delete(), + 'ref', + refs, + ); }); } @@ -601,18 +617,24 @@ export class IncrementalIngestionDatabaseManager { await this.client.transaction(async tx => { const existingRefsArray = ( - await tx<{ ref: string }>('ingestion_mark_entities') - .select('ref') - .whereIn('ref', refs) - ).map(e => e.ref); + await this.whereInArray( + tx<{ ref: string }>('ingestion_mark_entities').select('ref'), + 'ref', + refs, + ) + ).map((e: { ref: string }) => e.ref); const existingRefsSet = new Set(existingRefsArray); const newRefs = refs.filter(e => !existingRefsSet.has(e)); - await tx('ingestion_mark_entities') - .update('ingestion_mark_id', markId) - .whereIn('ref', existingRefsArray); + if (existingRefsArray.length > 0) { + await this.whereInArray( + tx('ingestion_mark_entities').update('ingestion_mark_id', markId), + 'ref', + existingRefsArray, + ); + } if (newRefs.length > 0) { await tx('ingestion_mark_entities').insert( From 0f8c977018bf4e2bd2053bbfc17a0f2ce1a9cebf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 13:30:21 +0200 Subject: [PATCH 2/3] Reuse whereInArray helper in deleteMarkEntities and fix test formatting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- ...ncrementalIngestionDatabaseManager.test.ts | 228 +++++++++--------- .../IncrementalIngestionDatabaseManager.ts | 8 +- 2 files changed, 119 insertions(+), 117 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts index 155b45cacb..91975edfb5 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.test.ts @@ -122,123 +122,123 @@ describe.each(databases.eachSupportedId())( expect(typeof result.total).toBe('number'); }); - it('createMarkEntities handles existing and new refs correctly', async () => { - const knex = await databases.init(databaseId); - await knex.migrate.latest({ directory: migrationsDir }); + it('createMarkEntities handles existing and new refs correctly', async () => { + const knex = await databases.init(databaseId); + await knex.migrate.latest({ directory: migrationsDir }); - const manager = new IncrementalIngestionDatabaseManager({ client: knex }); - const { ingestionId } = (await manager.createProviderIngestionRecord( - 'testProvider', - ))!; + const manager = new IncrementalIngestionDatabaseManager({ client: knex }); + const { ingestionId } = (await manager.createProviderIngestionRecord( + 'testProvider', + ))!; - const markId1 = uuid(); - await manager.createMark({ - record: { - id: markId1, - ingestion_id: ingestionId, - sequence: 1, - cursor: { data: 1 }, - }, + const markId1 = uuid(); + await manager.createMark({ + record: { + id: markId1, + ingestion_id: ingestionId, + sequence: 1, + cursor: { data: 1 }, + }, + }); + + const makeEntity = (name: string): DeferredEntity => ({ + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { namespace: 'default', name }, + }, + }); + + // First batch: create 3 entities + await manager.createMarkEntities(markId1, [ + makeEntity('a'), + makeEntity('b'), + makeEntity('c'), + ]); + + const rows1 = await knex('ingestion_mark_entities').select('ref'); + expect(rows1).toHaveLength(3); + + // Second batch with overlap: b and c already exist, d is new. + // Existing refs should be updated to the new mark, new refs inserted. + const markId2 = uuid(); + await manager.createMark({ + record: { + id: markId2, + ingestion_id: ingestionId, + sequence: 2, + cursor: { data: 2 }, + }, + }); + + await manager.createMarkEntities(markId2, [ + makeEntity('b'), + makeEntity('c'), + makeEntity('d'), + ]); + + const rows2 = await knex('ingestion_mark_entities') + .select('ref', 'ingestion_mark_id') + .orderBy('ref'); + expect(rows2).toHaveLength(4); + + // a stays on markId1, b and c moved to markId2, d is new on markId2 + expect( + rows2.find(r => r.ref === 'component:default/a')?.ingestion_mark_id, + ).toBe(markId1); + expect( + rows2.find(r => r.ref === 'component:default/b')?.ingestion_mark_id, + ).toBe(markId2); + expect( + rows2.find(r => r.ref === 'component:default/c')?.ingestion_mark_id, + ).toBe(markId2); + expect( + rows2.find(r => r.ref === 'component:default/d')?.ingestion_mark_id, + ).toBe(markId2); }); - const makeEntity = (name: string): DeferredEntity => ({ - entity: { - apiVersion: 'backstage.io/v1alpha1', - kind: 'Component', - metadata: { namespace: 'default', name }, - }, + it('deleteEntityRecordsByRef removes matching refs', async () => { + const knex = await databases.init(databaseId); + await knex.migrate.latest({ directory: migrationsDir }); + + const manager = new IncrementalIngestionDatabaseManager({ client: knex }); + const { ingestionId } = (await manager.createProviderIngestionRecord( + 'testProvider', + ))!; + + const markId = uuid(); + await manager.createMark({ + record: { + id: markId, + ingestion_id: ingestionId, + sequence: 1, + cursor: { data: 1 }, + }, + }); + + const makeEntity = (name: string): DeferredEntity => ({ + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { namespace: 'default', name }, + }, + }); + + await manager.createMarkEntities(markId, [ + makeEntity('x'), + makeEntity('y'), + makeEntity('z'), + ]); + + // Delete two of the three + await manager.deleteEntityRecordsByRef([ + { entityRef: 'component:default/x' }, + { entityRef: 'component:default/z' }, + ]); + + const remaining = await knex('ingestion_mark_entities').select('ref'); + expect(remaining).toHaveLength(1); + expect(remaining[0].ref).toBe('component:default/y'); }); - - // First batch: create 3 entities - await manager.createMarkEntities(markId1, [ - makeEntity('a'), - makeEntity('b'), - makeEntity('c'), - ]); - - const rows1 = await knex('ingestion_mark_entities').select('ref'); - expect(rows1).toHaveLength(3); - - // Second batch with overlap: b and c already exist, d is new. - // Existing refs should be updated to the new mark, new refs inserted. - const markId2 = uuid(); - await manager.createMark({ - record: { - id: markId2, - ingestion_id: ingestionId, - sequence: 2, - cursor: { data: 2 }, - }, - }); - - await manager.createMarkEntities(markId2, [ - makeEntity('b'), - makeEntity('c'), - makeEntity('d'), - ]); - - const rows2 = await knex('ingestion_mark_entities') - .select('ref', 'ingestion_mark_id') - .orderBy('ref'); - expect(rows2).toHaveLength(4); - - // a stays on markId1, b and c moved to markId2, d is new on markId2 - expect( - rows2.find(r => r.ref === 'component:default/a')?.ingestion_mark_id, - ).toBe(markId1); - expect( - rows2.find(r => r.ref === 'component:default/b')?.ingestion_mark_id, - ).toBe(markId2); - expect( - rows2.find(r => r.ref === 'component:default/c')?.ingestion_mark_id, - ).toBe(markId2); - expect( - rows2.find(r => r.ref === 'component:default/d')?.ingestion_mark_id, - ).toBe(markId2); - }); - - it('deleteEntityRecordsByRef removes matching refs', async () => { - const knex = await databases.init(databaseId); - await knex.migrate.latest({ directory: migrationsDir }); - - const manager = new IncrementalIngestionDatabaseManager({ client: knex }); - const { ingestionId } = (await manager.createProviderIngestionRecord( - 'testProvider', - ))!; - - const markId = uuid(); - await manager.createMark({ - record: { - id: markId, - ingestion_id: ingestionId, - sequence: 1, - cursor: { data: 1 }, - }, - }); - - const makeEntity = (name: string): DeferredEntity => ({ - entity: { - apiVersion: 'backstage.io/v1alpha1', - kind: 'Component', - metadata: { namespace: 'default', name }, - }, - }); - - await manager.createMarkEntities(markId, [ - makeEntity('x'), - makeEntity('y'), - makeEntity('z'), - ]); - - // Delete two of the three - await manager.deleteEntityRecordsByRef([ - { entityRef: 'component:default/x' }, - { entityRef: 'component:default/z' }, - ]); - - const remaining = await knex('ingestion_mark_entities').select('ref'); - expect(remaining).toHaveLength(1); - expect(remaining[0].ref).toBe('component:default/y'); - }); -}, + }, ); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts index 78dc646d72..ecf821364f 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -94,9 +94,11 @@ export class IncrementalIngestionDatabaseManager { const allIds = ids.map(entry => entry.id); if (this.client.client.config.client === 'pg') { - return await tx('ingestion_mark_entities') - .delete() - .whereRaw('?? = ANY(?)', ['id', allIds]); + return await this.whereInArray( + tx('ingestion_mark_entities').delete(), + 'id', + allIds, + ); } let deleted = 0; From 624c39c93c5a22ede95ec81ad13e495ed0091318 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 19 May 2026 13:37:49 +0200 Subject: [PATCH 3/3] Refactor whereInArray to use knex .modify() for linear query chains MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 (1M context) Signed-off-by: Fredrik Adelöw --- .../IncrementalIngestionDatabaseManager.ts | 50 ++++++++----------- 1 file changed, 21 insertions(+), 29 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts index ecf821364f..4a290d23a9 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -34,15 +34,15 @@ export class IncrementalIngestionDatabaseManager { this.client = options.client; } - private whereInArray( - query: Knex.QueryBuilder, - column: string, - values: string[], - ): Knex.QueryBuilder { - if (this.client.client.config.client === 'pg') { - return query.whereRaw('?? = ANY(?)', [column, values]); - } - return query.whereIn(column, values); + private whereInArray(column: string, values: string[]) { + const isPg = this.client.client.config.client === 'pg'; + return (qb: Knex.QueryBuilder) => { + if (isPg) { + qb.whereRaw('?? = ANY(?)', [column, values]); + } else { + qb.whereIn(column, values); + } + }; } /** @@ -94,11 +94,9 @@ export class IncrementalIngestionDatabaseManager { const allIds = ids.map(entry => entry.id); if (this.client.client.config.client === 'pg') { - return await this.whereInArray( - tx('ingestion_mark_entities').delete(), - 'id', - allIds, - ); + return await tx('ingestion_mark_entities') + .delete() + .modify(this.whereInArray('id', allIds)); } let deleted = 0; @@ -290,11 +288,9 @@ export class IncrementalIngestionDatabaseManager { async deleteEntityRecordsByRef(entities: { entityRef: string }[]) { const refs = entities.map(e => e.entityRef); await this.client.transaction(async tx => { - await this.whereInArray( - tx('ingestion_mark_entities').delete(), - 'ref', - refs, - ); + await tx('ingestion_mark_entities') + .delete() + .modify(this.whereInArray('ref', refs)); }); } @@ -619,11 +615,9 @@ export class IncrementalIngestionDatabaseManager { await this.client.transaction(async tx => { const existingRefsArray = ( - await this.whereInArray( - tx<{ ref: string }>('ingestion_mark_entities').select('ref'), - 'ref', - refs, - ) + await tx<{ ref: string }>('ingestion_mark_entities') + .select('ref') + .modify(this.whereInArray('ref', refs)) ).map((e: { ref: string }) => e.ref); const existingRefsSet = new Set(existingRefsArray); @@ -631,11 +625,9 @@ export class IncrementalIngestionDatabaseManager { const newRefs = refs.filter(e => !existingRefsSet.has(e)); if (existingRefsArray.length > 0) { - await this.whereInArray( - tx('ingestion_mark_entities').update('ingestion_mark_id', markId), - 'ref', - existingRefsArray, - ); + await tx('ingestion_mark_entities') + .update('ingestion_mark_id', markId) + .modify(this.whereInArray('ref', existingRefsArray)); } if (newRefs.length > 0) {