fix(tables): stream large CSV imports from storage instead of buffering the whole file

Author: TheodoreSpeaks

apps/sim/lib/table/import-runner.ts

@@ -1,4 +1,4 @@
-import { Readable } from 'node:stream'
+import { Transform } from 'node:stream'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
@@ -26,7 +26,7 @@ import {
   setTableSchemaForImport,
   updateImportProgress,
 } from '@/lib/table/service'
-import { downloadFile } from '@/lib/uploads/core/storage-service'
+import { downloadFileStream, headObject } from '@/lib/uploads/core/storage-service'
 import { normalizeColumn } from '@/app/api/table/utils'
 
 const logger = createLogger('TableImportRunner')
@@ -70,38 +70,31 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
     if (!loaded) throw new Error(`Import target table ${tableId} not found`)
     const table = loaded
 
-    const buffer = await downloadFile({ key: fileKey, context: 'workspace' })
+    // Total byte size for the progress estimate — a cheap HEAD, no download. May be null on
+    // the local dev provider, in which case the bar stays indeterminate (rows still show).
+    const totalBytes = (await headObject(fileKey, 'workspace'))?.size ?? 0
 
-    // Delete only after the download succeeds — otherwise a failed download would wipe the
-    // table with nothing to replace it with.
-    if (mode === 'replace') await deleteAllTableRows(tableId)
+    // Stream the file rather than buffering it — a ~1M-row import must never be held in memory.
+    const source = await downloadFileStream({ key: fileKey, context: 'workspace' })
 
-    // Estimate total data rows by counting line breaks (minus the header) for a
-    // determinate progress bar. It's an estimate — quoted newlines and blank lines
-    // make it imprecise — so the client caps the bar below 100% until the terminal
-    // `ready` event lands. Cheap: one O(bytes) pass over the already-buffered file.
-    let newlineCount = 0
-    for (let i = 0; i < buffer.length; i++) {
-      if (buffer[i] === 0x0a) newlineCount++
-    }
-    const estimatedTotal = Math.max(0, newlineCount - 1)
+    // Delete only after the stream opens (a missing object rejects above) — otherwise a failed
+    // download would wipe the table with nothing to replace it with.
+    if (mode === 'replace') await deleteAllTableRows(tableId)
 
-    // Publish the estimated total up front so the client shows a determinate bar at 0%
-    // immediately, instead of "0 rows and counting" until the first batch lands.
-    void appendTableEvent({
-      kind: 'import',
-      tableId,
-      importId,
-      status: 'importing',
-      progress: 0,
-      total: estimatedTotal,
+    // Count bytes as they flow so the row total can be extrapolated from byte progress.
+    let bytesRead = 0
+    const byteCounter = new Transform({
+      transform(chunk: Buffer, _enc, cb) {
+        bytesRead += chunk.length
+        cb(null, chunk)
+      },
     })
 
     const parser = createCsvParser(delimiter)
     // `.pipe` doesn't forward source errors; forward so the iterator throws.
-    const source = Readable.from(buffer)
     source.on('error', (err) => parser.destroy(err))
-    source.pipe(parser)
+    byteCounter.on('error', (err) => parser.destroy(err))
+    source.pipe(byteCounter).pipe(parser)
 
     let schema: TableSchema | null = null
     let headerToColumn: Map<string, string> | null = null
@@ -173,9 +166,19 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
         { ...table, schema },
         requestId
       )
-      if (inserted - lastReported >= PROGRESS_INTERVAL_ROWS) {
+      // Emit after the first batch lands, then every interval, so the bar appears early.
+      if (
+        inserted - lastReported >= PROGRESS_INTERVAL_ROWS ||
+        (lastReported === 0 && inserted > 0)
+      ) {
         lastReported = inserted
         await updateImportProgress(tableId, inserted)
+        // Extrapolate the total from rows-per-byte observed so far; self-refines as it runs.
+        // `Math.max(inserted, …)` keeps it monotonic; omit when the byte size is unknown.
+        const estimatedTotal =
+          totalBytes > 0 && bytesRead > 0
+            ? Math.max(inserted, Math.round((inserted / bytesRead) * totalBytes))
+            : undefined
         void appendTableEvent({
           kind: 'import',
           tableId,

apps/sim/lib/uploads/core/storage-service.ts

@@ -1,3 +1,4 @@
+import type { Readable } from 'node:stream'
 import { randomBytes } from 'crypto'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
@@ -222,6 +223,34 @@ export async function downloadFile(options: DownloadFileOptions): Promise<Buffer
   return readFile(filePath)
 }
 
+/**
+ * Stream a file out of the configured storage provider without buffering it in memory.
+ * The caller MUST fully consume or `destroy()` the returned stream. Used by the large-CSV
+ * import worker so a multi-hundred-MB file is never held resident.
+ */
+export async function downloadFileStream(options: {
+  key: string
+  context: StorageContext
+}): Promise<Readable> {
+  const { key, context } = options
+  const config = getStorageConfig(context)
+
+  if (USE_BLOB_STORAGE) {
+    const { downloadFromBlobStream } = await import('@/lib/uploads/providers/blob/client')
+    return downloadFromBlobStream(key, createBlobConfig(config))
+  }
+
+  if (USE_S3_STORAGE) {
+    const { downloadFromS3Stream } = await import('@/lib/uploads/providers/s3/client')
+    return downloadFromS3Stream(key, createS3Config(config))
+  }
+
+  const { createReadStream } = await import('fs')
+  const { join } = await import('path')
+  const { UPLOAD_DIR_SERVER } = await import('./setup.server')
+  return createReadStream(join(UPLOAD_DIR_SERVER, sanitizeFileKey(key)))
+}
+
 /**
  * Delete a file from the configured storage provider
  */

apps/sim/lib/uploads/providers/blob/client.ts

@@ -1,3 +1,4 @@
+import type { Readable } from 'node:stream'
 import type { BlobServiceClient as BlobServiceClientType } from '@azure/storage-blob'
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
@@ -341,6 +342,49 @@ export async function downloadFromBlob(
   return downloaded
 }
 
+/**
+ * Stream a blob out of storage without buffering it. The caller MUST fully consume or
+ * `destroy()` the returned stream. Used by the large-CSV import worker.
+ */
+export async function downloadFromBlobStream(
+  key: string,
+  customConfig?: BlobConfig
+): Promise<Readable> {
+  const { BlobServiceClient, StorageSharedKeyCredential } = await import('@azure/storage-blob')
+  let blobServiceClient: BlobServiceClientType
+  let containerName: string
+
+  if (customConfig) {
+    if (customConfig.connectionString) {
+      blobServiceClient = BlobServiceClient.fromConnectionString(customConfig.connectionString)
+    } else if (customConfig.accountName && customConfig.accountKey) {
+      const credential = new StorageSharedKeyCredential(
+        customConfig.accountName,
+        customConfig.accountKey
+      )
+      blobServiceClient = new BlobServiceClient(
+        `https://${customConfig.accountName}.blob.core.windows.net`,
+        credential
+      )
+    } else {
+      throw new Error('Invalid custom blob configuration')
+    }
+    containerName = customConfig.containerName
+  } else {
+    blobServiceClient = await getBlobServiceClient()
+    containerName = BLOB_CONFIG.containerName
+  }
+
+  const containerClient = blobServiceClient.getContainerClient(containerName)
+  const blockBlobClient = containerClient.getBlockBlobClient(key)
+
+  const downloadBlockBlobResponse = await blockBlobClient.download()
+  if (!downloadBlockBlobResponse.readableStreamBody) {
+    throw new Error('Failed to get readable stream from blob download')
+  }
+  return downloadBlockBlobResponse.readableStreamBody as Readable
+}
+
 /**
  * Check whether a blob exists (and return its size when it does).
  * Returns null when the blob is missing.

apps/sim/lib/uploads/providers/s3/client.ts

@@ -1,3 +1,4 @@
+import type { Readable } from 'node:stream'
 import {
   AbortMultipartUploadCommand,
   CompleteMultipartUploadCommand,
@@ -221,6 +222,24 @@ export async function downloadFromS3(
   })
 }
 
+/**
+ * Stream an object out of S3 without buffering it. The caller MUST fully consume or
+ * `destroy()` the returned stream. Used by the large-CSV import worker so a 1M-row file is
+ * never resident in memory.
+ */
+export async function downloadFromS3Stream(
+  key: string,
+  customConfig?: S3Config
+): Promise<Readable> {
+  const config = customConfig || { bucket: S3_CONFIG.bucket, region: S3_CONFIG.region }
+  const command = new GetObjectCommand({ Bucket: config.bucket, Key: key })
+  const response = await getS3Client().send(command)
+  if (!response.Body) {
+    throw new Error(`S3 object has no body: ${key}`)
+  }
+  return response.Body as Readable
+}
+
 /**
  * Check whether an object exists in S3 (and return its size when it does).
  * Returns null when the object is missing.