diff --git a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts index 80c9e0915fa..4a4bc42d24d 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.integration.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.integration.ts @@ -42,6 +42,7 @@ interface QueryPlan { 'Node Type': string 'Relation Name'?: string 'Index Name'?: string + 'Index Cond'?: string 'Shared Hit Blocks': number 'Shared Read Blocks': number Plans?: QueryPlan[] @@ -478,6 +479,50 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => { expect(plan).toMatch(/Index Cond:.*ROW\(/) }, 30_000) + it('probes each walked file for its revision rather than scanning every revision', async () => { + const files = 20 * FILE_SEARCH_BACKFILL_PAGE_SIZE + await connection`INSERT INTO workspace_files (id, workspace_id, context, content_updated_at) + SELECT md5(n::text), 'workspace-' || lpad((n % 7)::text, 2, '0'), 'workspace', '2026-09-16' + FROM generate_series(1, ${files}) n` + await connection.begin(async (tx) => { + await tx`SET LOCAL session_replication_role = replica` + await tx`INSERT INTO workspace_file_search_revision + (file_id, workspace_id, entity_type, entity_id, source_content_updated_at, status) + SELECT id, workspace_id, 'workspace', workspace_id, content_updated_at, 'ready' + FROM workspace_files` + }) + await connection`VACUUM (ANALYZE) workspace_files, workspace_file_search_revision` + + await prepareWorkspaceFileSearchDispatch() + statements.length = 0 + await prepareWorkspaceFileSearchDispatch() + + const walk = statements.find( + (statement) => + statement.query.includes('from "workspace_files"') && + statement.query.includes('order by CASE WHEN') + ) + expect(walk).toBeDefined() + const [explained] = await connection.unsafe( + `EXPLAIN (FORMAT JSON) ${walk?.query}`, + walk?.params as never[] + ) + const nodes = planNodes((explained['QUERY PLAN'] as { Plan: QueryPlan }[])[0].Plan) + + /** + * A page reads a thousand files, but a revision check the planner sizes against every file + * left in the walk hashes the whole revision table on each page instead of probing it once + * per file returned. + */ + expect( + nodes + .filter((node) => node['Relation Name'] === 'workspace_file_search_revision') + .map((node) => ({ index: node['Index Name'], condition: node['Index Cond'] })) + ).toEqual([ + { index: 'workspace_file_search_revision_pkey', condition: '(file_id = workspace_files.id)' }, + ]) + }, 30_000) + it('claims from the ordered pending index rather than a hash join over the backlog', async () => { await seedQueue('workspace-1', 10_000) await connection`ANALYZE workspace_files` diff --git a/apps/sim/lib/workspace-files/search/dispatcher.ts b/apps/sim/lib/workspace-files/search/dispatcher.ts index af7fc66a601..f27e76f8d29 100644 --- a/apps/sim/lib/workspace-files/search/dispatcher.ts +++ b/apps/sim/lib/workspace-files/search/dispatcher.ts @@ -217,7 +217,7 @@ function staleClaim(now: Date): SQL | undefined { * Seeds one page of the backfill that walks every live workspace and project file into the * revision table. * - * Two things keep this page cheap, and losing either one reintroduces a dispatch that times out. + * Three things keep this page cheap, and losing any one reintroduces a dispatch that times out. * * `workspace_files_search_owner_keyset_idx` supplies the `(owner type, owner id, id)` order under * this exact filter, spelled with the same owner expressions. Without it nothing does, so the page @@ -228,6 +228,11 @@ function staleClaim(now: Date): SQL | undefined { * can turn into an index condition; it stays a filter, so each page restarts at the low end of the * index and re-reads every page before it, making the walk quadratic in the file count. * + * Each file's revision is then found by a join on the revision's primary key, which the page's + * limit turns into one index probe per file returned. A correlated `EXISTS` in the select list is + * planned against the rows the scan could return before that limit, so it hashes the whole + * revision table on every page instead. The key is the file id, so the join never repeats a file. + * * The walk itself reads without locks, because nearly every file it passes already has its * revision. Locking the whole page would write a row lock into every heap tuple of it each * pass, and so would inserting rows that only conflict: the revision's owner trigger takes @@ -265,14 +270,13 @@ async function seedBackfillPage(tx: DbTransaction, now: Date): Promise { .select({ ...owner, fileId: workspaceFiles.id, - hasRevision: sql`${exists( - tx - .select({ fileId: workspaceFileSearchRevision.fileId }) - .from(workspaceFileSearchRevision) - .where(eq(workspaceFileSearchRevision.fileId, workspaceFiles.id)) - )}`, + hasRevision: sql`${workspaceFileSearchRevision.fileId} IS NOT NULL`, }) .from(workspaceFiles) + .leftJoin( + workspaceFileSearchRevision, + eq(workspaceFileSearchRevision.fileId, workspaceFiles.id) + ) .where( and( liveSearchableFile,