Skip to content

Commit 0f5590d

Browse files
fix(db): retire unused keyword search projections (#8810)
* fix(db): retire unused keyword search projections * fix(db): retire keyword writers before schema push * fix(db): require approval before retiring push writers
1 parent 325ab96 commit 0f5590d

38 files changed

Lines changed: 30674 additions & 2758 deletions

‎apps/sim/lib/knowledge/__integration__/connector-lease-pages.integration.ts‎

Lines changed: 7 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
120120
RETURN NEW;
121121
END $$`)
122122
)
123-
for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
123+
for (const projection of ['embedding_search']) {
124124
await db.execute(
125125
sql.raw(`DROP TRIGGER IF EXISTS log_lease_page_projection_write ON ${projection}`)
126126
)
@@ -151,7 +151,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
151151
await db.execute(sql`DROP FUNCTION IF EXISTS log_lease_page_acl_write()`)
152152
await db.execute(sql`DROP FUNCTION IF EXISTS fail_after_acl_writes()`)
153153
await db.execute(sql`DROP TABLE IF EXISTS lease_page_acl_writes`)
154-
for (const projection of ['embedding_search', 'embedding_keyword_tin'])
154+
for (const projection of ['embedding_search'])
155155
await db.execute(
156156
sql.raw(`DROP TRIGGER IF EXISTS log_lease_page_projection_write ON ${projection}`)
157157
)
@@ -300,25 +300,14 @@ describe('connector lease ACL pages in PostgreSQL', () => {
300300
)
301301
for (let offset = 0; offset < rows.length; offset += 200)
302302
await db.insert(embedding).values(rows.slice(offset, offset + 200))
303-
/**
304-
* Each chunk's projection rows, filled from its document as the backfill leaves them. The
305-
* embedding insert writes the vector projection's row; the keyword projection's own sync
306-
* trigger ships with the Tin migration, which a database without Tin skips. The fixture sets
307-
* the filled state itself, whatever triggers an earlier suite left behind; what is under
308-
* test is the document trigger that rewrites these rows.
309-
*/
303+
/** Seeds the filled vector state whose ACL updates must stay bounded. */
310304
const chunkIds = sql.join(
311305
documents.map((entry) => sql`${entry.id}`),
312306
sql`, `
313307
)
314308
await db.execute(sql`
315309
UPDATE embedding_search p SET enabled = true, connector_id = d.connector_id, acl = d.acl
316310
FROM document d WHERE d.id = p.document_id AND d.id IN (${chunkIds})`)
317-
await db.execute(sql`
318-
INSERT INTO embedding_keyword_tin (id, knowledge_base_id, document_id, enabled, content, connector_id, acl)
319-
SELECT e.id, e.knowledge_base_id, e.document_id, true, e.content, d.connector_id, d.acl
320-
FROM embedding e JOIN document d ON d.id = e.document_id WHERE e.document_id IN (${chunkIds})
321-
ON CONFLICT (id) DO UPDATE SET enabled = true, connector_id = EXCLUDED.connector_id, acl = EXCLUDED.acl`)
322311
/** Only writes made by the code under test are counted. */
323312
await db.execute(sql`DELETE FROM lease_page_projection_writes`)
324313
await db
@@ -337,10 +326,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
337326
db.execute<{ projection: string; acl: string | null; expected: string }>(sql`
338327
SELECT 'embedding_search' AS projection, array_to_string(p.acl, ',') AS acl,
339328
array_to_string(d.acl, ',') AS expected
340-
FROM embedding_search p JOIN document d ON d.id = p.document_id WHERE d.connector_id = ${connectorId}
341-
UNION ALL
342-
SELECT 'embedding_keyword_tin', array_to_string(p.acl, ','), array_to_string(d.acl, ',')
343-
FROM embedding_keyword_tin p JOIN document d ON d.id = p.document_id WHERE d.connector_id = ${connectorId}`)
329+
FROM embedding_search p JOIN document d ON d.id = p.document_id WHERE d.connector_id = ${connectorId}`)
344330

345331
/** Projection rows each transaction rewrote, per table. */
346332
const projectionRowsPerTransaction = async () =>
@@ -354,7 +340,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
354340
const seeded = await seedDocuments(ids.connectorId, [alice()], 30)
355341
await seedChunks(seeded)
356342
const before = [...(await projectionAcls(ids.connectorId))]
357-
expect(before).toHaveLength(2 * 30 * CHUNKS)
343+
expect(before).toHaveLength(30 * CHUNKS)
358344

359345
await expect(
360346
persistDocumentAcls(
@@ -367,7 +353,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
367353
const after = [...(await projectionAcls(ids.connectorId))]
368354
expect(after.filter((row) => row.acl !== bob() || row.expected !== bob())).toEqual([])
369355
const perTransaction = await projectionRowsPerTransaction()
370-
expect(perTransaction.reduce((total, rows) => total + rows, 0)).toBe(2 * 30 * CHUNKS)
356+
expect(perTransaction.reduce((total, rows) => total + rows, 0)).toBe(30 * CHUNKS)
371357
expect(Math.max(...perTransaction)).toBeLessThanOrEqual(PROJECTION_ROW_BATCH_SIZE)
372358
})
373359

@@ -384,7 +370,7 @@ describe('connector lease ACL pages in PostgreSQL', () => {
384370
).resolves.toBe(true)
385371

386372
const after = [...(await projectionAcls(members.connectorId))]
387-
expect(after).toHaveLength(2 * 30 * CHUNKS)
373+
expect(after).toHaveLength(30 * CHUNKS)
388374
expect(after.filter((row) => row.acl !== '' || row.expected !== '')).toEqual([])
389375
expect(Math.max(...(await projectionRowsPerTransaction()))).toBeLessThanOrEqual(
390376
PROJECTION_ROW_BATCH_SIZE

‎apps/sim/lib/knowledge/__integration__/knowledge-projection.integration.ts‎

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import {
88
import {
99
document,
1010
embedding,
11-
embeddingKeywordSearch,
1211
embeddingKeywordTin,
1312
embeddingSearch,
1413
knowledgeBase,
@@ -123,6 +122,25 @@ afterAll(async () => {
123122
})
124123

125124
describe('ordinary KB vector repair after indexed Search retirement', () => {
125+
it('keeps canonical chunk writes after the retired keyword projection is removed', async () => {
126+
const [relations] =
127+
await projector`SELECT to_regclass('public.embedding_keyword_search') AS keyword`
128+
expect(relations.keyword).toBeNull()
129+
const chunk = chunkRow(generateId(), 0)
130+
await db.insert(embedding).values(chunk)
131+
await db
132+
.update(embedding)
133+
.set({ content: 'updated searchable text' })
134+
.where(eq(embedding.id, chunk.id))
135+
const [canonical] =
136+
await projector`SELECT content_tsv @@ plainto_tsquery('english', 'searchable') AS matches
137+
FROM embedding WHERE id = ${chunk.id}`
138+
expect(canonical.matches).toBe(true)
139+
expect((await vectorsOf()).map((row) => row.id)).toEqual([chunk.id])
140+
await db.delete(embedding).where(eq(embedding.id, chunk.id))
141+
expect(await vectorsOf()).toEqual([])
142+
})
143+
126144
it('repairs deferred vectors in bounded pages without creating copied ACL or keyword rows', async () => {
127145
const chunks = Array.from({ length: 5 }, (_, index) => chunkRow(generateId(), index))
128146
await defer((tx) => tx.insert(embedding).values(chunks))
@@ -143,12 +161,6 @@ describe('ordinary KB vector repair after indexed Search retirement', () => {
143161
(row) => row.vector512?.[511] === 1 && row.acl === null && row.connectorId === null
144162
)
145163
).toBe(true)
146-
expect(
147-
await db
148-
.select()
149-
.from(embeddingKeywordSearch)
150-
.where(eq(embeddingKeywordSearch.documentId, documentId))
151-
).toEqual([])
152164
expect(
153165
await db
154166
.select()
@@ -167,12 +179,6 @@ describe('ordinary KB vector repair after indexed Search retirement', () => {
167179
expect(await markOf(searchDocumentId)).toMatchObject({ content: true })
168180
await runKnowledgeProjection(projector, {})
169181
expect(await vectorsOf(searchDocumentId)).toEqual([])
170-
expect(
171-
await db
172-
.select()
173-
.from(embeddingKeywordSearch)
174-
.where(eq(embeddingKeywordSearch.documentId, searchDocumentId))
175-
).toEqual([])
176182
expect(
177183
await db
178184
.select()

‎apps/sim/lib/knowledge/__integration__/search-latency.integration.ts‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -685,7 +685,6 @@ describe.skipIf(!enabled)('Knowledge search latency on a realistic indexed corpu
685685
await db.execute(sql`ANALYZE document`)
686686
await db.execute(sql`ANALYZE embedding`)
687687
await db.execute(sql`ANALYZE embedding_search`)
688-
await db.execute(sql`ANALYZE embedding_keyword_search`)
689688
if (evictSharedBuffers) await db.execute(sql`CREATE EXTENSION IF NOT EXISTS pg_buffercache`)
690689
report.server = (
691690
await db.execute(sql`SELECT version(), current_setting('work_mem') AS work_mem,

‎apps/sim/lib/knowledge/__integration__/workspace-kb-document-access.integration.ts‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ import {
1717
credentialGroupEnrollment,
1818
document,
1919
embedding,
20-
embeddingKeywordSearch,
2120
embeddingSearch,
2221
knowledgeBase,
2322
knowledgeConnector,
@@ -135,9 +134,6 @@ describe.each([
135134
.update(embeddingSearch)
136135
.set({ acl: ['s:nobody:-:stale'], connectorId: null })
137136
.where(eq(embeddingSearch.knowledgeBaseId, baseId))
138-
await db
139-
.delete(embeddingKeywordSearch)
140-
.where(eq(embeddingKeywordSearch.knowledgeBaseId, baseId))
141137
})
142138

143139
afterAll(async () => {

‎apps/sim/lib/knowledge/connectors/detachment.test.ts‎

Lines changed: 6 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,4 @@
1-
import {
2-
document,
3-
embeddingKeywordTin,
4-
embeddingSearch,
5-
knowledgeBase,
6-
knowledgeConnector,
7-
} from '@sim/db/schema'
1+
import { document, embeddingSearch, knowledgeBase, knowledgeConnector } from '@sim/db/schema'
82
import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
93
import { billingStorageMock, billingStorageMockFns } from '@sim/testing/mocks/billing-storage.mock'
104
import {
@@ -68,12 +62,10 @@ function queueBatch(documentIds: string[], reservedBytes = 0) {
6862
}
6963

7064
/** Row ids the projection release returns, in the order the handler updates the tables. */
71-
function releaseProjectionRows(searchRows: number, keywordRows: number) {
65+
function releaseProjectionRows(searchRows: number) {
7266
const rows = (count: number) =>
7367
Array.from({ length: count }, (_, index) => ({ id: `c-${index}` }))
74-
dbChainMockFns.returning
75-
.mockResolvedValueOnce(rows(searchRows))
76-
.mockResolvedValueOnce(rows(keywordRows))
68+
dbChainMockFns.returning.mockResolvedValueOnce(rows(searchRows))
7769
}
7870

7971
function updatedTables() {
@@ -92,7 +84,7 @@ describe('connector detachment', () => {
9284

9385
it('releases documents against the reservation, then settles what remains with the connector', async () => {
9486
queueBatch(['doc-1', 'doc-2'], 25)
95-
releaseProjectionRows(3, 3)
87+
releaseProjectionRows(3)
9688
dbChainMockFns.returning.mockResolvedValueOnce([
9789
{ fileSize: 10, deletedAt: null },
9890
{ fileSize: 5, deletedAt: new Date('2026-09-01T00:00:00.000Z') },
@@ -108,12 +100,7 @@ describe('connector detachment', () => {
108100
context()
109101
)
110102

111-
expect(updatedTables()).toEqual([
112-
embeddingSearch,
113-
embeddingKeywordTin,
114-
document,
115-
knowledgeConnector,
116-
])
103+
expect(updatedTables()).toEqual([embeddingSearch, document, knowledgeConnector])
117104
expect(dbChainMockFns.set).toHaveBeenCalledWith({ connectorId: null })
118105
expect(dbChainMockFns.set).toHaveBeenCalledWith(
119106
expect.objectContaining({ connectorId: null, deletedAt: expect.anything() })
@@ -131,7 +118,7 @@ describe('connector detachment', () => {
131118
it('keeps documents attached until their search rows fit one page, without spending retries', async () => {
132119
for (let batch = 0; batch < 4; batch++) {
133120
queueBatch(['doc-1'])
134-
releaseProjectionRows(250, 0)
121+
releaseProjectionRows(250)
135122
}
136123

137124
expect(await detachKnowledgeConnector(payload, context())).toMatchObject({

‎apps/sim/lib/knowledge/connectors/detachment.ts‎

Lines changed: 19 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,6 @@
11
import { db } from '@sim/db'
22
import { PROJECTION_ROW_BATCH_SIZE } from '@sim/db/knowledge-projection'
3-
import {
4-
document,
5-
embeddingKeywordTin,
6-
embeddingSearch,
7-
knowledgeBase,
8-
knowledgeConnector,
9-
} from '@sim/db/schema'
3+
import { document, embeddingSearch, knowledgeBase, knowledgeConnector } from '@sim/db/schema'
104
import { and, asc, eq, inArray, isNotNull, lt, ne, sql } from 'drizzle-orm'
115
import { z } from 'zod'
126
import {
@@ -78,8 +72,6 @@ export function keptDocumentBytes() {
7872
END`
7973
}
8074

81-
const SEARCH_PROJECTIONS = [embeddingSearch, embeddingKeywordTin] as const
82-
8375
/**
8476
* Settles what is left of a detached connector's reservation: an unreleased remainder is refunded,
8577
* and an overdraft, released bytes beyond what removal charged, is charged as already admitted.
@@ -196,8 +188,8 @@ export async function settleDetachedConnectorReservations(
196188
* Releases a detached connector's documents as standalone entries, then deletes the connector.
197189
*
198190
* Nulling a document's `connector_id` fires the projection trigger, which rewrites every enabled
199-
* chunk of it in both search projections, each a fresh index entry. So each transaction first
200-
* releases at most 250 projection rows per table of the next 100 documents, and flips those
191+
* chunk of it in the vector projection, each a fresh index entry. So each transaction first
192+
* releases at most 250 vector projection rows of the next 100 documents, and flips those
201193
* documents only once none of their rows still name the connector; the trigger then finds nothing
202194
* to rewrite. A document larger than one page spans several transactions, and its released rows
203195
* read as an upload in the meantime, which grants the same access: only workspace-access
@@ -299,27 +291,23 @@ export const detachKnowledgeConnector: OutboxHandler = async (rawPayload, contex
299291
return drained
300292
}
301293

302-
let projectionPageFull = false
303-
for (const projection of SEARCH_PROJECTIONS) {
304-
const page = tx
305-
.select({ id: projection.id })
306-
.from(projection)
307-
.where(
308-
and(
309-
inArray(projection.documentId, documentIds),
310-
eq(projection.enabled, true),
311-
isNotNull(projection.connectorId)
312-
)
294+
const page = tx
295+
.select({ id: embeddingSearch.id })
296+
.from(embeddingSearch)
297+
.where(
298+
and(
299+
inArray(embeddingSearch.documentId, documentIds),
300+
eq(embeddingSearch.enabled, true),
301+
isNotNull(embeddingSearch.connectorId)
313302
)
314-
.limit(PROJECTION_ROW_BATCH_SIZE)
315-
const released = await tx
316-
.update(projection)
317-
.set({ connectorId: null })
318-
.where(inArray(projection.id, page))
319-
.returning({ id: projection.id })
320-
if (released.length === PROJECTION_ROW_BATCH_SIZE) projectionPageFull = true
321-
}
322-
if (projectionPageFull) return 'progress'
303+
)
304+
.limit(PROJECTION_ROW_BATCH_SIZE)
305+
const released = await tx
306+
.update(embeddingSearch)
307+
.set({ connectorId: null })
308+
.where(inArray(embeddingSearch.id, page))
309+
.returning({ id: embeddingSearch.id })
310+
if (released.length === PROJECTION_ROW_BATCH_SIZE) return 'progress'
323311

324312
const releasedDocuments = await tx
325313
.update(document)

‎apps/sim/lib/knowledge/connectors/sync-content-pass.integration.ts‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -153,7 +153,7 @@ describe('completed listing reconciliation in PostgreSQL', () => {
153153
acl text[] NOT NULL DEFAULT '{ws}', acl_requirements jsonb NOT NULL DEFAULT '[]',
154154
acl_verified_at timestamp
155155
)`
156-
for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
156+
for (const projection of ['embedding_search']) {
157157
await sql`CREATE TABLE ${sql(projection)} (
158158
id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true,
159159
connector_id text, acl text[]
@@ -170,7 +170,6 @@ describe('completed listing reconciliation in PostgreSQL', () => {
170170
await sql`CREATE TRIGGER count_fan_out AFTER UPDATE OF connector_id, acl ON document
171171
FOR EACH ROW EXECUTE FUNCTION count_fan_out()`
172172
await sql`ALTER TABLE embedding_search DISABLE TRIGGER embedding_search_source_acl_set`
173-
await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set`
174173
}, 60_000)
175174

176175
afterAll(async () => {
@@ -181,7 +180,7 @@ describe('completed listing reconciliation in PostgreSQL', () => {
181180

182181
beforeEach(async () => {
183182
hardDelete.mockClear()
184-
await sql`TRUNCATE embedding_search, embedding_keyword_tin, document, fan_out, knowledge_connector`
183+
await sql`TRUNCATE embedding_search, document, fan_out, knowledge_connector`
185184
await sql`INSERT INTO knowledge_connector (id) VALUES (${CONNECTOR})`
186185
})
187186

@@ -191,7 +190,7 @@ describe('completed listing reconciliation in PostgreSQL', () => {
191190
{ id: 'absent-empty', acl: [], seenAt: ABSENT_SINCE },
192191
{ id: 'absent-granted', acl: [ALICE], seenAt: ABSENT_SINCE, verified: true },
193192
])
194-
for (const projection of ['embedding_search', 'embedding_keyword_tin']) {
193+
for (const projection of ['embedding_search']) {
195194
const prefix = projection === 'embedding_search' ? 'vec' : 'kw'
196195
await sql`INSERT INTO ${sql(projection)} (id, document_id, connector_id, acl) VALUES
197196
(${`${prefix}-empty-filled`}, 'absent-empty', ${CONNECTOR}, '{}'),
@@ -210,13 +209,7 @@ describe('completed listing reconciliation in PostgreSQL', () => {
210209
{ id: 'absent-empty', acl: [], verified: false, deleted: true },
211210
{ id: 'absent-granted', acl: [], verified: false, deleted: true },
212211
])
213-
expect(
214-
await sql`SELECT id, acl FROM embedding_search
215-
UNION ALL SELECT id, acl FROM embedding_keyword_tin ORDER BY id`
216-
).toEqual([
217-
{ id: 'kw-empty-filled', acl: [] },
218-
{ id: 'kw-granted-filled', acl: [] },
219-
{ id: 'kw-granted-unfilled', acl: null },
212+
expect(await sql`SELECT id, acl FROM embedding_search ORDER BY id`).toEqual([
220213
{ id: 'vec-empty-filled', acl: [] },
221214
{ id: 'vec-granted-filled', acl: [] },
222215
{ id: 'vec-granted-unfilled', acl: null },

0 commit comments

Comments
 (0)