improvement(tables): remove polling, eager drain, and parallelize batch updates

- Drop the per-page polling loop — SSE stream already patches execution
  cell state in real time and invalidates on buffer prune; polling was
  redundant and burned CPU/network on every open table
- Remove eager mount drain (fetchNextPage loop in use-table.ts); scroll
  handler and ensureAllRowsLoaded handle progressive/on-demand loading
- Parallelize chunkBatchUpdates with a 3-worker pool instead of serial
  chunks, reducing bulk-op round-trips by ~3x
- Delete mergePagePreservingIdentity and its tests (no longer called)

Author: waleedlatif1

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

@@ -198,16 +198,27 @@ interface TableGridProps {
  * an all-or-nothing result and partial success cannot leave the table in an
  * ambiguous half-cleared state.
  */
+const CHUNK_CONCURRENCY = 3
+
 async function chunkBatchUpdates(
   updates: Array<{ rowId: string; data: Record<string, unknown> }>,
   mutateAsync: (args: {
     updates: Array<{ rowId: string; data: Record<string, unknown> }>
   }) => Promise<unknown>
 ): Promise<void> {
   const size = TABLE_LIMITS.MAX_BULK_OPERATION_SIZE
+  const chunks: Array<Array<{ rowId: string; data: Record<string, unknown> }>> = []
   for (let i = 0; i < updates.length; i += size) {
-    await mutateAsync({ updates: updates.slice(i, i + size) })
+    chunks.push(updates.slice(i, i + size))
   }
+  let cursor = 0
+  await Promise.all(
+    Array.from({ length: Math.min(CHUNK_CONCURRENCY, chunks.length) }, async () => {
+      while (cursor < chunks.length) {
+        await mutateAsync({ updates: chunks[cursor++]! })
+      }
+    })
+  )
 }
 
 export function TableGrid({

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table.ts

@@ -1,6 +1,6 @@
 'use client'
 
-import { useCallback, useEffect, useMemo } from 'react'
+import { useCallback, useMemo } from 'react'
 import { useQueryClient } from '@tanstack/react-query'
 import type { ColumnDefinition, TableDefinition, TableRow, WorkflowGroup } from '@/lib/table'
 import { TABLE_LIMITS } from '@/lib/table/constants'
@@ -84,13 +84,6 @@ export function useTable({ workspaceId, tableId, queryOptions }: UseTableParams)
     enabled: Boolean(workspaceId && tableId),
   })
 
-  // prefetchInfiniteQuery is a no-op when data is fresh (staleTime not exceeded),
-  // so drive the drain through fetchNextPage — it appends one page at a time.
-  useEffect(() => {
-    if (!workspaceId || !tableId || !hasNextPage || isFetchingNextPage) return
-    void fetchNextPage()
-  }, [workspaceId, tableId, hasNextPage, isFetchingNextPage, fetchNextPage])
-
   const rows = useMemo<TableRow[]>(
     () => rowsData?.pages.flatMap((p) => p.rows) ?? [],
     [rowsData?.pages]

apps/sim/hooks/queries/tables.test.ts

@@ -82,7 +82,6 @@ vi.mock('@/components/emcn', () => ({
 }))
 
 import {
-  mergePagePreservingIdentity,
   tableKeys,
   tableRowsInfiniteOptions,
   tableRowsParamsKey,
@@ -371,75 +370,3 @@ describe('tableRowsInfiniteOptions', () => {
     expect(JSON.stringify(opts1.queryKey)).not.toBe(JSON.stringify(opts2.queryKey))
   })
 })
-
-describe('mergePagePreservingIdentity', () => {
-  const ts = '2024-01-01T00:00:00.000Z'
-  const ts2 = '2024-01-02T00:00:00.000Z'
-
-  function makeRow(id: string, updatedAt: string, extra?: Record<string, unknown>) {
-    return { id, updatedAt, data: { value: id, ...extra } }
-  }
-
-  it('returns fresh when totalCount differs', () => {
-    const prev = { rows: [makeRow('r1', ts)], totalCount: 1, nextOffset: undefined }
-    const fresh = { rows: [makeRow('r1', ts)], totalCount: 2, nextOffset: undefined }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result).toBe(fresh)
-  })
-
-  it('returns fresh when row counts differ', () => {
-    const prev = { rows: [makeRow('r1', ts)], totalCount: 2, nextOffset: undefined }
-    const fresh = {
-      rows: [makeRow('r1', ts), makeRow('r2', ts)],
-      totalCount: 2,
-      nextOffset: undefined,
-    }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result).toBe(fresh)
-  })
-
-  it('returns prev (same reference) when all rows are unchanged', () => {
-    const row1 = makeRow('r1', ts)
-    const row2 = makeRow('r2', ts)
-    const prev = { rows: [row1, row2], totalCount: 2, nextOffset: undefined }
-    const freshRow1 = makeRow('r1', ts)
-    const fresh = { rows: [freshRow1, makeRow('r2', ts)], totalCount: 2, nextOffset: undefined }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result).toBe(prev)
-  })
-
-  it('preserves identity for unchanged rows, uses fresh for updated rows', () => {
-    const row1 = makeRow('r1', ts)
-    const row2 = makeRow('r2', ts)
-    const prev = { rows: [row1, row2], totalCount: 2, nextOffset: undefined }
-    const updatedRow2 = makeRow('r2', ts2, { extra: 'new' })
-    const fresh = { rows: [makeRow('r1', ts), updatedRow2], totalCount: 2, nextOffset: undefined }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result).not.toBe(prev)
-    expect(result.rows[0]).toBe(row1)
-    expect(result.rows[1]).toBe(updatedRow2)
-  })
-
-  it('uses fresh row when ID is not found in prev', () => {
-    const row1 = makeRow('r1', ts)
-    const prev = { rows: [row1, makeRow('r2', ts)], totalCount: 2, nextOffset: undefined }
-    const newRow = makeRow('r3', ts)
-    const fresh = { rows: [makeRow('r1', ts), newRow], totalCount: 2, nextOffset: undefined }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result.rows[1]).toBe(newRow)
-  })
-
-  it('compares updatedAt as dates, not strings (ISO vs different string forms)', () => {
-    const row1 = makeRow('r1', ts)
-    const prev = { rows: [row1], totalCount: 1, nextOffset: undefined }
-    // Same point in time, different ISO string representation (with trailing Z vs +00:00)
-    const sameTimeDifferentFormat = '2024-01-01T00:00:00+00:00'
-    const fresh = {
-      rows: [makeRow('r1', sameTimeDifferentFormat)],
-      totalCount: 1,
-      nextOffset: undefined,
-    }
-    const result = mergePagePreservingIdentity(prev, fresh)
-    expect(result).toBe(prev)
-  })
-})

apps/sim/hooks/queries/tables.ts

@@ -4,7 +4,6 @@
  * React Query hooks for managing user-defined tables.
  */
 
-import { useEffect, useMemo } from 'react'
 import { createLogger } from '@sim/logger'
 import {
   type InfiniteData,
@@ -70,17 +69,10 @@ import type {
   WorkflowGroupDependencies,
   WorkflowGroupOutput,
 } from '@/lib/table'
-import {
-  areOutputsFilled,
-  hasRunningGroupExecution,
-  optimisticallyScheduleNewlyEligibleGroups,
-} from '@/lib/table/deps'
+import { areOutputsFilled, optimisticallyScheduleNewlyEligibleGroups } from '@/lib/table/deps'
 
 const logger = createLogger('TableQueries')
 
-const ROWS_POLL_INTERVAL_WHILE_RUNNING_MS = 2_000
-const ROWS_POLL_INTERVAL_IDLE_MS = 30_000
-
 type TableQueryScope = 'active' | 'archived' | 'all'
 
 export const tableKeys = {
@@ -251,24 +243,6 @@ export function useTableRows({
   })
 }
 
-/** @internal — exported for testing only. */
-export function mergePagePreservingIdentity(
-  prev: TableRowsResponse,
-  fresh: TableRowsResponse
-): TableRowsResponse {
-  if (prev.totalCount !== fresh.totalCount || prev.rows.length !== fresh.rows.length) return fresh
-  const prevById = new Map(prev.rows.map((r) => [r.id, r]))
-  let allSame = true
-  const nextRows = fresh.rows.map((freshRow) => {
-    const prevRow = prevById.get(freshRow.id)
-    if (prevRow && new Date(prevRow.updatedAt).getTime() === new Date(freshRow.updatedAt).getTime())
-      return prevRow
-    allSame = false
-    return freshRow
-  })
-  return allSame ? prev : { ...fresh, rows: nextRows }
-}
-
 export function tableRowsParamsKey({
   pageSize,
   filter,
@@ -316,99 +290,10 @@ export function useInfiniteTableRows({
   sort,
   enabled = true,
 }: InfiniteTableRowsParams) {
-  const queryClient = useQueryClient()
-  const paramsKey = tableRowsParamsKey({ pageSize, filter, sort })
-  // Memoize the key so the polling useEffect below doesn't fire on every render.
-  const queryKey = useMemo(() => tableKeys.infiniteRows(tableId, paramsKey), [tableId, paramsKey])
-
-  const query = useInfiniteQuery({
+  return useInfiniteQuery({
     ...tableRowsInfiniteOptions({ workspaceId, tableId, pageSize, filter, sort }),
-    queryKey,
     enabled: Boolean(workspaceId && tableId) && enabled,
   })
-
-  /**
-   * Per-page polling. Built-in `refetchInterval` would refetch every loaded
-   * page on each tick — wasteful when only one page has running cells.
-   * Instead, walk pages each tick and refetch ONLY the dirty ones, splicing
-   * results back into the cache. Polling stops when no page has in-flight
-   * cells, or while a mutation is running (optimistic-update guard).
-   */
-  useEffect(() => {
-    if (!enabled || !workspaceId || !tableId) return
-    let cancelled = false
-    let timeoutId: ReturnType<typeof setTimeout> | null = null
-    const tick = async () => {
-      if (cancelled) return
-      let hasDirty = false
-      if (queryClient.isMutating() !== 0) {
-        // Mutation in progress — skip network fetch to avoid racing optimistic
-        // updates, but stay on the short interval so we catch up quickly once
-        // the mutation settles.
-        hasDirty = true
-      } else {
-        const data = queryClient.getQueryData<InfiniteData<TableRowsResponse, number>>(queryKey)
-        const dirty: number[] = []
-        if (data) {
-          for (let i = 0; i < data.pages.length; i++) {
-            if (hasRunningGroupExecution(data.pages[i].rows)) {
-              dirty.push(data.pageParams[i] ?? i * pageSize)
-            }
-          }
-        }
-        hasDirty = dirty.length > 0
-        if (hasDirty) {
-          await Promise.all(
-            dirty.map(async (offset) => {
-              try {
-                const fresh = await fetchTableRows({
-                  workspaceId,
-                  tableId,
-                  limit: pageSize,
-                  offset,
-                  filter,
-                  sort,
-                  includeTotal: offset === 0,
-                })
-                if (cancelled) return
-                queryClient.setQueryData<InfiniteData<TableRowsResponse, number>>(
-                  queryKey,
-                  (prev) => {
-                    if (!prev) return prev
-                    const idx = prev.pageParams.indexOf(offset)
-                    if (idx === -1) return prev
-                    const merged = mergePagePreservingIdentity(prev.pages[idx], fresh)
-                    if (merged === prev.pages[idx]) return prev
-                    const nextPages = prev.pages.slice()
-                    nextPages[idx] = merged
-                    return { ...prev, pages: nextPages }
-                  }
-                )
-              } catch {
-                // Transient fetch failure — next tick retries. Don't kill the loop.
-              }
-            })
-          )
-        }
-      }
-      if (cancelled) return
-      // Recursive setTimeout instead of setInterval so a slow tick can't
-      // overlap the next one — out-of-order responses would otherwise let
-      // stale data overwrite fresh. Use a long interval when idle so tables
-      // with no running executions don't burn CPU on constant cache reads.
-      timeoutId = setTimeout(
-        () => void tick(),
-        hasDirty ? ROWS_POLL_INTERVAL_WHILE_RUNNING_MS : ROWS_POLL_INTERVAL_IDLE_MS
-      )
-    }
-    timeoutId = setTimeout(() => void tick(), ROWS_POLL_INTERVAL_WHILE_RUNNING_MS)
-    return () => {
-      cancelled = true
-      if (timeoutId !== null) clearTimeout(timeoutId)
-    }
-  }, [enabled, workspaceId, tableId, pageSize, filter, sort, queryClient, queryKey])
-
-  return query
 }
 
 /**