improvement(tables): filter-aware select-all runs, delete-job read mask, keyset index + autovacuum tuning

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/columns/run/route.ts

@@ -6,7 +6,7 @@ import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { runWorkflowColumn } from '@/lib/table/workflow-columns'
-import { accessError, checkAccess } from '@/app/api/table/utils'
+import { accessError, checkAccess, tableFilterError } from '@/app/api/table/utils'
 
 const logger = createLogger('TableRunColumnAPI')
 
@@ -25,16 +25,21 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
     const parsed = await parseRequest(runColumnContract, request, { params })
     if (!parsed.success) return parsed.response
     const { tableId } = parsed.data.params
-    const { workspaceId, groupIds, runMode, rowIds, limit } = parsed.data.body
+    const { workspaceId, groupIds, runMode, rowIds, filter, limit } = parsed.data.body
     const access = await checkAccess(tableId, auth.userId, 'write')
     if (!access.ok) return accessError(access, requestId, tableId)
 
+    // Validate the filter up front (the dispatcher reuses it) so a bad field fails fast.
+    const filterError = tableFilterError(filter, access.table.schema.columns)
+    if (filterError) return filterError
+
     const { dispatchId } = await runWorkflowColumn({
       tableId,
       workspaceId,
       groupIds,
       mode: runMode,
       rowIds,
+      filter,
       limit,
       requestId,
     })

apps/sim/app/api/table/[tableId]/delete-async/route.test.ts

@@ -2,34 +2,24 @@
  * @vitest-environment node
  */
 import { hybridAuthMockFns } from '@sim/testing'
-import { NextRequest } from 'next/server'
+import { NextRequest, NextResponse } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { TableDefinition } from '@/lib/table'
 
-const {
-  mockCheckAccess,
-  mockMarkTableJobRunning,
-  mockRunTableDelete,
-  mockBuildFilterClause,
-  TableQueryValidationError,
-} = vi.hoisted(() => ({
-  mockCheckAccess: vi.fn(),
-  mockMarkTableJobRunning: vi.fn(),
-  mockRunTableDelete: vi.fn(),
-  mockBuildFilterClause: vi.fn(),
-  TableQueryValidationError: class extends Error {},
-}))
+const { mockCheckAccess, mockMarkTableJobRunning, mockRunTableDelete, mockTableFilterError } =
+  vi.hoisted(() => ({
+    mockCheckAccess: vi.fn(),
+    mockMarkTableJobRunning: vi.fn(),
+    mockRunTableDelete: vi.fn(),
+    mockTableFilterError: vi.fn(),
+  }))
 
 vi.mock('@sim/utils/id', () => ({
   generateId: vi.fn().mockReturnValue('job-id-xyz'),
   generateShortId: vi.fn().mockReturnValue('short-id'),
 }))
 vi.mock('@/lib/table/service', () => ({ markTableJobRunning: mockMarkTableJobRunning }))
 vi.mock('@/lib/table/delete-runner', () => ({ runTableDelete: mockRunTableDelete }))
-vi.mock('@/lib/table/sql', () => ({
-  buildFilterClause: mockBuildFilterClause,
-  TableQueryValidationError,
-}))
 vi.mock('@/lib/core/utils/background', () => ({
   runDetached: (_label: string, work: () => Promise<unknown>) => {
     void work()
@@ -41,6 +31,7 @@ vi.mock('@/app/api/table/utils', async () => {
     checkAccess: mockCheckAccess,
     accessError: (result: { status: number }) =>
       NextResponse.json({ error: 'denied' }, { status: result.status }),
+    tableFilterError: mockTableFilterError,
   }
 })
 
@@ -90,7 +81,7 @@ describe('POST /api/table/[tableId]/delete-async', () => {
     mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
     mockMarkTableJobRunning.mockResolvedValue(true)
     mockRunTableDelete.mockResolvedValue(undefined)
-    mockBuildFilterClause.mockReturnValue({})
+    mockTableFilterError.mockReturnValue(null)
   })
 
   it('claims the job slot and kicks off the delete worker with filter + exclusions', async () => {
@@ -99,7 +90,11 @@ describe('POST /api/table/[tableId]/delete-async', () => {
 
     expect(response.status).toBe(200)
     expect(data.data).toEqual({ tableId: 'tbl_1', jobId: 'job-id-xyz' })
-    expect(mockMarkTableJobRunning).toHaveBeenCalledWith('tbl_1', 'job-id-xyz', 'delete')
+    expect(mockMarkTableJobRunning).toHaveBeenCalledWith('tbl_1', 'job-id-xyz', 'delete', {
+      filter: { status: 'archived' },
+      excludeRowIds: ['row_keep'],
+      cutoff: expect.any(String),
+    })
     expect(mockRunTableDelete).toHaveBeenCalledWith(
       expect.objectContaining({
         jobId: 'job-id-xyz',
@@ -128,9 +123,7 @@ describe('POST /api/table/[tableId]/delete-async', () => {
   })
 
   it('returns 400 on an invalid filter without claiming the slot', async () => {
-    mockBuildFilterClause.mockImplementation(() => {
-      throw new TableQueryValidationError('bad field')
-    })
+    mockTableFilterError.mockReturnValue(NextResponse.json({ error: 'bad field' }, { status: 400 }))
     const response = await makeRequest(validBody)
     expect(response.status).toBe(400)
     expect(mockMarkTableJobRunning).not.toHaveBeenCalled()

apps/sim/app/api/table/[tableId]/delete-async/route.ts

@@ -7,12 +7,10 @@ import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { runDetached } from '@/lib/core/utils/background'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { runTableDelete } from '@/lib/table/delete-runner'
 import { markTableJobRunning } from '@/lib/table/service'
-import { buildFilterClause, TableQueryValidationError } from '@/lib/table/sql'
-import type { Filter } from '@/lib/table/types'
-import { accessError, checkAccess } from '@/app/api/table/utils'
+import type { TableDeleteJobPayload } from '@/lib/table/types'
+import { accessError, checkAccess, tableFilterError } from '@/app/api/table/utils'
 
 const logger = createLogger('TableDeleteAsync')
 
@@ -57,24 +55,18 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
   }
 
   // Validate the filter up front so the caller gets immediate feedback (the worker reuses it).
-  if (filter) {
-    try {
-      buildFilterClause(filter as Filter, USER_TABLE_ROWS_SQL_NAME, table.schema.columns)
-    } catch (error) {
-      if (error instanceof TableQueryValidationError) {
-        return NextResponse.json({ error: error.message }, { status: 400 })
-      }
-      throw error
-    }
-  }
+  const filterError = tableFilterError(filter, table.schema.columns)
+  if (filterError) return filterError
 
   // Rows inserted after this instant are spared (created_at <= cutoff in the worker).
   const cutoff = new Date()
 
   // Atomically claim the job slot — one background job per table, so this also blocks while an
-  // import is in flight (and vice versa).
+  // import is in flight (and vice versa). The scope is persisted to the job's payload so read
+  // paths can mask the doomed rows while the job runs (see `pendingDeleteMask`).
   const jobId = generateId()
-  const claimed = await markTableJobRunning(tableId, jobId, 'delete')
+  const payload: TableDeleteJobPayload = { filter, excludeRowIds, cutoff: cutoff.toISOString() }
+  const claimed = await markTableJobRunning(tableId, jobId, 'delete', payload)
   if (!claimed) {
     return NextResponse.json(
       { error: 'A job is already in progress for this table' },
@@ -87,7 +79,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
       jobId,
       tableId,
       workspaceId,
-      filter: filter as Filter | undefined,
+      filter,
       excludeRowIds,
       cutoff,
     })

apps/sim/app/api/table/utils.ts

@@ -6,10 +6,32 @@ import {
   updateTableColumnBodySchema,
 } from '@/lib/api/contracts/tables'
 import type { MultipartError } from '@/lib/core/utils/multipart'
-import type { ColumnDefinition, TableDefinition } from '@/lib/table'
-import { getTableById } from '@/lib/table'
+import type { ColumnDefinition, Filter, TableDefinition } from '@/lib/table'
+import { buildFilterClause, getTableById, TableQueryValidationError } from '@/lib/table'
+import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
 
+/**
+ * Validates a `filter` against the table's column schema, returning a 400 response on a bad field
+ * (or `null` when the filter is valid or absent). Shared by the routes that accept a filter
+ * (`delete-async`, `columns/run`) so a bad field fails fast with a clear message.
+ */
+export function tableFilterError(
+  filter: Filter | undefined,
+  columns: ColumnDefinition[]
+): NextResponse | null {
+  if (!filter) return null
+  try {
+    buildFilterClause(filter, USER_TABLE_ROWS_SQL_NAME, columns)
+    return null
+  } catch (error) {
+    if (error instanceof TableQueryValidationError) {
+      return NextResponse.json({ error: error.message }, { status: 400 })
+    }
+    throw error
+  }
+}
+
 const logger = createLogger('TableUtils')
 
 /**

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx

@@ -11,7 +11,7 @@ import { Loader, TableX } from '@/components/emcn/icons'
 import type { RunLimit, RunMode, TableFindMatch } from '@/lib/api/contracts/tables'
 import { cn } from '@/lib/core/utils/cn'
 import { captureEvent } from '@/lib/posthog/client'
-import type { ColumnDefinition, TableRow as TableRowType, WorkflowGroup } from '@/lib/table'
+import type { ColumnDefinition, Filter, TableRow as TableRowType, WorkflowGroup } from '@/lib/table'
 import { getColumnId } from '@/lib/table/column-keys'
 import { TABLE_LIMITS } from '@/lib/table/constants'
 import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/providers/workspace-permissions-provider'
@@ -114,6 +114,8 @@ export interface SelectionSnapshot {
     rowIds: string[]
     allRows: boolean
     rowCount: number
+    /** Active filter when `allRows` is set — lets a filtered "select all" run only matching rows. */
+    filter?: Filter
   } | null
   /** Drives Play (`hasIncompleteOrFailed`) / Refresh (`hasCompleted`) /
    *  Stop (`hasInFlight`) visibility on the action bar. */
@@ -3360,13 +3362,15 @@ export function TableGrid({
     if (!rowSelectionIsEmpty(rowSelection)) {
       if (rowSelection.kind === 'all') {
         // `rowIds` is the loaded window (virtualized); the label total comes from the filter-scoped
-        // row count minus any deselected rows.
+        // row count minus any deselected rows. `filter` lets the run target the matching rows
+        // server-side (the dispatcher walks them) rather than the loaded window.
         const excluded = rowSelection.excluded?.size ?? 0
         return {
           groupIds: tableWorkflowGroupIds,
           rowIds: rows.map((r) => r.id),
           allRows: true,
           rowCount: Math.max(0, selectAllTotalRef.current - excluded),
+          filter: queryOptions.filter ?? undefined,
         }
       }
       const rowIds = rows.filter((r) => rowSelectionIncludes(rowSelection, r.id)).map((r) => r.id)

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx

@@ -251,6 +251,7 @@ export function Table({
     (args: {
       groupIds: string[]
       rowIds?: string[]
+      filter?: Filter
       runMode: RunMode
       limit?: RunLimit
       source: 'row' | 'rows' | 'column'
@@ -631,6 +632,8 @@ export function Table({
             runScope({
               groupIds: scope.groupIds,
               rowIds: scope.allRows ? undefined : scope.rowIds,
+              // `filter` is only populated on select-all, so it's already undefined otherwise.
+              filter: scope.filter,
               runMode: 'incomplete',
               source: 'rows',
             })
@@ -641,6 +644,7 @@ export function Table({
             runScope({
               groupIds: scope.groupIds,
               rowIds: scope.allRows ? undefined : scope.rowIds,
+              filter: scope.filter,
               runMode: 'all',
               source: 'rows',
             })

apps/sim/hooks/queries/tables.ts

@@ -1633,6 +1633,10 @@ interface RunColumnVariables {
   runMode?: RunMode
   /** Restrict to these rows. Server applies the same eligibility predicate. */
   rowIds?: string[]
+  /** "Select all under a filter" — run every row matching this filter (mutually exclusive with
+   *  `rowIds`). Optimistic stamping is skipped (like `limit`) since the matching set isn't known
+   *  client-side; the dispatcher's real pending stamps drive the UI. */
+  filter?: Filter
   /** Cap the run to the first `max` eligible rows. Omit for an unbounded run.
    *  Optimistic stamping is skipped when set — the dispatcher's real pending
    *  stamps drive the UI for the actual capped rows. */
@@ -1758,24 +1762,31 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
   const queryClient = useQueryClient()
 
   return useMutation({
-    mutationFn: async ({ groupIds, runMode = 'all', rowIds, limit }: RunColumnVariables) => {
+    mutationFn: async ({
+      groupIds,
+      runMode = 'all',
+      rowIds,
+      filter,
+      limit,
+    }: RunColumnVariables) => {
       return requestJson(runColumnContract, {
         params: { tableId },
         body: {
           workspaceId,
           groupIds,
           runMode,
           ...(rowIds && rowIds.length > 0 ? { rowIds } : {}),
+          ...(filter ? { filter } : {}),
           ...(limit ? { limit } : {}),
         },
       })
     },
-    onMutate: async ({ groupIds, runMode = 'all', rowIds, limit }) => {
-      // Capped runs touch only the first N eligible rows, chosen server-side by
-      // position. We can't predict that set client-side, so optimistic stamping
-      // is skipped — the dispatcher's real pending stamps (cell SSE) drive the
-      // UI within the first window.
-      if (limit)
+    onMutate: async ({ groupIds, runMode = 'all', rowIds, filter, limit }) => {
+      // Capped and filtered runs target a set we can't predict client-side (capped picks the first
+      // N by position; filtered matches a server-evaluated predicate), so optimistic stamping is
+      // skipped — the dispatcher's real pending stamps (cell SSE) drive the UI within the first
+      // window.
+      if (limit || filter)
         return { snapshots: undefined, runStateSnapshot: undefined, didBumpRunState: false }
       const targetRowIds = rowIds && rowIds.length > 0 ? new Set(rowIds) : null
       const targetGroupIds = new Set(groupIds)

apps/sim/lib/api/contracts/tables.ts

@@ -1109,6 +1109,9 @@ export const runColumnBodySchema = z.object({
   groupIds: z.array(z.string().min(1)).min(1),
   runMode: z.enum(['all', 'incomplete']).default('all'),
   rowIds: z.array(z.string().min(1)).min(1).optional(),
+  /** "Select all under a filter" — run every row matching this filter instead of `rowIds`. The
+   *  dispatcher walks only matching rows (paginated), so no id list is materialized. */
+  filter: nonEmptyFilterSchema.optional(),
   /** Cap the run to the first `max` eligible rows. Omit for an unbounded run. */
   limit: runLimitSchema.optional(),
 })

apps/sim/lib/table/dispatcher.ts

@@ -6,9 +6,11 @@ import { generateId } from '@sim/utils/id'
 import { and, asc, eq, gt, inArray, isNotNull, ne, or, type SQL, sql } from 'drizzle-orm'
 import { getJobQueue } from '@/lib/core/async-jobs/config'
 import { writeWorkflowGroupState } from '@/lib/table/cell-write'
+import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { isExecCancelledAfter } from '@/lib/table/deps'
 import { appendTableEvent } from '@/lib/table/events'
-import type { RowExecutionMetadata, RowExecutions, TableRow } from '@/lib/table/types'
+import { buildFilterClause } from '@/lib/table/sql'
+import type { Filter, RowExecutionMetadata, RowExecutions, TableRow } from '@/lib/table/types'
 import {
   buildEnqueueItems,
   buildPendingRuns,
@@ -32,6 +34,9 @@ export type DispatchMode = 'all' | 'incomplete' | 'new'
 export interface DispatchScope {
   groupIds: string[]
   rowIds?: string[]
+  /** "Select all matching a filter" — run every row matching this filter (mutually exclusive with
+   *  `rowIds`). Lets the action-bar Play/Refresh target a filtered view without materializing ids. */
+  filter?: Filter
 }
 
 /**
@@ -390,6 +395,15 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
   ]
   if (dispatch.scope.rowIds && dispatch.scope.rowIds.length > 0) {
     filters.push(inArray(userTableRows.id, dispatch.scope.rowIds))
+  } else if (dispatch.scope.filter) {
+    // "Select all under a filter": walk only the matching rows. Same cursor/window mechanism —
+    // non-matching rows are simply never selected, like mode eligibility.
+    const filterClause = buildFilterClause(
+      dispatch.scope.filter,
+      USER_TABLE_ROWS_SQL_NAME,
+      table.schema.columns
+    )
+    if (filterClause) filters.push(filterClause)
   }
   // `'new'` mode targets only rows whose targeted groups haven't been
   // attempted. Exclude a row only when EVERY targeted group already has a