draft(catalog-backend): search table dedup, covering indices, UNIQUE constraint
Single knex migration that cleans up duplicate search rows and creates covering indices including a UNIQUE constraint on (entity_id, key, value). Each step is idempotent and handles INVALID indices from interrupted runs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: Fredrik Adelöw <freben@spotify.com>
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,282 @@
|
||||
/*
|
||||
* Copyright 2026 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
|
||||
|
||||
/**
|
||||
* Deduplicates the search table, adds covering indices (including a UNIQUE
|
||||
* constraint on (entity_id, key, value)), and drops superseded indices.
|
||||
*
|
||||
* On PostgreSQL this uses CREATE INDEX CONCURRENTLY which avoids blocking
|
||||
* reads/writes but can take several minutes on large tables (13M+ rows).
|
||||
* The migration is fully rerun-safe: each step checks the current state
|
||||
* and skips work that is already done or cleans up INVALID indices left by
|
||||
* a previous interrupted attempt.
|
||||
*
|
||||
* ## Important note for large installations
|
||||
*
|
||||
* If your search table has millions of rows, this migration may take
|
||||
* 5-15 minutes. During this time the pod will appear unready. Other pods
|
||||
* running the previous version of Backstage will continue to serve
|
||||
* traffic normally — the index creation does not block reads or writes.
|
||||
*
|
||||
* To avoid the startup delay, you can run the following commands against
|
||||
* your PostgreSQL database BEFORE deploying this version:
|
||||
*
|
||||
* -- 1. Remove duplicate search rows
|
||||
* WITH cte AS (
|
||||
* SELECT ctid,
|
||||
* row_number() OVER (PARTITION BY entity_id, key, value) AS rn
|
||||
* FROM search
|
||||
* )
|
||||
* DELETE FROM search USING cte
|
||||
* WHERE search.ctid = cte.ctid AND cte.rn > 1;
|
||||
*
|
||||
* -- 2. Create indices (run each separately)
|
||||
* CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS
|
||||
* search_entity_key_value_idx ON search (entity_id, key, value);
|
||||
* CREATE INDEX CONCURRENTLY IF NOT EXISTS
|
||||
* search_key_value_entity_idx ON search (key, value, entity_id);
|
||||
* CREATE INDEX CONCURRENTLY IF NOT EXISTS
|
||||
* search_facets_covering_idx ON search (key, original_value, entity_id)
|
||||
* WHERE original_value IS NOT NULL;
|
||||
*
|
||||
* -- 3. Drop old indices
|
||||
* DROP INDEX CONCURRENTLY IF EXISTS search_key_value_idx;
|
||||
* DROP INDEX CONCURRENTLY IF EXISTS search_key_original_value_idx;
|
||||
*
|
||||
* If these commands have already completed, the migration will detect the
|
||||
* existing indices and skip all work — startup will be instant.
|
||||
*/
|
||||
|
||||
/**
|
||||
* @param {import('knex').Knex} knex
|
||||
*/
|
||||
exports.up = async function up(knex) {
|
||||
const client = knex.client.config.client;
|
||||
|
||||
if (client.includes('pg')) {
|
||||
await upPostgres(knex);
|
||||
} else if (client.includes('mysql')) {
|
||||
await upMysql(knex);
|
||||
} else {
|
||||
await upSqlite(knex);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* @param {import('knex').Knex} knex
|
||||
*/
|
||||
exports.down = async function down(_knex) {
|
||||
// Intentionally a no-op. The old non-covering indices are not worth
|
||||
// recreating, and the UNIQUE constraint should not be removed since the
|
||||
// code now relies on ON CONFLICT for correctness.
|
||||
};
|
||||
|
||||
exports.config = { transaction: false };
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// PostgreSQL
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** @param {import('knex').Knex} knex */
|
||||
async function upPostgres(knex) {
|
||||
// Step 1: Remove duplicate search rows. The window function's
|
||||
// PARTITION BY treats NULLs as equal, so this handles both NULL-value
|
||||
// and non-NULL-value duplicates. Idempotent: deletes 0 rows if clean.
|
||||
await knex.raw(`
|
||||
WITH cte AS (
|
||||
SELECT ctid,
|
||||
row_number() OVER (PARTITION BY entity_id, key, value) AS rn
|
||||
FROM search
|
||||
)
|
||||
DELETE FROM search
|
||||
USING cte
|
||||
WHERE search.ctid = cte.ctid AND cte.rn > 1
|
||||
`);
|
||||
|
||||
// Step 2: Create covering indices. Each call is idempotent — it checks
|
||||
// the index state and only does work if needed.
|
||||
await ensurePgIndex(knex, {
|
||||
name: 'search_entity_key_value_idx',
|
||||
columns: '(entity_id, key, value)',
|
||||
unique: true,
|
||||
});
|
||||
await ensurePgIndex(knex, {
|
||||
name: 'search_key_value_entity_idx',
|
||||
columns: '(key, value, entity_id)',
|
||||
unique: false,
|
||||
});
|
||||
await ensurePgIndex(knex, {
|
||||
name: 'search_facets_covering_idx',
|
||||
columns: '(key, original_value, entity_id)',
|
||||
where: 'WHERE original_value IS NOT NULL',
|
||||
unique: false,
|
||||
});
|
||||
|
||||
// Step 3: Drop superseded indices.
|
||||
await dropPgIndexIfExists(knex, 'search_key_value_idx');
|
||||
await dropPgIndexIfExists(knex, 'search_key_original_value_idx');
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates or replaces an index on the search table, handling all edge cases:
|
||||
* - Already valid with correct uniqueness: skip
|
||||
* - Exists but INVALID (interrupted CREATE): drop and recreate
|
||||
* - Exists but wrong uniqueness (e.g. non-unique but we need unique): drop and recreate
|
||||
* - Missing: create from scratch
|
||||
*
|
||||
* @param {import('knex').Knex} knex
|
||||
* @param {{ name: string; columns: string; unique: boolean; where?: string }} opts
|
||||
*/
|
||||
async function ensurePgIndex(knex, opts) {
|
||||
const { name, columns, unique, where } = opts;
|
||||
|
||||
const result = await knex.raw(
|
||||
`
|
||||
SELECT indisunique, indisvalid
|
||||
FROM pg_index
|
||||
WHERE indexrelid = (
|
||||
SELECT oid FROM pg_class WHERE relname = ?
|
||||
)
|
||||
`,
|
||||
[name],
|
||||
);
|
||||
|
||||
if (result.rows.length > 0) {
|
||||
const { indisunique, indisvalid } = result.rows[0];
|
||||
if (indisvalid && indisunique === unique) {
|
||||
return; // Already correct
|
||||
}
|
||||
// Wrong state — drop and recreate
|
||||
await knex.raw(`DROP INDEX CONCURRENTLY IF EXISTS "${name}"`);
|
||||
}
|
||||
|
||||
const uniqueKw = unique ? 'UNIQUE' : '';
|
||||
const whereClause = where || '';
|
||||
await knex.raw(
|
||||
`CREATE ${uniqueKw} INDEX CONCURRENTLY "${name}" ON search ${columns} ${whereClause}`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {import('knex').Knex} knex
|
||||
* @param {string} name
|
||||
*/
|
||||
async function dropPgIndexIfExists(knex, name) {
|
||||
const result = await knex.raw(
|
||||
`
|
||||
SELECT 1 FROM pg_class WHERE relname = ?
|
||||
`,
|
||||
[name],
|
||||
);
|
||||
if (result.rows.length > 0) {
|
||||
await knex.raw(`DROP INDEX CONCURRENTLY IF EXISTS "${name}"`);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// MySQL
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** @param {import('knex').Knex} knex */
|
||||
async function upMysql(knex) {
|
||||
// Dedup via temp table
|
||||
await knex.transaction(async trx => {
|
||||
await trx.raw(
|
||||
'CREATE TEMPORARY TABLE IF NOT EXISTS `_search_keep` (' +
|
||||
'`entity_id` VARCHAR(255), `key` VARCHAR(255), ' +
|
||||
'`value` VARCHAR(255), `original_value` VARCHAR(255))',
|
||||
);
|
||||
await trx.raw('DELETE FROM `_search_keep`');
|
||||
await trx.raw(
|
||||
'INSERT INTO `_search_keep` ' +
|
||||
'SELECT `entity_id`, `key`, `value`, MAX(`original_value`) ' +
|
||||
'FROM `search` GROUP BY `entity_id`, `key`, `value`',
|
||||
);
|
||||
await trx.raw('DELETE FROM `search`');
|
||||
await trx.raw(
|
||||
'INSERT INTO `search` (`entity_id`, `key`, `value`, `original_value`) ' +
|
||||
'SELECT * FROM `_search_keep`',
|
||||
);
|
||||
await trx.raw('DROP TEMPORARY TABLE `_search_keep`');
|
||||
});
|
||||
|
||||
// Drop old indices if present, then create new ones
|
||||
await mysqlDropIndexIfExists(knex, 'search_key_value_idx');
|
||||
await mysqlDropIndexIfExists(knex, 'search_key_original_value_idx');
|
||||
await mysqlDropIndexIfExists(knex, 'search_entity_key_value_idx');
|
||||
await mysqlDropIndexIfExists(knex, 'search_key_value_entity_idx');
|
||||
await mysqlDropIndexIfExists(knex, 'search_facets_covering_idx');
|
||||
|
||||
await knex.schema.alterTable('search', table => {
|
||||
table.unique(['entity_id', 'key', 'value'], 'search_entity_key_value_idx');
|
||||
table.index(['key', 'value', 'entity_id'], 'search_key_value_entity_idx');
|
||||
});
|
||||
// MySQL doesn't support partial indices — create a full one
|
||||
await knex.schema.alterTable('search', table => {
|
||||
table.index(
|
||||
['key', 'original_value', 'entity_id'],
|
||||
'search_facets_covering_idx',
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** @param {import('knex').Knex} knex @param {string} name */
|
||||
async function mysqlDropIndexIfExists(knex, name) {
|
||||
const [rows] = await knex.raw(
|
||||
`SHOW INDEX FROM \`search\` WHERE Key_name = ?`,
|
||||
[name],
|
||||
);
|
||||
if (rows.length > 0) {
|
||||
await knex.schema.alterTable('search', t => {
|
||||
t.dropIndex([], name);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SQLite
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** @param {import('knex').Knex} knex */
|
||||
async function upSqlite(knex) {
|
||||
await knex.transaction(async trx => {
|
||||
await trx.raw(`
|
||||
DELETE FROM search
|
||||
WHERE rowid NOT IN (
|
||||
SELECT MIN(rowid) FROM search GROUP BY entity_id, key, value
|
||||
)
|
||||
`);
|
||||
});
|
||||
|
||||
// Drop old, create new — SQLite is fast on small tables
|
||||
await knex.raw('DROP INDEX IF EXISTS search_key_value_idx');
|
||||
await knex.raw('DROP INDEX IF EXISTS search_key_original_value_idx');
|
||||
await knex.raw('DROP INDEX IF EXISTS search_entity_key_value_idx');
|
||||
await knex.raw('DROP INDEX IF EXISTS search_key_value_entity_idx');
|
||||
await knex.raw('DROP INDEX IF EXISTS search_facets_covering_idx');
|
||||
|
||||
await knex.raw(
|
||||
'CREATE UNIQUE INDEX search_entity_key_value_idx ON search (entity_id, key, value)',
|
||||
);
|
||||
await knex.raw(
|
||||
'CREATE INDEX search_key_value_entity_idx ON search (key, value, entity_id)',
|
||||
);
|
||||
await knex.raw(
|
||||
'CREATE INDEX search_facets_covering_idx ON search (key, original_value, entity_id)',
|
||||
);
|
||||
}
|
||||
@@ -228,5 +228,16 @@ export function buildEntitySearch(
|
||||
);
|
||||
}
|
||||
|
||||
return mapToRows(raw, entityId);
|
||||
const rows = mapToRows(raw, entityId);
|
||||
|
||||
// Deduplicate by (key, value). Duplicate array values in the entity data
|
||||
// (e.g. tags: ['java', 'java']) produce identical search rows which would
|
||||
// violate the unique constraint on (entity_id, key, value).
|
||||
const seen = new Set<string>();
|
||||
return rows.filter(row => {
|
||||
const k = `${row.key}\0${row.value ?? ''}`;
|
||||
if (seen.has(k)) return false;
|
||||
seen.add(k);
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -190,20 +190,17 @@ describe.each(databases.eachSupportedId())('syncSearchRows, %p', databaseId => {
|
||||
);
|
||||
});
|
||||
|
||||
it('inserts a row when only original_value casing differs from existing', async () => {
|
||||
// Two rows with the same (key, value) but different original_value
|
||||
// casing must coexist — the case-insensitive MySQL collation must not
|
||||
// cause the INSERT to skip the second row.
|
||||
it('keeps one row when original_value casing differs for same key+value', async () => {
|
||||
// Two entries with the same lowercased (key, value) but different
|
||||
// original_value casing are deduplicated — the UNIQUE constraint on
|
||||
// (entity_id, key, value) allows only one. The last occurrence wins.
|
||||
await syncSearchRows(knex, 'e1', [row('a', 'v', 'V')]);
|
||||
await syncSearchRows(knex, 'e1', [row('a', 'v', 'V'), row('a', 'v', 'v')]);
|
||||
|
||||
const rows = await getSearchRows();
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(rows).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ key: 'a', value: 'v', original_value: 'V' }),
|
||||
expect.objectContaining({ key: 'a', value: 'v', original_value: 'v' }),
|
||||
]),
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]).toEqual(
|
||||
expect.objectContaining({ key: 'a', value: 'v', original_value: 'v' }),
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -48,14 +48,24 @@ export async function syncSearchRows(
|
||||
entityId: string,
|
||||
searchEntries: DbSearchRow[],
|
||||
): Promise<void> {
|
||||
// Dedup by (key, value) — the UNIQUE constraint on (entity_id, key, value)
|
||||
// rejects duplicates, and the same lowercased value with different original
|
||||
// casing is semantically a single entry. Keep the last occurrence so that
|
||||
// if the entity data has e.g. tags: ['Java', 'java'], the later one wins.
|
||||
const dedupMap = new Map<string, DbSearchRow>();
|
||||
for (const entry of searchEntries) {
|
||||
dedupMap.set(`${entry.key}\0${entry.value ?? ''}`, entry);
|
||||
}
|
||||
const deduped = [...dedupMap.values()];
|
||||
|
||||
const client = knex.client.config.client;
|
||||
|
||||
if (client === 'pg') {
|
||||
await syncPostgres(knex, entityId, searchEntries);
|
||||
await syncPostgres(knex, entityId, deduped);
|
||||
} else if (client.includes('mysql')) {
|
||||
await syncMysql(knex, entityId, searchEntries);
|
||||
await syncMysql(knex, entityId, deduped);
|
||||
} else {
|
||||
await syncBulkReplace(knex, entityId, searchEntries);
|
||||
await syncBulkReplace(knex, entityId, deduped);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -110,6 +120,8 @@ async function syncPostgres(
|
||||
AND COALESCE(s.value, chr(1)) = COALESCE(d.value, chr(1))
|
||||
AND COALESCE(s.original_value, chr(1)) = COALESCE(d.original_value, chr(1))
|
||||
)
|
||||
ON CONFLICT (entity_id, key, value)
|
||||
DO UPDATE SET original_value = EXCLUDED.original_value
|
||||
`,
|
||||
[keys, values, originalValues, entityId, entityId, entityId],
|
||||
);
|
||||
|
||||
@@ -2013,22 +2013,26 @@ describe('DefaultEntitiesCatalog', () => {
|
||||
|
||||
await Promise.all(entities.map(e => addEntityToSearch(e)));
|
||||
|
||||
// Manually insert duplicate search entries for the same entities
|
||||
// I'm not sure exactly how this happens but I have seen it in the real world
|
||||
await knex<DbSearchRow>('search').insert([
|
||||
{
|
||||
entity_id: 'uid-a',
|
||||
key: 'metadata.title',
|
||||
value: 'a test entity',
|
||||
original_value: 'A Test Entity',
|
||||
},
|
||||
{
|
||||
entity_id: 'uid-b',
|
||||
key: 'metadata.title',
|
||||
value: 'b test entity',
|
||||
original_value: 'B Test Entity',
|
||||
},
|
||||
]);
|
||||
// The UNIQUE constraint on (entity_id, key, value) prevents
|
||||
// duplicate search rows. Verify that duplicates are silently
|
||||
// rejected and the query still returns correct results.
|
||||
await knex<DbSearchRow>('search')
|
||||
.insert([
|
||||
{
|
||||
entity_id: 'uid-a',
|
||||
key: 'metadata.title',
|
||||
value: 'a test entity',
|
||||
original_value: 'A Test Entity',
|
||||
},
|
||||
{
|
||||
entity_id: 'uid-b',
|
||||
key: 'metadata.title',
|
||||
value: 'b test entity',
|
||||
original_value: 'B Test Entity',
|
||||
},
|
||||
])
|
||||
.onConflict(['entity_id', 'key', 'value'])
|
||||
.ignore();
|
||||
|
||||
const catalog = new DefaultEntitiesCatalog({
|
||||
database: knex,
|
||||
@@ -2407,15 +2411,19 @@ describe('DefaultEntitiesCatalog', () => {
|
||||
spec: {},
|
||||
});
|
||||
|
||||
// Manually insert a duplicate search entry, this shouldn't happen but does in reality
|
||||
await knex<DbSearchRow>('search').insert([
|
||||
{
|
||||
entity_id: 'uid-a',
|
||||
key: 'metadata.name',
|
||||
value: 'one',
|
||||
original_value: 'one',
|
||||
},
|
||||
]);
|
||||
// Attempt to insert a duplicate — the UNIQUE constraint silently
|
||||
// rejects it via ON CONFLICT IGNORE.
|
||||
await knex<DbSearchRow>('search')
|
||||
.insert([
|
||||
{
|
||||
entity_id: 'uid-a',
|
||||
key: 'metadata.name',
|
||||
value: 'one',
|
||||
original_value: 'one',
|
||||
},
|
||||
])
|
||||
.onConflict(['entity_id', 'key', 'value'])
|
||||
.ignore();
|
||||
|
||||
const catalog = new DefaultEntitiesCatalog({
|
||||
database: knex,
|
||||
|
||||
Reference in New Issue
Block a user