Merge pull request #27341 from backstage/blam/move-refresh-state

catalog: Add `entity_ref` to the `final_entities` table

.changeset/tender-feet-scream.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': minor
+---
+
+Added `entity_ref` column to `final_entities` in order to move `refresh_state` away from the read path

plugins/catalog-backend/migrations/20241024104700_add_entity_ref_to_final_entities.js

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+// @ts-check
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('final_entities', table => {
+    table
+      .string('entity_ref')
+      .nullable()
+      .comment('The entity reference of the entity');
+  });
+
+  await knex
+    .update({
+      entity_ref: knex
+        .select('r.entity_ref')
+        .from('refresh_state as r')
+        .where('r.entity_id', knex.raw('f.entity_id')),
+    })
+    .table('final_entities as f');
+
+  await knex.schema.alterTable('final_entities', table => {
+    table.string('entity_ref').notNullable().alter({ alterNullable: true });
+    table.unique(['entity_ref'], {
+      indexName: 'final_entities_entity_ref_uniq',
+    });
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = async function down(knex) {
+  await knex.schema.alterTable('final_entities', table => {
+    table.dropUnique([], 'final_entities_entity_ref_uniq');
+    table.dropColumn('entity_ref');
+  });
+};

plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts

@@ -252,24 +252,28 @@ describe('markForStitching', () => {
         {
           entity_id: '1',
           final_entity: '{}',
+          entity_ref: 'k:ns/one',
           hash: 'old',
           stitch_ticket: 'old',
         },
         {
           entity_id: '2',
           final_entity: '{}',
+          entity_ref: 'k:ns/two',
           hash: 'old',
           stitch_ticket: 'old',
         },
         {
           entity_id: '3',
           final_entity: '{}',
+          entity_ref: 'k:ns/three',
           hash: 'old',
           stitch_ticket: 'old',
         },
         {
           entity_id: '4',
           final_entity: '{}',
+          entity_ref: 'k:ns/four',
           hash: 'old',
           stitch_ticket: 'old',
         },

plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts

@@ -69,6 +69,7 @@ export async function performStitching(options: {
     .insert({
       entity_id: entityResult[0].entity_id,
       hash: '',
+      entity_ref: entityRef,
       stitch_ticket: stitchTicket,
     })
     .onConflict('entity_id')
@@ -212,6 +213,7 @@ export async function performStitching(options: {
       final_entity: JSON.stringify(entity),
       hash,
       last_updated_at: knex.fn.now(),
+      entity_ref: entityRef,
     })
     .where('entity_id', entityId)
     .where('stitch_ticket', stitchTicket)

plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts

@@ -69,6 +69,7 @@ describe('deleteOrphanedEntities', () => {
       await knex<DbFinalEntitiesRow>('final_entities').insert({
         entity_id: `id-${ref}`,
         hash: 'original',
+        entity_ref: ref,
         stitch_ticket: '',
       });
     }

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

@@ -179,6 +179,7 @@ export type DbFinalEntitiesRow = {
   stitch_ticket: string;
   final_entity?: string;
   last_updated_at?: string | Date;
+  entity_ref: string;
 };
 
 export type DbSearchRow = {

plugins/catalog-backend/src/providers/DefaultLocationStore.test.ts

@@ -204,6 +204,7 @@ describe('DefaultLocationStore', () => {
           hash: 'hash',
           last_updated_at: new Date(),
           stitch_ticket: '',
+          entity_ref: 'k:ns/n',
         });
 
         await knex<DbSearchRow>('search').insert({

plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts

@@ -76,6 +76,7 @@ describe('DefaultEntitiesCatalog', () => {
 
     await knex<DbFinalEntitiesRow>('final_entities').insert({
       entity_id: id,
+      entity_ref: entityRef,
       final_entity: entityJson,
       hash: 'h',
       stitch_ticket: '',
@@ -113,6 +114,7 @@ describe('DefaultEntitiesCatalog', () => {
 
     await knex<DbFinalEntitiesRow>('final_entities').insert({
       entity_id: id,
+      entity_ref: entityRef,
       final_entity: entityJson,
       hash: 'h',
       stitch_ticket: '',

plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts

@@ -251,13 +251,7 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
     });
 
     if (!request?.order) {
-      entitiesQuery = entitiesQuery
-        .leftOuterJoin(
-          'refresh_state',
-          'refresh_state.entity_id',
-          'final_entities.entity_id',
-        )
-        .orderBy('refresh_state.entity_ref', 'asc'); // default sort
+      entitiesQuery = entitiesQuery.orderBy('final_entities.entity_ref', 'asc'); // default sort
     } else {
       entitiesQuery.orderBy('final_entities.entity_id', 'asc'); // stable sort
     }
@@ -326,16 +320,11 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
 
     for (const chunk of lodashChunk(request.entityRefs, 200)) {
       let query = this.database<DbFinalEntitiesRow>('final_entities')
-        .innerJoin<DbRefreshStateRow>(
-          'refresh_state',
-          'refresh_state.entity_id',
-          'final_entities.entity_id',
-        )
         .select({
-          entityRef: 'refresh_state.entity_ref',
+          entityRef: 'final_entities.entity_ref',
           entity: 'final_entities.final_entity',
         })
-        .whereIn('refresh_state.entity_ref', chunk);
+        .whereIn('final_entities.entity_ref', chunk);
 
       if (request?.filter) {
         query = parseFilter(
@@ -343,7 +332,7 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
           query,
           this.database,
           false,
-          'refresh_state.entity_id',
+          'final_entities.entity_id',
         );
       }
 
@@ -669,11 +658,8 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
   }
 
   async entityAncestry(rootRef: string): Promise<EntityAncestryResponse> {
-    const [rootRow] = await this.database<DbRefreshStateRow>('refresh_state')
-      .leftJoin<DbFinalEntitiesRow>('final_entities', {
-        'refresh_state.entity_id': 'final_entities.entity_id',
-      })
-      .where('refresh_state.entity_ref', '=', rootRef)
+    const [rootRow] = await this.database<DbFinalEntitiesRow>('final_entities')
+      .where('final_entities.entity_ref', '=', rootRef)
       .select({
         entityJson: 'final_entities.final_entity',
       });
@@ -698,16 +684,13 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
       const parentRows = await this.database<DbRefreshStateReferencesRow>(
         'refresh_state_references',
       )
-        .innerJoin<DbRefreshStateRow>('refresh_state', {
-          'refresh_state_references.source_entity_ref':
-            'refresh_state.entity_ref',
-        })
         .innerJoin<DbFinalEntitiesRow>('final_entities', {
-          'refresh_state.entity_id': 'final_entities.entity_id',
+          'refresh_state_references.source_entity_ref':
+            'final_entities.entity_ref',
         })
         .where('refresh_state_references.target_entity_ref', '=', currentRef)
         .select({
-          parentEntityRef: 'refresh_state.entity_ref',
+          parentEntityRef: 'final_entities.entity_ref',
           parentEntityJson: 'final_entities.final_entity',
         });
 

plugins/catalog-backend/src/tests/migrations.test.ts

@@ -93,6 +93,7 @@ describe('migrations', () => {
           hash: 'h',
           stitch_ticket: '',
           final_entity: '{}',
+          entity_ref: 'k:ns/n1',
         })
         .into('final_entities');
 
@@ -395,4 +396,110 @@ describe('migrations', () => {
       await knex.destroy();
     },
   );
+
+  it.each(databases.eachSupportedId())(
+    '20241024104700_add_entity_ref_to_final_entities.js, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+
+      await migrateUntilBefore(
+        knex,
+        '20241024104700_add_entity_ref_to_final_entities.js',
+      );
+
+      await knex
+        .insert({
+          entity_id: 'id1',
+          entity_ref: 'k:ns/n',
+          unprocessed_entity: '{}',
+          processed_entity: '{}',
+          errors: '[]',
+          next_update_at: knex.fn.now(),
+          last_discovery_at: knex.fn.now(),
+        })
+        .into('refresh_state');
+      await knex
+        .insert({
+          entity_id: 'id2',
+          entity_ref: 'k:ns/n2',
+          unprocessed_entity: '{}',
+          processed_entity: '{}',
+          errors: '[]',
+          next_update_at: knex.fn.now(),
+          last_discovery_at: knex.fn.now(),
+        })
+        .into('refresh_state');
+
+      // Insert a simple entity before the migration
+      await knex
+        .insert({
+          entity_id: 'id1',
+          hash: '3f5a4d6ba8507be297bb7cd87c4b55b63e3f4c14',
+          stitch_ticket: '52367ed7-120b-405f-b7e0-cdd90f956312',
+          final_entity: '{}',
+        })
+        .into('final_entities');
+
+      // verify that the entity_ref column is not present
+      const preColumnInfo = await knex('final_entities').columnInfo();
+      expect(preColumnInfo.entity_ref).toBeUndefined();
+
+      await migrateUpOnce(knex);
+
+      // verify that the entity_ref column has been added
+      const afterColumnInfo = await knex('final_entities').columnInfo();
+      expect(afterColumnInfo.entity_ref).not.toBeUndefined();
+
+      // verify that the contents of the entity_ref column are correct
+      await expect(knex('final_entities')).resolves.toEqual(
+        expect.arrayContaining([
+          {
+            entity_id: 'id1',
+            hash: '3f5a4d6ba8507be297bb7cd87c4b55b63e3f4c14',
+            stitch_ticket: '52367ed7-120b-405f-b7e0-cdd90f956312',
+            final_entity: '{}',
+            entity_ref: 'k:ns/n',
+            last_updated_at: null,
+          },
+        ]),
+      );
+
+      // Verify that duplicates of entity_ref are not allowed. We lie about the
+      // ref to make it easier to trigger the problem. Also using a weak
+      // expectation due to sqlite flakiness; it rejects, but not necessarily
+      // with a valid error.
+      await expect(
+        knex
+          .insert({
+            entity_id: 'id2',
+            entity_ref: 'k:ns/n',
+            hash: 'other',
+            stitch_ticket: 'other',
+            final_entity: '{}',
+          })
+          .into('final_entities'),
+      ).rejects.toEqual(expect.anything());
+
+      await migrateDownOnce(knex);
+
+      // verify that the entity_ref column has been removed
+      const revertedColumnInfo = await knex('final_entities').columnInfo();
+      expect(revertedColumnInfo.entity_ref).toBeUndefined();
+
+      // verify that the contents are correct
+      await expect(knex('final_entities')).resolves.toEqual(
+        expect.arrayContaining([
+          {
+            entity_id: 'id1',
+            hash: '3f5a4d6ba8507be297bb7cd87c4b55b63e3f4c14',
+            stitch_ticket: '52367ed7-120b-405f-b7e0-cdd90f956312',
+            final_entity: '{}',
+            last_updated_at: null,
+          },
+        ]),
+      );
+
+      await knex.destroy();
+    },
+  );
 });