improvement(tables): cut the DB round trips a table read and write spend on protocol

The grid's first page spent more time on round trips than on work. Four of them
were avoidable:

- `pendingDeleteMask` probed `table_jobs` on every read, though the table a
  request just loaded already carries its latest non-export job, and the
  `one_active_per_table` unique index makes that row the running delete when one
  exists. Callers that hold a table across a long walk (the export stream, the
  snapshot builder) keep probing per page, so a delete starting mid-walk still
  begins masking.
- The run-state sidecar was read for every table, including the ones that
  declare no workflow group and therefore cannot have a row — four chunked
  queries on a 1000-row page, all returning nothing.
- The drain opened a transaction per batch. The guards are fixed for the call,
  so each extra batch paid `BEGIN` + `set_config` + `COMMIT` for nothing.
- `setTableTxTimeouts` issued three `SET LOCAL` statements; `set_config(…, true)`
  is the same thing and fits in one round trip, as the read guards already do.

A 1000-row page goes from 22 statements to 14, a 50-row page from 15 to 13, and
every write transaction drops two.

Author: waleedlatif1

apps/sim/app/api/v1/tables/[tableId]/rows/route.ts

@@ -185,6 +185,8 @@ export const GET = withRouteHandler(async (request: NextRequest, context: TableR
         offset: validated.offset,
         includeTotal: validated.includeTotal,
         withExecutions: false,
+        // `table` was loaded a few lines above, for this read.
+        trustLoadedJob: true,
       },
       requestId
     )

apps/sim/lib/table/__tests__/service-filter-threading.test.ts

@@ -50,6 +50,7 @@ vi.mock('@/lib/table/rows/executions', () => ({
   })),
   loadExecutionsByRow: mockLoadExecutionsByRow,
   loadExecutionsForRow: vi.fn(async () => ({})),
+  tableMayHaveRunState: vi.fn(() => true),
   writeExecutionsPatch: vi.fn(async () => 'wrote'),
 }))
 
@@ -65,6 +66,7 @@ vi.mock('@/lib/table/validation', () => ({
   checkBatchUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
 }))
 
+import { tableMayHaveRunState } from '@/lib/table/rows/executions'
 import {
   deleteRow,
   deleteRowsByFilter,
@@ -666,3 +668,89 @@ describe('queryRows byte budget', () => {
     })
   })
 })
+
+/**
+ * Two reads the row path used to make unconditionally, each one round trip on the grid's hot
+ * page. Both are now answered from state the caller already holds; these pin that the query is
+ * actually skipped rather than merely ignored, and that the fallbacks still run.
+ */
+describe('queryRows round-trip elision', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    vi.mocked(tableMayHaveRunState).mockReturnValue(true)
+  })
+
+  const hydrated = (
+    fields: Partial<Pick<TableDefinition, 'jobStatus' | 'jobType'>>
+  ): TableDefinition => ({ ...TABLE, jobStatus: null, jobType: null, ...fields })
+
+  it('skips the delete-job probe when the hydrated table shows no running delete', async () => {
+    await queryRows(
+      hydrated({}),
+      { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true },
+      'req-1'
+    )
+
+    // With no probe, the drain batch is the FIRST bounded query rather than the second.
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 6)
+  })
+
+  it('still probes when the hydrated table shows a running delete job', async () => {
+    await queryRows(
+      hydrated({ jobStatus: 'running', jobType: 'delete' }),
+      { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true },
+      'req-1'
+    )
+
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1)
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6)
+  })
+
+  /** An unhydrated definition cannot rule the job out, so it keeps the lookup it always had. */
+  it('still probes when the table carries no job fields at all', async () => {
+    await queryRows(
+      TABLE,
+      { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true },
+      'req-1'
+    )
+
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1)
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6)
+  })
+
+  /**
+   * A caller that holds its table across a long walk (the export stream) must keep re-asking, so
+   * a delete job starting mid-walk still begins masking its doomed rows.
+   */
+  it('still probes for a caller that did not vouch for its table', async () => {
+    await queryRows(hydrated({}), { limit: 5, includeTotal: false, withExecutions: false }, 'req-1')
+
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1)
+    expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6)
+  })
+
+  it('skips the run-state read for a table that can hold none, still reporting empty executions', async () => {
+    vi.mocked(tableMayHaveRunState).mockReturnValue(false)
+    dbChainMockFns.limit.mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'row-1', data: {}, position: 0, orderKey: 'a0', createdAt: null, updatedAt: null },
+    ])
+
+    const result = await queryRows(TABLE, { limit: 5, includeTotal: false }, 'req-1')
+
+    expect(mockLoadExecutionsByRow).not.toHaveBeenCalled()
+    expect(result.rows[0].executions).toEqual({})
+  })
+
+  it('reads run state for a table that can hold it', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'row-1', data: {}, position: 0, orderKey: 'a0', createdAt: null, updatedAt: null },
+    ])
+
+    await queryRows(TABLE, { limit: 5, includeTotal: false }, 'req-1')
+
+    expect(mockLoadExecutionsByRow).toHaveBeenCalledTimes(1)
+  })
+})

apps/sim/lib/table/__tests__/update-row.test.ts

@@ -68,6 +68,31 @@ function findExecutedRawSql(substring: string): string | undefined {
   return undefined
 }
 
+/** Every string reachable from a drizzle `sql` fragment — its literal chunks AND its bound values. */
+function collectStrings(node: unknown, out: string[] = []): string[] {
+  if (typeof node === 'string') out.push(node)
+  else if (Array.isArray(node)) for (const entry of node) collectStrings(entry, out)
+  else if (node && typeof node === 'object')
+    for (const entry of Object.values(node as Record<string, unknown>)) collectStrings(entry, out)
+  return out
+}
+
+/**
+ * Whether one `set_config(<setting>, '<value>', true)` guard was executed.
+ *
+ * The guards bind their values as parameters, so the setting name lives in the statement's
+ * literal chunks while the duration lives in its bound values — asserting on a rendered string
+ * would only re-check the placeholder.
+ */
+function executedTxTimeout(setting: string, value: string): boolean {
+  return dbChainMockFns.execute.mock.calls.some(([arg]) => {
+    const strings = collectStrings(arg)
+    return (
+      strings.some((entry) => entry.includes(`set_config('${setting}'`)) && strings.includes(value)
+    )
+  })
+}
+
 /**
  * The `data` payload of the last `.set(...)` row write. `updateRow` always writes a JSONB merge
  * (`data = data || {changed}::jsonb`), so this is a `sql` fragment exposing `{ strings, values }`.
@@ -420,11 +445,9 @@ describe('mutation paths — SET LOCAL timeouts', () => {
       insertRow({ tableId: 'tbl-1', data: { name: 'a' }, workspaceId: 'ws-1' }, TABLE, 'req-1')
     ).rejects.toBeDefined()
 
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '10000ms'")).toBeDefined()
-    expect(findExecutedRawSql("SET LOCAL lock_timeout = '3000ms'")).toBeDefined()
-    expect(
-      findExecutedRawSql("SET LOCAL idle_in_transaction_session_timeout = '5000ms'")
-    ).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '10000ms')).toBe(true)
+    expect(executedTxTimeout('lock_timeout', '3000ms')).toBe(true)
+    expect(executedTxTimeout('idle_in_transaction_session_timeout', '5000ms')).toBe(true)
   })
 
   it('batchInsertRows raises statement_timeout to 60s', async () => {
@@ -436,7 +459,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
       )
     ).rejects.toBeDefined()
 
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '60000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '60000ms')).toBe(true)
   })
 
   it('replaceTableRows scales statement_timeout with (existing + new) row count', async () => {
@@ -450,7 +473,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     )
 
     // (100_000 + 50_000) × 3ms/row = 450_000ms; above 120_000 floor, below 600_000 cap
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '450000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '450000ms')).toBe(true)
   })
 
   it('replaceTableRows caps scaled timeout at 10 minutes for very large tables', async () => {
@@ -459,7 +482,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     await replaceTableRows({ tableId: 'tbl-1', workspaceId: 'ws-1', rows: [] }, hugeTable, 'req-1')
 
     // 10M × 3ms = 30M ms, capped at 600_000ms (10 min)
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '600000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '600000ms')).toBe(true)
   })
 
   it('replaceTableRows uses the 120s floor on small tables', async () => {
@@ -472,7 +495,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     )
 
     // 12 × 3ms = 36ms → floored at 120_000ms
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '120000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '120000ms')).toBe(true)
   })
 
   it('renameColumn is metadata-only — no per-row JSONB rewrite regardless of row count', async () => {
@@ -491,7 +514,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     await deleteColumn({ tableId: 'tbl-1', columnName: 'age' }, 'req-1')
 
     // 100 × 2ms = 200ms → floored at 60_000ms
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '60000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '60000ms')).toBe(true)
   })
 
   it('replaceTableRows acquires the per-table advisory lock to serialize concurrent replaces', async () => {

apps/sim/lib/table/application/rows.test.ts

@@ -769,6 +769,7 @@ describe('row query and upsert application semantics', () => {
         includeTotal: false,
         withExecutions: false,
         runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES,
+        trustLoadedJob: true,
       },
       expect.any(String)
     )

apps/sim/lib/table/application/rows.ts

@@ -436,6 +436,8 @@ export const listTableRows = defineAuthorizedTableUseCase({
           includeTotal: false,
           withExecutions: input.includeRunState ?? false,
           runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES,
+          // `context.table` was loaded by this use case's own resolver, for this read.
+          trustLoadedJob: true,
         },
         requestId(input)
       )
@@ -573,6 +575,8 @@ export const queryTableRows = defineAuthorizedTableUseCase({
           withExecutions: input.includeRunState ?? false,
           runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES,
           columnIds,
+          // `context.table` was loaded by this use case's own resolver, for this read.
+          trustLoadedJob: true,
         },
         requestId(input),
         readProvenance

apps/sim/lib/table/rows/executions.test.ts

@@ -4,8 +4,12 @@
 import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { DbOrTx } from '@/lib/db/types'
-import { loadExecutionsByRow, writeExecutionsPatch } from '@/lib/table/rows/executions'
-import type { RowExecutionMetadata } from '@/lib/table/types'
+import {
+  loadExecutionsByRow,
+  tableMayHaveRunState,
+  writeExecutionsPatch,
+} from '@/lib/table/rows/executions'
+import type { RowExecutionMetadata, TableSchema } from '@/lib/table/types'
 
 const EXECUTION_STATE: RowExecutionMetadata = {
   status: 'running',
@@ -258,3 +262,39 @@ describe('loadExecutionsByRow', () => {
     expect(byRow.size).toBe(750)
   })
 })
+
+describe('tableMayHaveRunState', () => {
+  const column = (overrides: Partial<TableSchema['columns'][number]> = {}) => ({
+    id: 'col_1',
+    name: 'title',
+    type: 'string' as const,
+    ...overrides,
+  })
+
+  it('is false for a schema that declares no group', () => {
+    expect(tableMayHaveRunState({ columns: [column()] })).toBe(false)
+    expect(tableMayHaveRunState({ columns: [column()], workflowGroups: [] })).toBe(false)
+  })
+
+  it('is true once the schema declares a group', () => {
+    expect(
+      tableMayHaveRunState({
+        columns: [column()],
+        workflowGroups: [{ id: 'group-1' }] as TableSchema['workflowGroups'],
+      })
+    ).toBe(true)
+  })
+
+  /**
+   * A column still pointing at a group is group state whatever the group list says, so an
+   * unexpected schema shape must keep the sidecar read rather than silently drop run state.
+   */
+  it('is true for a column that still names a group the list has lost', () => {
+    expect(
+      tableMayHaveRunState({
+        columns: [column({ workflowGroupId: 'group-1' })],
+        workflowGroups: [],
+      })
+    ).toBe(true)
+  })
+})

apps/sim/lib/table/rows/executions.ts

@@ -40,6 +40,24 @@ interface LoadExecutionsOptions {
   budgetBytes?: number
 }
 
+/**
+ * Whether a table can have any run-state sidecar at all.
+ *
+ * `tableRowExecutions` is keyed by `(rowId, groupId)`, and every writer takes its `groupId` from
+ * a group on the table's own schema. Group and column deletes strip the matching sidecar rows in
+ * the same transaction that removes the group ({@link stripGroupExecutions}), so a schema that
+ * declares no group cannot have a surviving row — the sidecar read would return nothing, and the
+ * caller would fill in the same empty map it gets by skipping.
+ *
+ * Both signals are checked rather than just `workflowGroups`: a column still carrying a
+ * `workflowGroupId` means the table has group state whatever the group list looks like, so an
+ * unexpected schema shape keeps the query instead of silently dropping run state.
+ */
+export function tableMayHaveRunState(schema: TableSchema): boolean {
+  if (schema.workflowGroups && schema.workflowGroups.length > 0) return true
+  return schema.columns.some((column) => column.workflowGroupId !== undefined)
+}
+
 /**
  * Loads `tableRowExecutions` rows for the given row ids and groups them into a
  * `Map<rowId, RowExecutions>` suitable for plugging into `TableRow.executions`.

apps/sim/lib/table/rows/pending-delete-mask.ts

@@ -9,6 +9,43 @@ import type { TableDefinition, TableDeleteJobPayload } from '@/lib/table/types'
 
 const logger = createLogger('TablePendingDeleteMask')
 
+/**
+ * Whether {@link PendingDeleteMaskOptions.trustLoadedJob} can rule a running delete job out.
+ *
+ * Every table loaded through `getTableById` / `listTables` already carries its latest non-export
+ * job, folded into that same SELECT as a lateral, so a caller that loaded its table for this very
+ * read already holds the answer and the lookup is pure overhead on the hot path.
+ *
+ * The derivation is exact, not a heuristic. `table_jobs_one_active_per_table` is unique on
+ * `table_id WHERE status = 'running' AND type <> 'export'` — the same predicate the lateral
+ * filters on — so a running delete job is the ONLY running non-export job on its table, and no
+ * further non-export job can be inserted while it holds that slot. It is therefore the newest
+ * non-export job by `started_at`, which is exactly the row the lateral returns.
+ *
+ * `jobStatus === undefined` means the fields were never hydrated (a `TableDefinition` assembled
+ * by some other path), which is indistinguishable from "no job" in the shape alone — so that case
+ * falls back to the query rather than assuming. A hydrated table with no job has
+ * `jobStatus: null`.
+ */
+function hydratedJobRulesOutDelete(table: TableDefinition): boolean {
+  if (table.jobStatus === undefined) return false
+  return !(table.jobStatus === 'running' && table.jobType === 'delete')
+}
+
+export interface PendingDeleteMaskOptions {
+  /**
+   * Answer from `table`'s own latest-job fields when they rule a running delete out, instead of
+   * querying for one.
+   *
+   * Only for a caller whose `table` was loaded for this read: the fields are then as fresh as the
+   * query would have been. A caller that loads a table once and then pages for a while — the
+   * export runner, the snapshot builder — must NOT set this, because a delete job starting
+   * mid-walk would never appear in its snapshot and its later pages would stop masking doomed
+   * rows. Those callers keep re-asking per page, which is what makes the mask appear mid-walk.
+   */
+  trustLoadedJob?: boolean
+}
+
 /**
  * Visibility mask for a running delete job: returns a clause keeping only rows the job will NOT
  * delete, or `undefined` when no delete job is running. The job's persisted scope
@@ -20,7 +57,11 @@ const logger = createLogger('TablePendingDeleteMask')
  * `(doomed) IS NOT TRUE` rather than `NOT (doomed)`: JSONB predicates evaluate to NULL on missing
  * cells, and those rows are NOT selected for deletion (NULL ≠ TRUE) — they must stay visible.
  */
-export async function pendingDeleteMask(table: TableDefinition): Promise<SQL | undefined> {
+export async function pendingDeleteMask(
+  table: TableDefinition,
+  options?: PendingDeleteMaskOptions
+): Promise<SQL | undefined> {
+  if (options?.trustLoadedJob && hydratedJobRulesOutDelete(table)) return undefined
   const [job] = await db
     .select({ payload: tableJobs.payload })
     .from(tableJobs)