perf(file-search): probe each backfill page's revisions by primary key

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`

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<number> {
     .select({
       ...owner,
       fileId: workspaceFiles.id,
-      hasRevision: sql<boolean>`${exists(
-        tx
-          .select({ fileId: workspaceFileSearchRevision.fileId })
-          .from(workspaceFileSearchRevision)
-          .where(eq(workspaceFileSearchRevision.fileId, workspaceFiles.id))
-      )}`,
+      hasRevision: sql<boolean>`${workspaceFileSearchRevision.fileId} IS NOT NULL`,
     })
     .from(workspaceFiles)
+    .leftJoin(
+      workspaceFileSearchRevision,
+      eq(workspaceFileSearchRevision.fileId, workspaceFiles.id)
+    )
     .where(
       and(
         liveSearchableFile,