Merge pull request #27718 from backstage/freben/better-deltas

compute deltas based on hashes just like full mutations

.changeset/itchy-rats-pump.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': patch
+---
+
+Compute deltas more efficiently, which generally leads to less wasted processing cycles

plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts

@@ -19,13 +19,14 @@ import {
   TestDatabaseId,
   TestDatabases,
 } from '@backstage/backend-test-utils';
-import { Entity } from '@backstage/catalog-model';
+import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import { Knex } from 'knex';
 import * as uuid from 'uuid';
 import { DefaultProviderDatabase } from './DefaultProviderDatabase';
 import { applyDatabaseMigrations } from './migrations';
 import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables';
 import { LoggerService } from '@backstage/backend-plugin-api';
+import { generateStableHash } from './util';
 
 jest.setTimeout(60_000);
 
@@ -690,13 +691,8 @@ describe('DefaultProviderDatabase', () => {
     it.each(databases.eachSupportedId())(
       'should successfully fall back from batch to individual mode on conflicts, %p',
       async databaseId => {
-        const fakeLogger = {
-          debug: jest.fn(),
-        };
-        const { knex, db } = await createDatabase(
-          databaseId,
-          fakeLogger as any,
-        );
+        const fakeLogger = mockServices.logger.mock();
+        const { knex, db } = await createDatabase(databaseId, fakeLogger);
 
         await createLocations(knex, ['component:default/a']);
 
@@ -744,11 +740,8 @@ describe('DefaultProviderDatabase', () => {
     it.each(databases.eachSupportedId())(
       'should gracefully handle accidental duplicate refresh state references when deletion happens during a full sync, %p',
       async databaseId => {
-        const fakeLogger = { debug: jest.fn() };
-        const { knex, db } = await createDatabase(
-          databaseId,
-          fakeLogger as any,
-        );
+        const fakeLogger = mockServices.logger.mock();
+        const { knex, db } = await createDatabase(databaseId, fakeLogger);
 
         await createLocations(knex, ['component:default/a']);
 
@@ -773,5 +766,162 @@ describe('DefaultProviderDatabase', () => {
         expect(state).toEqual([]);
       },
     );
+
+    it.each(databases.eachSupportedId())(
+      'should properly translate deltas into add/update/remove, %p',
+      async databaseId => {
+        const fakeLogger = mockServices.logger.mock();
+        const { knex, db } = await createDatabase(databaseId, fakeLogger);
+
+        const entity1Before: Entity = {
+          apiVersion: '1',
+          kind: 'k',
+          metadata: { namespace: 'ns', name: 'n1' },
+        };
+        const entity1After: Entity = {
+          ...entity1Before,
+          apiVersion: '2',
+        };
+
+        const entity2Before: Entity = {
+          apiVersion: '1',
+          kind: 'k',
+          metadata: { namespace: 'ns', name: 'n2' },
+        };
+        const entity2After: Entity = {
+          ...entity2Before,
+          apiVersion: '2',
+        };
+
+        const entity3: Entity = {
+          apiVersion: '1',
+          kind: 'k',
+          metadata: { namespace: 'ns', name: 'n3' },
+        };
+
+        const entity4: Entity = {
+          apiVersion: '1',
+          kind: 'k',
+          metadata: { namespace: 'ns', name: 'n4' },
+        };
+
+        await insertRefreshStateRow(knex, {
+          entity_id: 'id1',
+          entity_ref: stringifyEntityRef(entity1Before),
+          last_discovery_at: new Date(),
+          next_update_at: new Date(),
+          errors: '[]',
+          unprocessed_entity: JSON.stringify(entity1Before),
+          unprocessed_hash: generateStableHash(entity1Before),
+        });
+        await insertRefRow(knex, {
+          source_key: 'my-provider',
+          target_entity_ref: stringifyEntityRef(entity1Before),
+        });
+        await insertRefreshStateRow(knex, {
+          entity_id: 'id2',
+          entity_ref: stringifyEntityRef(entity2Before),
+          last_discovery_at: new Date(),
+          next_update_at: new Date(),
+          errors: '[]',
+          unprocessed_entity: JSON.stringify(entity2Before),
+          unprocessed_hash: generateStableHash(entity2After), // lie about the hash!
+        });
+        await insertRefRow(knex, {
+          source_key: 'my-provider',
+          target_entity_ref: stringifyEntityRef(entity2Before),
+        });
+        await insertRefreshStateRow(knex, {
+          entity_id: 'id4',
+          entity_ref: stringifyEntityRef(entity4),
+          last_discovery_at: new Date(),
+          next_update_at: new Date(),
+          errors: '[]',
+          unprocessed_entity: JSON.stringify(entity4),
+          unprocessed_hash: generateStableHash(entity4),
+        });
+        await insertRefRow(knex, {
+          source_key: 'my-provider',
+          target_entity_ref: stringifyEntityRef(entity4),
+        });
+
+        await db.transaction(async tx => {
+          await db.replaceUnprocessedEntities(tx, {
+            type: 'delta',
+            sourceKey: 'my-provider',
+            added: [
+              // we used the right hashes for entity1, so this will turn into an update
+              { entity: entity1After },
+              // we lied about the hash for entity2, so this will become a no-op
+              { entity: entity2After },
+              // this didn't exist, so will become an add
+              { entity: entity3 },
+            ],
+            removed: [{ entityRef: stringifyEntityRef(entity4) }],
+          });
+        });
+
+        const state = await knex<DbRefreshStateRow>('refresh_state')
+          .select(['entity_ref', 'unprocessed_entity', 'unprocessed_hash'])
+          .orderBy('entity_ref');
+
+        expect(state).toEqual([
+          {
+            entity_ref: stringifyEntityRef(entity1After),
+            unprocessed_entity: JSON.stringify(entity1After),
+            unprocessed_hash: generateStableHash(entity1After),
+          },
+          {
+            entity_ref: stringifyEntityRef(entity2After),
+            unprocessed_entity: JSON.stringify(entity2Before), // didn't change
+            unprocessed_hash: generateStableHash(entity2After),
+          },
+          {
+            entity_ref: stringifyEntityRef(entity3),
+            unprocessed_entity: JSON.stringify(entity3),
+            unprocessed_hash: generateStableHash(entity3),
+          },
+        ]);
+      },
+    );
+
+    it.each(databases.eachSupportedId())(
+      'can handle large deltas without exploding, %p',
+      async databaseId => {
+        const fakeLogger = mockServices.logger.mock();
+        const { knex, db } = await createDatabase(databaseId, fakeLogger);
+
+        const count = 10000;
+        const padded = (n: number) => String(n).padStart(8, '0');
+
+        const entities = Array.from({ length: count }, (_, i) => ({
+          entity: {
+            apiVersion: '1',
+            kind: 'k',
+            metadata: { namespace: 'ns', name: padded(i) },
+          },
+        }));
+
+        await db.transaction(async tx => {
+          await db.replaceUnprocessedEntities(tx, {
+            type: 'delta',
+            sourceKey: 'my-provider',
+            added: entities,
+            removed: [],
+          });
+        });
+
+        const state = await knex<DbRefreshStateRow>('refresh_state')
+          .select(['entity_ref', 'unprocessed_entity', 'unprocessed_hash'])
+          .orderBy('entity_ref');
+
+        expect(state).toHaveLength(count);
+        expect(state[0]).toEqual({
+          entity_ref: stringifyEntityRef(entities[0].entity),
+          unprocessed_entity: JSON.stringify(entities[0].entity),
+          unprocessed_hash: generateStableHash(entities[0].entity),
+        });
+      },
+    );
   });
 });

plugins/catalog-backend/src/database/DefaultProviderDatabase.ts

@@ -75,10 +75,10 @@ export class DefaultProviderDatabase implements ProviderDatabase {
   }
 
   async replaceUnprocessedEntities(
-    txOpaque: Transaction,
+    txOpaque: Knex | Transaction,
     options: ReplaceUnprocessedEntitiesOptions,
   ): Promise<void> {
-    const tx = txOpaque as Knex.Transaction;
+    const tx = txOpaque as Knex | Knex.Transaction;
     const { toAdd, toUpsert, toRemove } = await this.createDelta(tx, options);
 
     if (toRemove.length) {
@@ -205,7 +205,7 @@ export class DefaultProviderDatabase implements ProviderDatabase {
   }
 
   private async createDelta(
-    tx: Knex.Transaction,
+    tx: Knex | Knex.Transaction,
     options: ReplaceUnprocessedEntitiesOptions,
   ): Promise<{
     toAdd: { deferred: DeferredEntity; hash: string }[];
@@ -213,14 +213,32 @@ export class DefaultProviderDatabase implements ProviderDatabase {
     toRemove: string[];
   }> {
     if (options.type === 'delta') {
-      return {
-        toAdd: [],
-        toUpsert: options.added.map(e => ({
-          deferred: e,
-          hash: generateStableHash(e.entity),
-        })),
-        toRemove: options.removed.map(e => e.entityRef),
-      };
+      const toAdd = new Array<{ deferred: DeferredEntity; hash: string }>();
+      const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>();
+      const toRemove = options.removed.map(e => e.entityRef);
+
+      for (const chunk of lodash.chunk(options.added, 1000)) {
+        const entityRefs = chunk.map(e => stringifyEntityRef(e.entity));
+        const rows = await tx<DbRefreshStateRow>('refresh_state')
+          .select(['entity_ref', 'unprocessed_hash'])
+          .whereIn('entity_ref', entityRefs);
+        const oldHashes = new Map(
+          rows.map(row => [row.entity_ref, row.unprocessed_hash]),
+        );
+
+        chunk.forEach((deferred, i) => {
+          const entityRef = entityRefs[i];
+          const newHash = generateStableHash(deferred.entity);
+          const oldHash = oldHashes.get(entityRef);
+          if (oldHash === undefined) {
+            toAdd.push({ deferred, hash: newHash });
+          } else if (newHash !== oldHash) {
+            toUpsert.push({ deferred, hash: newHash });
+          }
+        });
+      }
+
+      return { toAdd, toUpsert, toRemove };
     }
 
     // Grab all of the existing references from the same source, and their locationKeys as well

plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts

@@ -24,7 +24,7 @@ import { DbRefreshStateRow } from '../../tables';
  * @returns The conflicting key if there is one.
  */
 export async function checkLocationKeyConflict(options: {
-  tx: Knex.Transaction;
+  tx: Knex | Knex.Transaction;
   entityRef: string;
   locationKey?: string;
 }): Promise<string | undefined> {

plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts

@@ -28,7 +28,7 @@ import {
  * true if successful and false if there was a conflict.
  */
 export async function insertUnprocessedEntity(options: {
-  tx: Knex.Transaction;
+  tx: Knex | Knex.Transaction;
   entity: Entity;
   hash: string;
   locationKey?: string;

plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts

@@ -25,7 +25,7 @@ import { DbRefreshStateRow } from '../../tables';
  * Updating the entity will also cause it to be scheduled for immediate processing.
  */
 export async function updateUnprocessedEntity(options: {
-  tx: Knex.Transaction;
+  tx: Knex | Knex.Transaction;
   entity: Entity;
   hash: string;
   locationKey?: string;

plugins/catalog-backend/src/tests/performance/providerDeltaMutations.test.ts

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { TestDatabases, mockServices } from '@backstage/backend-test-utils';
+import { Knex } from 'knex';
+import { applyDatabaseMigrations } from '../../database/migrations';
+import { describePerformanceTest, performanceTraceEnabled } from './lib/env';
+import { DefaultProviderDatabase } from '../../database/DefaultProviderDatabase';
+import { DeferredEntity } from '@backstage/plugin-catalog-node';
+
+// #region Helpers
+
+jest.setTimeout(600_000);
+
+const databases = TestDatabases.create({
+  ids: [/* 'MYSQL_8', */ 'POSTGRES_16', /* 'POSTGRES_12',*/ 'SQLITE_3'],
+  disableDocker: false,
+});
+
+const traceLog: typeof console.log = performanceTraceEnabled
+  ? console.log
+  : () => {};
+
+// #endregion
+// #region Tests
+
+describePerformanceTest('providerDeltaMutations', () => {
+  let knex: Knex;
+
+  describe.each(databases.eachSupportedId())('%p', databaseId => {
+    beforeAll(async () => {
+      knex = await databases.init(databaseId);
+      await applyDatabaseMigrations(knex);
+    });
+
+    afterAll(async () => {
+      await knex.destroy();
+    });
+
+    it.each([200, 50_000])(
+      'inserts and then overwrites identical sets using delta mutations, batch size %p',
+      async batchSize => {
+        const sut = new DefaultProviderDatabase({
+          database: knex,
+          logger: mockServices.logger.mock(),
+        });
+
+        const sourceKey = 'source-key';
+        const entities: DeferredEntity[] = Array.from(
+          { length: batchSize },
+          (_, i) => ({
+            entity: {
+              apiVersion: 'backstage.io/v1alpha1',
+              kind: 'Component',
+              metadata: {
+                name: `component-${i}`,
+                namespace: 'default',
+              },
+              spec: {
+                type: 'service',
+                lifecycle: 'production',
+              },
+            },
+          }),
+        );
+
+        let start = Date.now();
+        await sut.replaceUnprocessedEntities(knex, {
+          type: 'delta',
+          sourceKey,
+          added: entities,
+          removed: [],
+        });
+        let perSecond = Math.round(batchSize / ((Date.now() - start) / 1000));
+        traceLog(
+          `${databaseId} inserted ${perSecond} entities per second at a batch size of ${batchSize}`,
+        );
+
+        start = Date.now();
+        const rounds = 20;
+        for (let i = 0; i < rounds; i++) {
+          await sut.replaceUnprocessedEntities(knex, {
+            type: 'delta',
+            sourceKey,
+            added: entities,
+            removed: [],
+          });
+        }
+        perSecond = Math.round(
+          (batchSize * rounds) / ((Date.now() - start) / 1000),
+        );
+        traceLog(
+          `${databaseId} updated ${perSecond} entities per second at a batch size of ${batchSize}`,
+        );
+
+        expect(true).toBe(true);
+      },
+    );
+  });
+});
+
+// #endregion