refactor(table): row executions sidecar + left-to-right dep retrigger + cancel counter refresh

Split per-row workflow-group execution state out of the user_table_rows.executions
JSONB column into a new table_row_executions sidecar keyed by (row_id, group_id).
Dispatcher filters, "X running" counter, bulk clears, and the cancellation guard
all hit indexed columns instead of walking JSONB. Wire shape unchanged — server
merges sidecar rows back into row.executions on the way out.

Also:
- deriveExecClearsForDataPatch now walks workflowGroups left-to-right with a
  propagating dirtied-column set so transitive dep chains (edit col A → group 1
  re-runs → group 2 depends on group 1's output → group 2 re-runs) collapse to
  a single forward pass.
- useCancelTableRuns.onSettled invalidates the activeDispatches query so the
  top-right counter and row gutter Stop button refetch from the server after
  any Stop (per-cell, row, or table-wide). countRunningCells is the source of
  truth; client no longer needs duplicate state.

Three migrations on this branch (0209 + 0210 + new sidecar) collapsed into one
since the feature is unreleased.

Author: TheodoreSpeaks

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

@@ -1,8 +1,8 @@
 import { db } from '@sim/db'
-import { userTableRows } from '@sim/db/schema'
+import { tableRowExecutions, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { and, eq, sql } from 'drizzle-orm'
+import { and, eq, inArray, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
   type BatchInsertTableRowsBodyInput,
@@ -17,7 +17,14 @@ import { isZodError, validationErrorResponse } from '@/lib/api/server/validation
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import type { Filter, RowData, Sort, TableSchema } from '@/lib/table'
+import type {
+  Filter,
+  RowData,
+  RowExecutionMetadata,
+  RowExecutions,
+  Sort,
+  TableSchema,
+} from '@/lib/table'
 import {
   batchInsertRows,
   batchUpdateRows,
@@ -277,7 +284,6 @@ export const GET = withRouteHandler(
         .select({
           id: userTableRows.id,
           data: userTableRows.data,
-          executions: userTableRows.executions,
           position: userTableRows.position,
           createdAt: userTableRows.createdAt,
           updatedAt: userTableRows.updatedAt,
@@ -308,6 +314,41 @@ export const GET = withRouteHandler(
 
       const rows = await query.limit(validated.limit).offset(validated.offset)
 
+      // Sidecar: fetch per-(row, group) execution state and group into a map
+      // so the response preserves the legacy `row.executions[groupId]` wire
+      // shape. One indexed-IN scan against table_row_executions.
+      const executionsByRow = new Map<string, RowExecutions>()
+      if (rows.length > 0) {
+        const execRows = await db
+          .select()
+          .from(tableRowExecutions)
+          .where(
+            inArray(
+              tableRowExecutions.rowId,
+              rows.map((r) => r.id)
+            )
+          )
+        for (const e of execRows) {
+          const existing = executionsByRow.get(e.rowId) ?? {}
+          const meta: RowExecutionMetadata = {
+            status: e.status as RowExecutionMetadata['status'],
+            executionId: e.executionId ?? null,
+            jobId: e.jobId ?? null,
+            workflowId: e.workflowId,
+            error: e.error ?? null,
+            ...(e.runningBlockIds && e.runningBlockIds.length > 0
+              ? { runningBlockIds: e.runningBlockIds }
+              : {}),
+            ...(e.blockErrors && Object.keys(e.blockErrors as Record<string, string>).length > 0
+              ? { blockErrors: e.blockErrors as Record<string, string> }
+              : {}),
+            ...(e.cancelledAt ? { cancelledAt: e.cancelledAt.toISOString() } : {}),
+          }
+          existing[e.groupId] = meta
+          executionsByRow.set(e.rowId, existing)
+        }
+      }
+
       logger.info(
         `[${requestId}] Queried ${rows.length} rows from table ${tableId} (total: ${totalCount ?? 'n/a'})`
       )
@@ -318,7 +359,7 @@ export const GET = withRouteHandler(
           rows: rows.map((r) => ({
             id: r.id,
             data: r.data,
-            executions: r.executions ?? {},
+            executions: executionsByRow.get(r.id) ?? {},
             position: r.position,
             createdAt:
               r.createdAt instanceof Date ? r.createdAt.toISOString() : String(r.createdAt),

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

@@ -1,7 +1,7 @@
 'use client'
 
-import { useEffect, useRef, useState } from 'react'
 import type React from 'react'
+import { useEffect, useRef, useState } from 'react'
 import { parse } from 'tldts'
 import { Badge, Checkbox, Tooltip } from '@/components/emcn'
 import { cn } from '@/lib/core/utils/cn'
@@ -65,7 +65,9 @@ export function resolveCellRender({
       // (workflow yielded for human-in-the-loop). Render as Pending rather
       // than Queued so the user can tell it's not just waiting to start.
       const isPaused =
-        exec?.status === 'pending' && typeof exec.jobId === 'string' && exec.jobId.startsWith('paused-')
+        exec?.status === 'pending' &&
+        typeof exec.jobId === 'string' &&
+        exec.jobId.startsWith('paused-')
       if (isPaused) return { kind: 'pending-upstream' }
       if (exec?.status === 'queued' || exec?.status === 'pending') return { kind: 'queued' }
       return { kind: 'pending-upstream' }

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

@@ -2793,9 +2793,7 @@ export function TableGrid({
     // it has a real executionId and a viewable trace, same as
     // running/completed/error.
     const isPaused =
-      status === 'pending' &&
-      typeof exec?.jobId === 'string' &&
-      exec.jobId.startsWith('paused-')
+      status === 'pending' && typeof exec?.jobId === 'string' && exec.jobId.startsWith('paused-')
     return {
       rowId: row.id,
       groupId,

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx

@@ -290,9 +290,7 @@ function WorkflowSidebarBody({
   // Deps default to none selected. With auto-run on, at least one is required
   // (enforced via `depsValid` below); a legacy group with empty deps will
   // surface the error on first open until the user picks at least one column.
-  const [deps, setDeps] = useState<string[]>(
-    () => existingGroup?.dependencies?.columns ?? []
-  )
+  const [deps, setDeps] = useState<string[]>(() => existingGroup?.dependencies?.columns ?? [])
   // `selectedOutputs` is encoded `${blockId}::${path}`. Seeded once `blockOutputGroups`
   // resolves (we may not have the workflow blocks loaded at first render); see the
   // post-load reconciliation below.
@@ -901,9 +899,7 @@ function WorkflowSidebarBody({
                   depOptions={depOptions}
                   deps={deps}
                   onChangeDeps={setDeps}
-                  error={
-                    showValidation && deps.length === 0 ? 'Select at least one column' : null
-                  }
+                  error={showValidation && deps.length === 0 ? 'Select at least one column' : null}
                 />
               </>
             )}

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

@@ -6,7 +6,7 @@ import { useQueryClient } from '@tanstack/react-query'
 import type { ActiveDispatch } from '@/lib/api/contracts/tables'
 import type { RowData, RowExecutionMetadata, RowExecutions } from '@/lib/table'
 import type { TableEvent, TableEventEntry } from '@/lib/table/events'
-import { snapshotAndMutateRows, tableKeys, type TableRunState } from '@/hooks/queries/tables'
+import { snapshotAndMutateRows, type TableRunState, tableKeys } from '@/hooks/queries/tables'
 
 const logger = createLogger('useTableEventStream')
 

apps/sim/background/workflow-column-execution.ts

@@ -6,7 +6,12 @@ import { generateId } from '@sim/utils/id'
 import { task } from '@trigger.dev/sdk'
 import { eq } from 'drizzle-orm'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
-import type { RowData, RowExecutionMetadata, TableDefinition, WorkflowGroup } from '@/lib/table/types'
+import type {
+  RowData,
+  RowExecutionMetadata,
+  TableDefinition,
+  WorkflowGroup,
+} from '@/lib/table/types'
 import type { WorkflowGroupCellPayload } from '@/lib/table/workflow-columns'
 
 export type { WorkflowGroupCellPayload }

apps/sim/hooks/queries/tables.ts

@@ -944,6 +944,11 @@ export function useCancelTableRuns({ workspaceId, tableId }: RowMutationContext)
     },
     onSettled: () => {
       queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) })
+      // Refetch the run-state snapshot — server re-derives runningCellCount +
+      // runningByRowId from the freshly-updated sidecar via countRunningCells.
+      // Without this, the counter and row gutter button stay stale until the
+      // user refetches manually.
+      queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) })
     },
   })
 }

apps/sim/lib/table/dispatcher.ts

@@ -1,19 +1,19 @@
 import { db } from '@sim/db'
-import { tableRunDispatches, userTableRows } from '@sim/db/schema'
+import { tableRowExecutions, tableRunDispatches, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { and, asc, eq, gt, inArray, type SQL, sql } from 'drizzle-orm'
 import { getJobQueue } from '@/lib/core/async-jobs/config'
 import { writeWorkflowGroupState } from '@/lib/table/cell-write'
 import { appendTableEvent } from '@/lib/table/events'
-import type { TableRow } from '@/lib/table/types'
+import type { RowExecutionMetadata, RowExecutions, TableRow } from '@/lib/table/types'
 import {
   buildEnqueueItems,
   buildPendingRuns,
-  type WorkflowGroupCellPayload,
   TABLE_CONCURRENCY_LIMIT,
   toTableRow,
+  type WorkflowGroupCellPayload,
 } from './workflow-columns'
 
 const logger = createLogger('TableRunDispatcher')
@@ -68,11 +68,10 @@ export async function bulkClearWorkflowGroupCells(input: {
   const outputCols = Array.from(new Set(groups.flatMap((g) => g.outputs.map((o) => o.columnName))))
   const groupIds = groups.map((g) => g.id)
 
-  // Build `data - 'col1' - 'col2' - ...` and `executions - 'gid1' - 'gid2' - ...`.
+  // Step 1: clear the targeted output columns from `data` on every row in
+  // scope. Identical chain to the previous JSONB-only path.
   let dataExpr: SQL = sql`coalesce(${userTableRows.data}, '{}'::jsonb)`
   for (const col of outputCols) dataExpr = sql`(${dataExpr}) - ${col}::text`
-  let execExpr: SQL = sql`coalesce(${userTableRows.executions}, '{}'::jsonb)`
-  for (const gid of groupIds) execExpr = sql`(${execExpr}) - ${gid}::text`
 
   const filters: SQL[] = [eq(userTableRows.tableId, tableId)]
   if (rowIds && rowIds.length > 0) {
@@ -87,22 +86,47 @@ export async function bulkClearWorkflowGroupCells(input: {
     )
     const allFilled = filledChecks.reduce((acc, expr) => sql`${acc} AND ${expr}`)
     filters.push(sql`NOT (${allFilled})`)
-    // Also skip rows where ANY targeted group has an in-flight exec from
-    // another dispatch — clobbering its `executions[gid]` would race with
-    // the in-flight worker. An `incomplete` run by definition shouldn't
-    // touch rows another dispatch is actively working on.
-    const inFlightChecks = groupIds.map(
-      (gid) =>
-        sql`${userTableRows.executions} -> ${gid}::text ->> 'status' IN ('queued', 'running', 'pending')`
+    // Also skip rows where ANY targeted group has an in-flight exec — those
+    // belong to another dispatch and clobbering them would race. Encoded as
+    // a NOT EXISTS subquery against the sidecar's `(table_id, status)`
+    // partial index.
+    filters.push(
+      sql`NOT EXISTS (
+        SELECT 1 FROM ${tableRowExecutions} re
+        WHERE re.row_id = ${userTableRows.id}
+          AND re.group_id = ANY(ARRAY[${sql.join(
+            groupIds.map((gid) => sql`${gid}`),
+            sql`, `
+          )}]::text[])
+          AND re.status IN ('queued', 'running', 'pending')
+      )`
     )
-    const anyInFlight = inFlightChecks.reduce((acc, expr) => sql`${acc} OR ${expr}`)
-    filters.push(sql`NOT (${anyInFlight})`)
   }
 
-  await db
-    .update(userTableRows)
-    .set({ data: dataExpr, executions: execExpr, updatedAt: new Date() })
-    .where(and(...filters))
+  await db.transaction(async (trx) => {
+    await trx
+      .update(userTableRows)
+      .set({ data: dataExpr, updatedAt: new Date() })
+      .where(and(...filters))
+
+    // Step 2: delete the targeted groups' executions for the rows in scope.
+    // Reuse the same row-scope filter via a subquery.
+    const execFilters: SQL[] = [
+      eq(tableRowExecutions.tableId, tableId),
+      inArray(tableRowExecutions.groupId, groupIds),
+    ]
+    if (rowIds && rowIds.length > 0) {
+      execFilters.push(inArray(tableRowExecutions.rowId, rowIds))
+    }
+    if (mode === 'incomplete') {
+      // For `incomplete`, only delete entries that aren't already in-flight
+      // — terminal states (completed/error/cancelled) get wiped so the
+      // dispatcher re-enqueues; in-flight entries stay so we don't race
+      // with their worker.
+      execFilters.push(sql`${tableRowExecutions.status} NOT IN ('queued', 'running', 'pending')`)
+    }
+    await trx.delete(tableRowExecutions).where(and(...execFilters))
+  })
 }
 
 export async function insertDispatch(input: {
@@ -142,27 +166,20 @@ export async function insertDispatch(input: {
 export async function countRunningCells(
   tableId: string
 ): Promise<{ total: number; byRowId: Record<string, number> }> {
+  // Hits the `(table_id, status)` partial index on table_row_executions.
   const rows = await db
     .select({
-      id: userTableRows.id,
-      runningCount: sql<number>`(
-        SELECT count(*)::int FROM jsonb_each(${userTableRows.executions}) e
-        WHERE e.value->>'status' = 'running'
-      )`,
+      rowId: tableRowExecutions.rowId,
+      runningCount: sql<number>`count(*)::int`,
     })
-    .from(userTableRows)
-    .where(
-      and(
-        eq(userTableRows.tableId, tableId),
-        sql`${userTableRows.executions} IS NOT NULL`,
-        sql`${userTableRows.executions} != '{}'::jsonb`
-      )
-    )
+    .from(tableRowExecutions)
+    .where(and(eq(tableRowExecutions.tableId, tableId), eq(tableRowExecutions.status, 'running')))
+    .groupBy(tableRowExecutions.rowId)
   let total = 0
   const byRowId: Record<string, number> = {}
   for (const r of rows) {
     if (r.runningCount > 0) {
-      byRowId[r.id] = r.runningCount
+      byRowId[r.rowId] = r.runningCount
       total += r.runningCount
     }
   }
@@ -265,19 +282,22 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     filters.push(inArray(userTableRows.id, dispatch.scope.rowIds))
   }
   // `'new'` mode targets only rows whose targeted groups haven't been
-  // attempted. Exclude a row only when EVERY targeted group already has an
-  // `executions[gid]` entry — if any one is missing, the row still has work
-  // to do and per-group JS filtering in `classifyEligibility` handles the
-  // rest. `jsonb_exists_all` is the function form of `?&` — safer than the
-  // operator, which collides with prepared-statement placeholder parsing in
-  // some drivers. Drizzle interpolates a JS array as a tuple of placeholders,
-  // not a Postgres array — emit `ARRAY[...]` literally.
+  // attempted. Exclude a row only when EVERY targeted group already has a
+  // sidecar entry — if any one is missing, the row still has work to do
+  // and per-group JS filtering in `classifyEligibility` handles the rest.
   if (dispatch.mode === 'new' && dispatch.scope.groupIds.length > 0) {
+    const gids = dispatch.scope.groupIds
     filters.push(
-      sql`NOT jsonb_exists_all(coalesce(${userTableRows.executions}, '{}'::jsonb), ARRAY[${sql.join(
-        dispatch.scope.groupIds.map((gid) => sql`${gid}`),
-        sql`, `
-      )}]::text[])`
+      sql`NOT EXISTS (
+        SELECT 1 FROM ${tableRowExecutions} re
+        WHERE re.row_id = ${userTableRows.id}
+          AND re.group_id = ANY(ARRAY[${sql.join(
+            gids.map((gid) => sql`${gid}`),
+            sql`, `
+          )}]::text[])
+        GROUP BY re.row_id
+        HAVING count(DISTINCT re.group_id) = ${gids.length}
+      )`
     )
   }
 
@@ -303,12 +323,40 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     return 'done'
   }
 
+  // Pre-fetch executions for the chunk so per-row eligibility doesn't fan
+  // out into one query per row. Returns `Map<rowId, RowExecutions>`.
+  const chunkRowIds = chunk.map((r) => r.id)
+  const execRows = await db
+    .select()
+    .from(tableRowExecutions)
+    .where(inArray(tableRowExecutions.rowId, chunkRowIds))
+  const executionsByRow = new Map<string, RowExecutions>()
+  for (const r of execRows) {
+    const existing = executionsByRow.get(r.rowId) ?? {}
+    const meta: RowExecutionMetadata = {
+      status: r.status as RowExecutionMetadata['status'],
+      executionId: r.executionId ?? null,
+      jobId: r.jobId ?? null,
+      workflowId: r.workflowId,
+      error: r.error ?? null,
+      ...(r.runningBlockIds && r.runningBlockIds.length > 0
+        ? { runningBlockIds: r.runningBlockIds }
+        : {}),
+      ...(r.blockErrors && Object.keys(r.blockErrors as Record<string, string>).length > 0
+        ? { blockErrors: r.blockErrors as Record<string, string> }
+        : {}),
+      ...(r.cancelledAt ? { cancelledAt: r.cancelledAt.toISOString() } : {}),
+    }
+    existing[r.groupId] = meta
+    executionsByRow.set(r.rowId, existing)
+  }
+
   // Strip rows the user cancelled mid-cascade (post-dispatch tombstones)
   // before running the shared eligibility filter — `buildPendingRuns`
   // doesn't know about the per-dispatch cancel tombstone.
   const tombstoneFiltered: TableRow[] = []
   for (const r of chunk) {
-    const tableRow = toTableRow(r)
+    const tableRow = toTableRow(r, executionsByRow.get(r.id) ?? {})
     const tombstoned = dispatch.scope.groupIds.some((gid) => {
       const exec = tableRow.executions?.[gid]
       if (!exec?.cancelledAt) return false