Function processBackfillPage

apps/sim/lib/table/backfill-runner.ts:101–182

* Backfills one page of rows: pulls each target output's value out of the rows' saved trace * spans (materialized from object storage with bounded concurrency) and writes it into row data. * Returns the number of rows updated.

(opts: {
  table: TableDefinition
  outputs: WorkflowGroupOutput[]
  overwrite: boolean
  execs: Array<{ rowId: string; executionId: string | null }>
  requestId: string
  actorUserId?: string | null
})

 * Returns the number of rows updated.
 */
async function processBackfillPage(opts: {
  table: TableDefinition
  outputs: WorkflowGroupOutput[]
  overwrite: boolean
  execs: Array<{ rowId: string; executionId: string | null }>
  requestId: string
  actorUserId?: string | null
}): Promise<number> {
  const { table, outputs, overwrite, execs, requestId, actorUserId } = opts

  const executionIdsByRow = new Map<string, string>()
  for (const e of execs) {
    if (!e.executionId) continue
    executionIdsByRow.set(e.rowId, e.executionId)
  }
  if (executionIdsByRow.size === 0) return 0

  const rowRecords = await db
    .select({ id: userTableRows.id, data: userTableRows.data })
    .from(userTableRows)
    .where(
      and(
        eq(userTableRows.tableId, table.id),
        inArray(userTableRows.id, Array.from(executionIdsByRow.keys()))
      )
    )

  const executionIds = Array.from(new Set(executionIdsByRow.values()))
  const logs = await db
    .select({
      executionId: workflowExecutionLogs.executionId,
      workflowId: workflowExecutionLogs.workflowId,
      workspaceId: workflowExecutionLogs.workspaceId,
      executionData: workflowExecutionLogs.executionData,
    })
    .from(workflowExecutionLogs)
    .where(inArray(workflowExecutionLogs.executionId, executionIds))

  const logByExecutionId = new Map<string, { traceSpans?: BackfillTraceSpan[] }>()
  // Heavy execution data may live in object storage; resolve pointers (bounded concurrency).
  await mapWithConcurrency(logs, MATERIALIZE_CONCURRENCY, async (log) => {
    const executionData = await materializeExecutionData(
      log.executionData as Record<string, unknown> | null,
      { workspaceId: log.workspaceId, workflowId: log.workflowId, executionId: log.executionId }
    )
    logByExecutionId.set(
      log.executionId,
      (executionData as { traceSpans?: BackfillTraceSpan[] }) ?? {}
    )
  })

  const updates: Array<{ rowId: string; data: RowData }> = []
  for (const r of rowRecords) {
    const execId = executionIdsByRow.get(r.id)
    if (!execId) continue
    const log = logByExecutionId.get(execId)
    if (!log) continue