From a9aae1b6232490e1a7c3d96d0d9a567c246f6e8e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 08:37:02 -0700 Subject: [PATCH 1/7] fix(tables): sniff CSV delimiter and capture the real header row on import Table CSV import derived the delimiter purely from the file extension, so semicolon-delimited exports (European-locale Excel) parsed as a single column. Header derivation also read Object.keys(rows[0]), so with relax_column_count a ragged/sparse first data row dropped its trailing columns from schema inference. - Add detectCsvDelimiter: trial-parse the head against ,/;/tab/| and pick the candidate with the most columns (tie-break on row-width consistency), quote-aware so separators inside quoted cells don't skew the guess; extension is now only the fallback - Add sniffCsvDelimiterFromStream: memory-bounded (64KB) peek-then-replay for the streaming import paths so multi-GB files sniff without buffering - Capture the true header row via the csv-parse columns callback on every path, fixing dropped columns when the first row is short - Wire into the create route, append/replace route, background runner, client dialog preview, and parseFileRows (copilot) --- .../app/api/table/[tableId]/import/route.ts | 18 ++- apps/sim/app/api/table/import-csv/route.ts | 19 ++- .../import-csv-dialog/import-csv-dialog.tsx | 24 +++- apps/sim/lib/table/csv-delimiter-stream.ts | 72 ++++++++++ apps/sim/lib/table/import-runner.ts | 17 ++- apps/sim/lib/table/import.test.ts | 129 ++++++++++++++++- apps/sim/lib/table/import.ts | 130 +++++++++++++++--- 7 files changed, 369 insertions(+), 40 deletions(-) create mode 100644 apps/sim/lib/table/csv-delimiter-stream.ts diff --git a/apps/sim/app/api/table/[tableId]/import/route.ts b/apps/sim/app/api/table/[tableId]/import/route.ts index 63ed87220fb..1d6482390fd 100644 --- a/apps/sim/app/api/table/[tableId]/import/route.ts +++ b/apps/sim/app/api/table/[tableId]/import/route.ts @@ -36,6 +36,7 @@ import { validateMapping, wouldExceedRowLimit, } from '@/lib/table' +import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { importAppendRows, importReplaceRows } from '@/lib/table/import-data' import { getUserSettings } from '@/lib/users/queries' import { @@ -176,11 +177,19 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro timezone = timezoneValidation.data } - const delimiter = extensionValidation.data === 'tsv' ? '\t' : ',' - const parser = createCsvParser(delimiter) + // The extension only picks the fallback — the separator is sniffed from the file's + // head so semicolon/pipe exports (European-locale Excel) don't land in one column. + const { delimiter, stream: csvStream } = await sniffCsvDelimiterFromStream( + file.stream, + extensionValidation.data === 'tsv' ? '\t' : ',' + ) + let headers: string[] = [] + const parser = createCsvParser(delimiter, (parsedHeaders) => { + headers = parsedHeaders + }) // `.pipe` doesn't forward source errors; forward them so the iterator throws. - file.stream.on('error', (streamErr) => parser.destroy(streamErr)) - file.stream.pipe(parser) + csvStream.on('error', (streamErr) => parser.destroy(streamErr)) + csvStream.pipe(parser) const rows: Record[] = [] for await (const record of parser as AsyncIterable>) { rows.push(record) @@ -188,7 +197,6 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro if (rows.length === 0) { return NextResponse.json({ error: 'CSV file has no data rows' }, { status: 400 }) } - const headers = Object.keys(rows[0]) let effectiveMapping = mapping ?? buildAutoMapping(headers, table.schema) let prospectiveTable: TableDefinition = table diff --git a/apps/sim/app/api/table/import-csv/route.ts b/apps/sim/app/api/table/import-csv/route.ts index fb29cffab8f..2c0a3cf6614 100644 --- a/apps/sim/app/api/table/import-csv/route.ts +++ b/apps/sim/app/api/table/import-csv/route.ts @@ -26,6 +26,7 @@ import { type TableDefinition, type TableSchema, } from '@/lib/table' +import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { getUserSettings } from '@/lib/users/queries' import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils' import { @@ -107,12 +108,20 @@ export const POST = withRouteHandler(async (request: NextRequest) => { { status: 400 } ) } - const delimiter = extensionResult.data === 'tsv' ? '\t' : ',' + // The extension only picks the fallback — the separator is sniffed from the file's + // head so semicolon/pipe exports (European-locale Excel) don't land in one column. + const { delimiter, stream: csvStream } = await sniffCsvDelimiterFromStream( + file.stream, + extensionResult.data === 'tsv' ? '\t' : ',' + ) - const parser = createCsvParser(delimiter) + let csvHeaders: string[] = [] + const parser = createCsvParser(delimiter, (headers) => { + csvHeaders = headers + }) // `.pipe` doesn't forward source errors; forward them so the iterator throws. - file.stream.on('error', (err) => parser.destroy(err)) - file.stream.pipe(parser) + csvStream.on('error', (err) => parser.destroy(err)) + csvStream.pipe(parser) interface ImportState { table: TableDefinition @@ -139,7 +148,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => { /** Infer the schema from the buffered sample and create the (empty) table. */ const buildTable = async (sampleRows: Record[]): Promise => { - const inferred = inferSchemaFromCsv(Object.keys(sampleRows[0]), sampleRows) + const inferred = inferSchemaFromCsv(csvHeaders, sampleRows) const schema: TableSchema = { columns: inferred.columns.map(normalizeColumn) } const planLimits = await getWorkspaceTableLimits(workspaceId) const tableName = sanitizeName(file.filename.replace(/\.[^.]+$/, ''), 'imported_table').slice( diff --git a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx index 2eee7a1c4ee..7b1097f7696 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx @@ -25,7 +25,13 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { truncate } from '@sim/utils/string' import { CSV_ASYNC_IMPORT_THRESHOLD_BYTES } from '@/lib/table/constants' -import { buildAutoMapping, parseCsvBuffer } from '@/lib/table/import' +import { + buildAutoMapping, + CSV_DELIMITER_SNIFF_BYTES, + type CsvDelimiter, + detectCsvDelimiter, + parseCsvBuffer, +} from '@/lib/table/import' import type { TableDefinition } from '@/lib/table/types' import { type CsvImportMode, @@ -107,8 +113,13 @@ interface ParsedCsv { sampleRows: Record[] } -/** Parses the head of a CSV/TSV for the mapping + sample, dropping any truncated final line. */ -async function parseCsvPreview(file: File, delimiter: ',' | '\t') { +/** + * Parses the head of a CSV/TSV for the mapping + sample, dropping any truncated final line. + * + * The separator is sniffed from the same leading bytes the server sniffs, so the mapping shown + * here always matches the columns the import will actually produce. + */ +async function parseCsvPreview(file: File, fallbackDelimiter: CsvDelimiter) { const sliced = file.size > CSV_PREVIEW_BYTES const blob = sliced ? file.slice(0, CSV_PREVIEW_BYTES) : file let bytes = new Uint8Array(await blob.arrayBuffer()) @@ -116,6 +127,10 @@ async function parseCsvPreview(file: File, delimiter: ',' | '\t') { const lastNewline = bytes.lastIndexOf(0x0a) if (lastNewline > 0) bytes = bytes.subarray(0, lastNewline + 1) } + const delimiter = await detectCsvDelimiter( + bytes.subarray(0, CSV_DELIMITER_SNIFF_BYTES), + fallbackDelimiter + ) return parseCsvBuffer(bytes, delimiter) } @@ -180,8 +195,7 @@ export function ImportCsvDialog({ setParsing(true) setParseError(null) try { - const delimiter: ',' | '\t' = ext === 'tsv' ? '\t' : ',' - const { headers, rows } = await parseCsvPreview(file, delimiter) + const { headers, rows } = await parseCsvPreview(file, ext === 'tsv' ? '\t' : ',') const autoMapping = buildAutoMapping(headers, table.schema) setParsed({ file, diff --git a/apps/sim/lib/table/csv-delimiter-stream.ts b/apps/sim/lib/table/csv-delimiter-stream.ts new file mode 100644 index 00000000000..de1146b8916 --- /dev/null +++ b/apps/sim/lib/table/csv-delimiter-stream.ts @@ -0,0 +1,72 @@ +import { Readable } from 'node:stream' +import { + CSV_DELIMITER_SNIFF_BYTES, + type CsvDelimiter, + detectCsvDelimiter, +} from '@/lib/table/import' + +export interface SniffedCsvStream { + delimiter: CsvDelimiter + /** + * The full file contents, replayed from byte zero. Use this in place of the + * source stream — the source has already been partially consumed. + */ + stream: Readable +} + +/** + * Reads the head of a CSV/TSV stream, sniffs its field separator, then returns a + * stream that replays the buffered head followed by the untouched remainder. + * + * Only {@link CSV_DELIMITER_SNIFF_BYTES} are ever held in memory, so this stays + * safe on multi-GB imports. The buffered head is trimmed to its last newline + * before sniffing so a mid-record cut can't skew the column counts. + */ +export async function sniffCsvDelimiterFromStream( + source: Readable, + fallback: CsvDelimiter = ',' +): Promise { + const reader = source[Symbol.asyncIterator]() + const chunks: Buffer[] = [] + let size = 0 + let exhausted = false + + while (size < CSV_DELIMITER_SNIFF_BYTES) { + const next = await reader.next() + if (next.done) { + exhausted = true + break + } + const chunk = Buffer.isBuffer(next.value) ? next.value : Buffer.from(next.value as Uint8Array) + chunks.push(chunk) + size += chunk.length + } + + const head = Buffer.concat(chunks) + let sample = head + if (!exhausted) { + const lastNewline = head.lastIndexOf(0x0a) + if (lastNewline > 0) sample = head.subarray(0, lastNewline + 1) + } + + const delimiter = await detectCsvDelimiter(sample, fallback) + + const stream = Readable.from( + (async function* replay() { + if (head.length > 0) yield head + if (exhausted) return + while (true) { + const next = await reader.next() + if (next.done) return + yield next.value + } + })() + ) + + // `Readable.from` closes the generator on destroy, which returns the source + // iterator — but an early destroy of the wrapper before the generator is + // pulled would otherwise leak the source's socket. + stream.on('close', () => source.destroy()) + + return { delimiter, stream } +} diff --git a/apps/sim/lib/table/import-runner.ts b/apps/sim/lib/table/import-runner.ts index 64694f97bf3..85f85875577 100644 --- a/apps/sim/lib/table/import-runner.ts +++ b/apps/sim/lib/table/import-runner.ts @@ -19,6 +19,7 @@ import { } from '@/lib/table' import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing' import { withGeneratedColumnIds } from '@/lib/table/column-keys' +import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { appendTableEvent } from '@/lib/table/events' import { addImportColumns, @@ -103,6 +104,11 @@ export async function runTableImport(payload: TableImportPayload): Promise // Stream the file rather than buffering it — a ~1M-row import must never be held in memory. source = await downloadFileStream({ key: fileKey, context: 'workspace' }) + // The kickoff route's extension-derived delimiter is only the fallback — the separator is + // sniffed from the file's head so semicolon/pipe exports don't collapse into one column. + const sniffed = await sniffCsvDelimiterFromStream(source, delimiter) + const csvStream = sniffed.stream + // Append must continue after the existing rows; create/replace start empty. Read once up // front (the import is the table's sole writer) and assign contiguous positions / threaded // order keys from it. @@ -123,11 +129,14 @@ export async function runTableImport(payload: TableImportPayload): Promise }, }) - const parser = createCsvParser(delimiter) + let csvHeaders: string[] = [] + const parser = createCsvParser(sniffed.delimiter, (headers) => { + csvHeaders = headers + }) // `.pipe` doesn't forward source errors; forward so the iterator throws. - source.on('error', (err) => parser.destroy(err)) + csvStream.on('error', (err) => parser.destroy(err)) byteCounter.on('error', (err) => parser.destroy(err)) - source.pipe(byteCounter).pipe(parser) + csvStream.pipe(byteCounter).pipe(parser) let schema: TableSchema | null = null let headerToColumn: Map | null = null @@ -142,7 +151,7 @@ export async function runTableImport(payload: TableImportPayload): Promise * map onto the existing schema, optionally auto-creating `createColumns` first. */ const resolveSetup = async () => { - const headers = Object.keys(sample[0]) + const headers = csvHeaders if (mode === 'create') { const inferred = inferSchemaFromCsv(headers, sample) diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index d4997a123fe..4bd7e5959bc 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -3,6 +3,7 @@ */ import { Readable } from 'node:stream' import { describe, expect, it } from 'vitest' +import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { buildAutoMapping, CsvImportValidationError, @@ -10,6 +11,7 @@ import { coerceValue, createCsvParser, csvParseOptions, + detectCsvDelimiter, inferColumnType, inferSchemaFromCsv, parseCsvBuffer, @@ -316,8 +318,131 @@ describe('import', () => { }) describe('csvParseOptions', () => { - it('sets columns, bom, and the delimiter', () => { - expect(csvParseOptions('\t')).toMatchObject({ columns: true, bom: true, delimiter: '\t' }) + it('sets bom and the delimiter, with a header-capturing columns callback', () => { + const options = csvParseOptions('\t') + expect(options).toMatchObject({ bom: true, delimiter: '\t' }) + expect(typeof options.columns).toBe('function') + }) + + it('invokes onHeaders with the full header row', () => { + let captured: string[] | null = null + const options = csvParseOptions(',', (h) => { + captured = h + }) + // csv-parse calls the columns function with the parsed header array. + ;(options.columns as (h: string[]) => string[])(['a', 'b', 'c']) + expect(captured).toEqual(['a', 'b', 'c']) + }) + }) + + describe('parseCsvBuffer header derivation', () => { + it('reports every header even when the first data row is ragged (short)', async () => { + // With relax_column_count a short first row omits trailing keys, so deriving + // headers from Object.keys(rows[0]) would drop c and d. The header callback fixes this. + const { headers } = await parseCsvBuffer('a,b,c,d\n1,2\n3,4,5,6\n') + expect(headers).toEqual(['a', 'b', 'c', 'd']) + }) + + it('feeds the full header set into inferSchemaFromCsv for a ragged file', async () => { + const { headers, rows } = await parseCsvBuffer('a,b,c\n1\n2,3,4\n') + const { columns } = inferSchemaFromCsv(headers, rows) + expect(columns.map((c) => c.name)).toEqual(['a', 'b', 'c']) + }) + }) + + describe('detectCsvDelimiter', () => { + it('detects a comma-delimited file', async () => { + expect(await detectCsvDelimiter('a,b,c\n1,2,3\n')).toBe(',') + }) + + it('detects a semicolon-delimited file (European Excel export)', async () => { + expect(await detectCsvDelimiter('a;b;c\n1;2;3\n')).toBe(';') + }) + + it('detects tab and pipe delimiters', async () => { + expect(await detectCsvDelimiter('a\tb\tc\n1\t2\t3\n')).toBe('\t') + expect(await detectCsvDelimiter('a|b|c\n1|2|3\n')).toBe('|') + }) + + it('ignores delimiters that appear only inside quoted fields', async () => { + // Semicolon-separated, but the values are full of commas and newlines — a raw + // character-frequency count would wrongly pick the comma. A real parse does not. + const csv = 'id;body\n1;"hello, world\nsecond line"\n2;"a, b, c"\n' + expect(await detectCsvDelimiter(csv)).toBe(';') + }) + + it('falls back for a single-column file rather than latching onto in-value characters', async () => { + expect(await detectCsvDelimiter('text\n"hello, world"\n"a, b"\n')).toBe(',') + expect(await detectCsvDelimiter('text\n"hello, world"\n', ';')).toBe(';') + }) + + it('returns the fallback for empty input', async () => { + expect(await detectCsvDelimiter('', ';')).toBe(';') + }) + }) + + describe('sniffCsvDelimiterFromStream', () => { + async function collect(stream: Readable): Promise { + const chunks: Buffer[] = [] + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array)) + } + return Buffer.concat(chunks) + } + + it('detects the delimiter and replays the full file byte-for-byte', async () => { + const rows = ['id;a;b;c'] + for (let i = 0; i < 5000; i++) rows.push(`${i};"val, with comma";x;y`) + const full = Buffer.from(`${rows.join('\n')}\n`) + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from([full])) + expect(delimiter).toBe(';') + expect((await collect(stream)).equals(full)).toBe(true) + }) + + it('handles a file smaller than the sniff window (exhausted during sniff)', async () => { + const full = Buffer.from('a;b;c\n1;2;3\n') + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from([full])) + expect(delimiter).toBe(';') + expect((await collect(stream)).equals(full)).toBe(true) + }) + + it('reassembles data split across many small chunks', async () => { + const parts = ['na', 'me,ag', 'e\nfo', 'o,1\nba', 'r,2\n'].map((s) => Buffer.from(s)) + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from(parts)) + expect(delimiter).toBe(',') + expect((await collect(stream)).equals(Buffer.concat(parts))).toBe(true) + }) + + it('destroys the source when the replay stream is destroyed early', async () => { + let destroyed = false + const pad = 'z'.repeat(290) + async function* infinite() { + let i = 0 + while (true) yield Buffer.from(`r${i++};x;${pad}\n`) + } + const source = Readable.from(infinite()) + source.on('close', () => { + destroyed = true + }) + const { stream } = await sniffCsvDelimiterFromStream(source) + stream.destroy() + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(destroyed).toBe(true) + }) + + it('surfaces a source error that occurs during replay', async () => { + async function* failing() { + yield Buffer.from(`a;b;c\n${'1;2;3\n'.repeat(20000)}`) + throw new Error('storage read failed') + } + const { stream } = await sniffCsvDelimiterFromStream(Readable.from(failing())) + await expect(collect(stream)).rejects.toThrow(/storage read failed/) + }) + + it('returns the fallback for an empty stream', async () => { + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from([]), ';') + expect(delimiter).toBe(';') + expect((await collect(stream)).length).toBe(0) }) }) }) diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts index 104186e32e1..4e63ad07550 100644 --- a/apps/sim/lib/table/import.ts +++ b/apps/sim/lib/table/import.ts @@ -16,14 +16,39 @@ import { getColumnId } from '@/lib/table/column-keys' import { type NormalizeDateCellOptions, normalizeDateCellValue } from '@/lib/table/dates' import type { ColumnDefinition, RowData, TableSchema } from '@/lib/table/types' +/** + * Field separators we sniff for, in tie-break priority order. Semicolon files are + * the standard CSV export of European-locale Excel; pipe shows up in log exports. + */ +export const CSV_DELIMITER_CANDIDATES = [',', ';', '\t', '|'] as const + +export type CsvDelimiter = (typeof CSV_DELIMITER_CANDIDATES)[number] + +/** + * Bytes inspected when sniffing the delimiter. Read from the head of the file on + * every path (client preview, streamed upload, background worker) so all of them + * observe the same prefix and therefore agree on the result. + */ +export const CSV_DELIMITER_SNIFF_BYTES = 64 * 1024 + /** * Single source of truth for the `csv-parse` options used by both the buffered - * sync parser and the streaming parser. `columns: true` emits each record as an - * object keyed by the (first-row) headers. + * sync parser and the streaming parser. + * + * `columns` is a function rather than `true` so the *actual header row* is + * captured. With `relax_column_count`, a record shorter than the header simply + * omits the trailing keys, so `Object.keys(records[0])` under-reports the schema + * whenever the first data row is ragged — the header callback is authoritative. */ -export function csvParseOptions(delimiter = ','): CsvParseOptions { +export function csvParseOptions( + delimiter = ',', + onHeaders?: (headers: string[]) => void +): CsvParseOptions { return { - columns: true, + columns: (header: string[]) => { + onHeaders?.(header) + return header + }, skip_empty_lines: true, trim: true, relax_column_count: true, @@ -40,9 +65,75 @@ export function csvParseOptions(delimiter = ','): CsvParseOptions { * file stream into it and iterate records with `for await`; backpressure flows * back to the source while each record is processed. Use this for HTTP uploads * so the file is never fully buffered in memory. + * + * `onHeaders` fires once, before the first record, with the full header row. + */ +export function createCsvParser(delimiter = ',', onHeaders?: (headers: string[]) => void): Parser { + return parseCsvStream(csvParseOptions(delimiter, onHeaders)) +} + +/** Decodes CSV bytes as UTF-8, passing strings through unchanged. */ +export function decodeCsvText(input: Buffer | Uint8Array | string): string { + if (typeof input === 'string') return input + if (typeof Buffer !== 'undefined' && Buffer.isBuffer(input)) return input.toString('utf-8') + return new TextDecoder('utf-8').decode(input as Uint8Array) +} + +/** + * Sniffs the field separator by trial-parsing the sample with each candidate and + * keeping the one that yields the most columns. + * + * A real parse (rather than counting raw characters) is what makes this safe on + * files whose quoted cells contain other candidates — `alarms.csv` exports a + * semicolon-separated file whose `raw_text` cells are full of commas and + * newlines, and a naive frequency count picks the comma. Ties are broken by how + * consistently the records match the header width, then by candidate order, so a + * genuinely single-column file falls back to `fallback` rather than latching onto + * a separator that appears inside its values. */ -export function createCsvParser(delimiter = ','): Parser { - return parseCsvStream(csvParseOptions(delimiter)) +export async function detectCsvDelimiter( + input: Buffer | Uint8Array | string, + fallback: CsvDelimiter = ',' +): Promise { + const { parse } = await import('csv-parse/sync') + const text = decodeCsvText(input) + if (text.trim() === '') return fallback + + let best: { delimiter: CsvDelimiter; fields: number; consistency: number } | null = null + + for (const delimiter of CSV_DELIMITER_CANDIDATES) { + let records: string[][] + try { + records = parse(text, { + columns: false, + skip_empty_lines: true, + relax_column_count: true, + relax_quotes: true, + skip_records_with_error: true, + bom: true, + delimiter, + }) as string[][] + } catch { + continue + } + + if (records.length === 0) continue + const fields = records[0].length + if (fields < 2) continue + + const matching = records.reduce((n, r) => (r.length === fields ? n + 1 : n), 0) + const consistency = matching / records.length + + if ( + !best || + fields > best.fields || + (fields === best.fields && consistency > best.consistency) + ) { + best = { delimiter, fields, consistency } + } + } + + return best?.delimiter ?? fallback } /** Narrower type than `COLUMN_TYPES` used internally for coercion. */ @@ -103,24 +194,22 @@ export async function parseCsvBuffer( ): Promise<{ headers: string[]; rows: Record[] }> { const { parse } = await import('csv-parse/sync') - let text: string - if (typeof input === 'string') { - text = input - } else if (typeof Buffer !== 'undefined' && Buffer.isBuffer(input)) { - text = input.toString('utf-8') - } else { - text = new TextDecoder('utf-8').decode(input as Uint8Array) - } + const text = decodeCsvText(input) - // double-cast-allowed: shared csvParseOptions() loses the `columns: true` literal that drives - // csv-parse's record-vs-string[][] overload, but `columns: true` is always set so records are objects - const parsed = parse(text, csvParseOptions(delimiter)) as unknown as Record[] + let headers: string[] = [] + // double-cast-allowed: shared csvParseOptions() loses the `columns` literal that drives + // csv-parse's record-vs-string[][] overload, but `columns` is always set so records are objects + const parsed = parse( + text, + csvParseOptions(delimiter, (h) => { + headers = h + }) + ) as unknown as Record[] if (parsed.length === 0) { throw new Error('CSV file has no data rows') } - const headers = Object.keys(parsed[0]) if (headers.length === 0) { throw new Error('CSV file has no headers') } @@ -504,7 +593,10 @@ export async function parseFileRows( return parseJsonRows(buffer) } if (ext === 'csv' || ext === 'tsv' || contentType === 'text/csv') { - const delimiter = ext === 'tsv' ? '\t' : ',' + const delimiter = await detectCsvDelimiter( + buffer.subarray(0, CSV_DELIMITER_SNIFF_BYTES), + ext === 'tsv' ? '\t' : ',' + ) return parseCsvBuffer(buffer, delimiter) } throw new Error(`Unsupported file format: "${ext ?? fileName}". Supported: csv, tsv, json`) From a6cc15cf3ad992bd711bc810f877e5e21ff478e7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 08:48:52 -0700 Subject: [PATCH 2/7] fix(tables): rank CSV delimiter by row consistency and bound the peek buffer Addresses review findings on the delimiter sniffer: - Rank candidates by row-width consistency first (modal column count), then column count. Ranking on column count alone let a comma that appears in a semicolon file's unquoted header win over the real separator; the wide-but-ragged split now loses to the uniform one. - Move the partial-trailing-line trim into detectCsvDelimiter so every path (stream, client preview, parseFileRows) prepares the sniff sample identically and can't disagree on the delimiter. - Cap the stream peeker's detection copy at the sniff window via Buffer.concat's length arg and replay the buffered chunks by reference, so an oversized source chunk can't break the bounded-memory contract. --- apps/sim/lib/table/csv-delimiter-stream.ts | 23 +++++----- apps/sim/lib/table/import.test.ts | 29 +++++++++++++ apps/sim/lib/table/import.ts | 49 +++++++++++++++++----- 3 files changed, 79 insertions(+), 22 deletions(-) diff --git a/apps/sim/lib/table/csv-delimiter-stream.ts b/apps/sim/lib/table/csv-delimiter-stream.ts index de1146b8916..b9baa573c9e 100644 --- a/apps/sim/lib/table/csv-delimiter-stream.ts +++ b/apps/sim/lib/table/csv-delimiter-stream.ts @@ -18,9 +18,13 @@ export interface SniffedCsvStream { * Reads the head of a CSV/TSV stream, sniffs its field separator, then returns a * stream that replays the buffered head followed by the untouched remainder. * - * Only {@link CSV_DELIMITER_SNIFF_BYTES} are ever held in memory, so this stays - * safe on multi-GB imports. The buffered head is trimmed to its last newline - * before sniffing so a mid-record cut can't skew the column counts. + * Memory is bounded: it buffers only the source chunks needed to reach the + * {@link CSV_DELIMITER_SNIFF_BYTES} window (one chunk past it, at worst), and the + * detection sample it copies is capped at exactly that window regardless of how + * large a single upstream chunk is. Those buffered chunks are then replayed + * *by reference* — never re-copied — so a multi-GB import stays O(sniff window) + * plus the single in-flight chunk. {@link detectCsvDelimiter} drops any partial + * trailing line, so no newline trimming is needed here. */ export async function sniffCsvDelimiterFromStream( source: Readable, @@ -42,18 +46,15 @@ export async function sniffCsvDelimiterFromStream( size += chunk.length } - const head = Buffer.concat(chunks) - let sample = head - if (!exhausted) { - const lastNewline = head.lastIndexOf(0x0a) - if (lastNewline > 0) sample = head.subarray(0, lastNewline + 1) - } - + // Copy at most the sniff window for detection — `Buffer.concat`'s length arg truncates, + // so an oversized final chunk can't inflate this allocation past CSV_DELIMITER_SNIFF_BYTES. + const sample = Buffer.concat(chunks, Math.min(size, CSV_DELIMITER_SNIFF_BYTES)) const delimiter = await detectCsvDelimiter(sample, fallback) const stream = Readable.from( (async function* replay() { - if (head.length > 0) yield head + // Replay the already-read chunks by reference (no re-copy), then drain the rest. + for (const chunk of chunks) yield chunk if (exhausted) return while (true) { const next = await reader.next() diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index 4bd7e5959bc..d6d00045363 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -376,6 +376,23 @@ describe('import', () => { expect(await detectCsvDelimiter('text\n"hello, world"\n', ';')).toBe(';') }) + it('prefers the consistent delimiter over one that only widens the header row', async () => { + // The header splits into more fields on comma, but only semicolon yields a uniform + // column count across the data rows — consistency must win over raw field count. + expect(await detectCsvDelimiter('a,b;c,d;e,f\n1;2;3\n4;5;6\n')).toBe(';') + }) + + it('detects the real delimiter when the header itself has quoted commas', async () => { + const csv = '"last, first";age;city\n"Doe, John";30;NYC\n"Roe, Jane";25;LA\n' + expect(await detectCsvDelimiter(csv)).toBe(';') + }) + + it('drops a partial trailing line so a mid-record byte cut cannot skew detection', async () => { + // Simulates a fixed byte-window slice that ends mid-record; the last (partial) line + // is ignored, so the complete rows above still drive the result. + expect(await detectCsvDelimiter('a;b;c\n1;2;3\n4;5;')).toBe(';') + }) + it('returns the fallback for empty input', async () => { expect(await detectCsvDelimiter('', ';')).toBe(';') }) @@ -406,6 +423,18 @@ describe('import', () => { expect((await collect(stream)).equals(full)).toBe(true) }) + it('detects and replays exactly when a single chunk far exceeds the sniff window', async () => { + // One >1 MiB chunk arrives before the size check — detection must still cap its copy + // and the replay must emit the original bytes unchanged. + const rows = ['a;b;c'] + for (let i = 0; i < 40000; i++) rows.push(`${i};${'x'.repeat(20)};y`) + const full = Buffer.from(`${rows.join('\n')}\n`) + expect(full.length).toBeGreaterThan(1024 * 1024) + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from([full])) + expect(delimiter).toBe(';') + expect((await collect(stream)).equals(full)).toBe(true) + }) + it('reassembles data split across many small chunks', async () => { const parts = ['na', 'me,ag', 'e\nfo', 'o,1\nba', 'r,2\n'].map((s) => Buffer.from(s)) const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from(parts)) diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts index 4e63ad07550..f6016b4970b 100644 --- a/apps/sim/lib/table/import.ts +++ b/apps/sim/lib/table/import.ts @@ -81,22 +81,37 @@ export function decodeCsvText(input: Buffer | Uint8Array | string): string { /** * Sniffs the field separator by trial-parsing the sample with each candidate and - * keeping the one that yields the most columns. + * keeping the one whose column count is most *consistent* across rows. * * A real parse (rather than counting raw characters) is what makes this safe on * files whose quoted cells contain other candidates — `alarms.csv` exports a * semicolon-separated file whose `raw_text` cells are full of commas and - * newlines, and a naive frequency count picks the comma. Ties are broken by how - * consistently the records match the header width, then by candidate order, so a - * genuinely single-column file falls back to `fallback` rather than latching onto - * a separator that appears inside its values. + * newlines, and a naive frequency count picks the comma. + * + * Ranking is **consistency first, then column count**. The true delimiter yields + * a uniform column count across every row; a spurious one (e.g. a comma that + * happens to appear in an unquoted header of a semicolon file) produces a wide + * first row but ragged data rows. Ranking on column count alone would wrongly + * prefer that wide-but-inconsistent split, so consistency dominates and column + * count only breaks ties. The per-candidate column count is the *modal* (most + * common) row width, so one stray ragged row doesn't distort it. A single-column + * file (no candidate reaches two columns) falls back to `fallback`. + * + * All callers funnel through here, so the sample is prepared identically for + * every import path: a possibly-partial trailing line (from slicing a fixed byte + * window out of a larger file) is dropped before parsing so a mid-record cut + * can't skew the counts. */ export async function detectCsvDelimiter( input: Buffer | Uint8Array | string, fallback: CsvDelimiter = ',' ): Promise { const { parse } = await import('csv-parse/sync') - const text = decodeCsvText(input) + const decoded = decodeCsvText(input) + // Drop a partial final line when the sample is a prefix of a larger file; keep the + // whole thing when it's a single line (no newline) so a tiny full file still parses. + const lastNewline = decoded.lastIndexOf('\n') + const text = lastNewline > 0 ? decoded.slice(0, lastNewline + 1) : decoded if (text.trim() === '') return fallback let best: { delimiter: CsvDelimiter; fields: number; consistency: number } | null = null @@ -118,16 +133,28 @@ export async function detectCsvDelimiter( } if (records.length === 0) continue - const fields = records[0].length + + // Modal row width: the most frequent column count, tie-broken toward the wider one. + const widthCounts = new Map() + for (const record of records) { + widthCounts.set(record.length, (widthCounts.get(record.length) ?? 0) + 1) + } + let fields = 0 + let modalFreq = 0 + for (const [width, freq] of widthCounts) { + if (freq > modalFreq || (freq === modalFreq && width > fields)) { + fields = width + modalFreq = freq + } + } if (fields < 2) continue - const matching = records.reduce((n, r) => (r.length === fields ? n + 1 : n), 0) - const consistency = matching / records.length + const consistency = modalFreq / records.length if ( !best || - fields > best.fields || - (fields === best.fields && consistency > best.consistency) + consistency > best.consistency || + (consistency === best.consistency && fields > best.fields) ) { best = { delimiter, fields, consistency } } From 33286cfbae6958870a1a913fb9d307bee9721ca3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 09:01:40 -0700 Subject: [PATCH 3/7] fix(tables): score CSV delimiter by width*consistency and dedupe headers Addresses round-2 review findings: - Score each delimiter candidate by modalWidth * consistency instead of consistency alone. Consistency-only let a separator that appears uniformly inside values (e.g. a pipe once per row) beat a real delimiter whose rows are legitimately ragged; the product rewards a split that is both wide and uniform, with width breaking ties. Genuinely symmetric files (two valid columns under either separator) fall back to the global-default candidate order. - Dedupe the captured header row to match the parser's record keys. csv-parse collapses duplicate column names onto one key (last value wins), so reporting the raw duplicates made schema inference invent a phantom empty column and could fail mapping validation. dedupeHeaders keeps first-occurrence order, case-sensitively, matching the emitted keys. --- apps/sim/lib/table/import.test.ts | 33 +++++++++++++++++ apps/sim/lib/table/import.ts | 60 ++++++++++++++++++++++--------- 2 files changed, 76 insertions(+), 17 deletions(-) diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index d6d00045363..beba5ca6b06 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -11,6 +11,7 @@ import { coerceValue, createCsvParser, csvParseOptions, + dedupeHeaders, detectCsvDelimiter, inferColumnType, inferSchemaFromCsv, @@ -348,6 +349,26 @@ describe('import', () => { const { columns } = inferSchemaFromCsv(headers, rows) expect(columns.map((c) => c.name)).toEqual(['a', 'b', 'c']) }) + + it('collapses duplicate header names to match the parser record keys', async () => { + // csv-parse keys duplicate columns onto one object key (last value wins); the reported + // headers must match, so schema inference does not invent a phantom empty column. + const { headers, rows } = await parseCsvBuffer('a,a,b\n1,2,3\n4,5,6\n') + expect(headers).toEqual(['a', 'b']) + expect(Object.keys(rows[0])).toEqual(['a', 'b']) + const { columns } = inferSchemaFromCsv(headers, rows) + expect(columns.map((c) => c.name)).toEqual(['a', 'b']) + }) + }) + + describe('dedupeHeaders', () => { + it('drops later exact duplicates, preserving first-occurrence order', () => { + expect(dedupeHeaders(['a', 'b', 'a', 'c', 'b'])).toEqual(['a', 'b', 'c']) + }) + + it('keeps case-distinct names (matches csv-parse key sensitivity)', () => { + expect(dedupeHeaders(['Name', 'name'])).toEqual(['Name', 'name']) + }) }) describe('detectCsvDelimiter', () => { @@ -387,6 +408,18 @@ describe('import', () => { expect(await detectCsvDelimiter(csv)).toBe(';') }) + it('prefers a wider ragged split over a narrow one that only looks uniform', async () => { + // Semicolon is the real delimiter (2- and 3-column rows) while a pipe appears once per + // row inside values. Scoring width*consistency keeps the wider semicolon split. + expect(await detectCsvDelimiter('a|label;b;c\nx|y;1\nz|w;2;3\n')).toBe(';') + }) + + it('falls back to comma order only when the split is genuinely ambiguous', async () => { + // Both ; and , yield exactly two uniform columns — no content signal distinguishes them, + // so the global-default candidate order (comma first) decides and either reading is valid. + expect(await detectCsvDelimiter('name;value,unit\na;1,kg\nb;2,kg\n')).toBe(',') + }) + it('drops a partial trailing line so a mid-record byte cut cannot skew detection', async () => { // Simulates a fixed byte-window slice that ends mid-record; the last (partial) line // is ignored, so the complete rows above still drive the result. diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts index f6016b4970b..cde23146ccc 100644 --- a/apps/sim/lib/table/import.ts +++ b/apps/sim/lib/table/import.ts @@ -46,7 +46,12 @@ export function csvParseOptions( ): CsvParseOptions { return { columns: (header: string[]) => { - onHeaders?.(header) + // Deliver headers deduped to match the record keys: csv-parse collapses duplicate + // column names into a single object key (last value wins), so the consumer's schema + // inference and mapping must see the same unique set — not the raw duplicates, which + // would invent phantom columns that never receive a value. csv-parse still keys on the + // raw array we return here, so returning it unchanged preserves that collapsing. + onHeaders?.(dedupeHeaders(header)) return header }, skip_empty_lines: true, @@ -72,6 +77,22 @@ export function createCsvParser(delimiter = ',', onHeaders?: (headers: string[]) return parseCsvStream(csvParseOptions(delimiter, onHeaders)) } +/** + * Drops later exact-duplicate header names, preserving first-occurrence order. Mirrors how + * `csv-parse` collapses duplicate column names into a single record key (last value wins), so + * the header set stays in lockstep with the object keys the parser actually emits. + */ +export function dedupeHeaders(headers: string[]): string[] { + const seen = new Set() + const unique: string[] = [] + for (const header of headers) { + if (seen.has(header)) continue + seen.add(header) + unique.push(header) + } + return unique +} + /** Decodes CSV bytes as UTF-8, passing strings through unchanged. */ export function decodeCsvText(input: Buffer | Uint8Array | string): string { if (typeof input === 'string') return input @@ -88,14 +109,22 @@ export function decodeCsvText(input: Buffer | Uint8Array | string): string { * semicolon-separated file whose `raw_text` cells are full of commas and * newlines, and a naive frequency count picks the comma. * - * Ranking is **consistency first, then column count**. The true delimiter yields - * a uniform column count across every row; a spurious one (e.g. a comma that - * happens to appear in an unquoted header of a semicolon file) produces a wide - * first row but ragged data rows. Ranking on column count alone would wrongly - * prefer that wide-but-inconsistent split, so consistency dominates and column - * count only breaks ties. The per-candidate column count is the *modal* (most - * common) row width, so one stray ragged row doesn't distort it. A single-column - * file (no candidate reaches two columns) falls back to `fallback`. + * Each candidate is scored `modalWidth × consistency` — the modal (most common) + * row width times the fraction of rows at that width. This balances the two + * failure modes: ranking on width alone lets a delimiter that merely widens the + * header win (a comma splitting a semicolon file's unquoted header), while + * ranking on consistency alone lets a separator that appears uniformly *inside* + * values win over a real delimiter whose rows are legitimately ragged (a pipe + * embedded once per row beating a semicolon that yields 2- and 3-column rows). + * The product rewards a split that is both wide and uniform; ties break toward + * the wider split, then toward candidate order. Using the modal width (not the + * first row's) keeps one stray ragged row from distorting the score, and a + * single-column file (no candidate reaches two columns) falls back to `fallback`. + * + * Files that stay exactly tied on both score and width are genuinely ambiguous — + * e.g. `name;value,unit` splits into two columns under either `;` or `,` — so the + * global-default candidate order (comma first) decides, and both readings still + * produce a valid table. * * All callers funnel through here, so the sample is prepared identically for * every import path: a possibly-partial trailing line (from slicing a fixed byte @@ -114,7 +143,7 @@ export async function detectCsvDelimiter( const text = lastNewline > 0 ? decoded.slice(0, lastNewline + 1) : decoded if (text.trim() === '') return fallback - let best: { delimiter: CsvDelimiter; fields: number; consistency: number } | null = null + let best: { delimiter: CsvDelimiter; fields: number; score: number } | null = null for (const delimiter of CSV_DELIMITER_CANDIDATES) { let records: string[][] @@ -149,14 +178,11 @@ export async function detectCsvDelimiter( } if (fields < 2) continue - const consistency = modalFreq / records.length + // Reward a split that is both wide and uniform; ties break toward the wider split. + const score = fields * (modalFreq / records.length) - if ( - !best || - consistency > best.consistency || - (consistency === best.consistency && fields > best.fields) - ) { - best = { delimiter, fields, consistency } + if (!best || score > best.score || (score === best.score && fields > best.fields)) { + best = { delimiter, fields, score } } } From 3811a5d752e66f788fbb7897f8fc7d90f73af60b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 10:38:03 -0700 Subject: [PATCH 4/7] fix(tables): keep the final row when sniffing a complete CSV MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit detectCsvDelimiter always trimmed to the last newline, so a complete file with no trailing newline lost its final data row from scoring — leaving the header alone and letting a header-widening separator beat the real one. It now takes a `complete` flag: the trim runs only for truncated prefixes, where a mid-record cut must be dropped. The stream sniffer passes it from `exhausted`, and the buffered callers (client preview, parseFileRows) pass it when the sample covers the whole file. --- .../import-csv-dialog/import-csv-dialog.tsx | 5 ++++- apps/sim/lib/table/csv-delimiter-stream.ts | 8 +++++-- apps/sim/lib/table/import.test.ts | 18 ++++++++++++++++ apps/sim/lib/table/import.ts | 21 ++++++++++++------- 4 files changed, 41 insertions(+), 11 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx index 7b1097f7696..6ea241ab22f 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx @@ -127,9 +127,12 @@ async function parseCsvPreview(file: File, fallbackDelimiter: CsvDelimiter) { const lastNewline = bytes.lastIndexOf(0x0a) if (lastNewline > 0) bytes = bytes.subarray(0, lastNewline + 1) } + // The sniff sample is the whole file only when nothing was sliced off and it fits the window; + // otherwise it's a truncated prefix whose last line may be partial. const delimiter = await detectCsvDelimiter( bytes.subarray(0, CSV_DELIMITER_SNIFF_BYTES), - fallbackDelimiter + fallbackDelimiter, + { complete: !sliced && bytes.length <= CSV_DELIMITER_SNIFF_BYTES } ) return parseCsvBuffer(bytes, delimiter) } diff --git a/apps/sim/lib/table/csv-delimiter-stream.ts b/apps/sim/lib/table/csv-delimiter-stream.ts index b9baa573c9e..fa3b5491957 100644 --- a/apps/sim/lib/table/csv-delimiter-stream.ts +++ b/apps/sim/lib/table/csv-delimiter-stream.ts @@ -48,8 +48,12 @@ export async function sniffCsvDelimiterFromStream( // Copy at most the sniff window for detection — `Buffer.concat`'s length arg truncates, // so an oversized final chunk can't inflate this allocation past CSV_DELIMITER_SNIFF_BYTES. - const sample = Buffer.concat(chunks, Math.min(size, CSV_DELIMITER_SNIFF_BYTES)) - const delimiter = await detectCsvDelimiter(sample, fallback) + // `exhausted` means the whole file fit in the window, so its last row must count even + // without a trailing newline; otherwise the sample is a prefix and its tail may be partial. + const capped = Math.min(size, CSV_DELIMITER_SNIFF_BYTES) + const sample = Buffer.concat(chunks, capped) + const complete = exhausted && size <= CSV_DELIMITER_SNIFF_BYTES + const delimiter = await detectCsvDelimiter(sample, fallback, { complete }) const stream = Readable.from( (async function* replay() { diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index beba5ca6b06..076314043f5 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -426,6 +426,15 @@ describe('import', () => { expect(await detectCsvDelimiter('a;b;c\n1;2;3\n4;5;')).toBe(';') }) + it('keeps the final row of a complete file that has no trailing newline', async () => { + // Without complete:true the trim would drop the only distinguishing data row, leaving the + // header alone and letting comma (which merely widens the header) win over semicolon. + const csv = 'a,b;c,d;e,f\n1;2;3' + expect(await detectCsvDelimiter(csv, ',', { complete: true })).toBe(';') + // A truncated prefix (complete:false, the default) still drops the partial tail. + expect(await detectCsvDelimiter(csv)).toBe(',') + }) + it('returns the fallback for empty input', async () => { expect(await detectCsvDelimiter('', ';')).toBe(';') }) @@ -456,6 +465,15 @@ describe('import', () => { expect((await collect(stream)).equals(full)).toBe(true) }) + it('counts the final row of a fully-buffered file with no trailing newline', async () => { + // The whole file fits the sniff window (exhausted), so complete:true flows through and the + // last row is scored — semicolon must win despite comma widening the header row. + const full = Buffer.from('a,b;c,d;e,f\n1;2;3') + const { delimiter, stream } = await sniffCsvDelimiterFromStream(Readable.from([full])) + expect(delimiter).toBe(';') + expect((await collect(stream)).equals(full)).toBe(true) + }) + it('detects and replays exactly when a single chunk far exceeds the sniff window', async () => { // One >1 MiB chunk arrives before the size check — detection must still cap its copy // and the replay must emit the original bytes unchanged. diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts index cde23146ccc..cae20c4875d 100644 --- a/apps/sim/lib/table/import.ts +++ b/apps/sim/lib/table/import.ts @@ -127,20 +127,24 @@ export function decodeCsvText(input: Buffer | Uint8Array | string): string { * produce a valid table. * * All callers funnel through here, so the sample is prepared identically for - * every import path: a possibly-partial trailing line (from slicing a fixed byte - * window out of a larger file) is dropped before parsing so a mid-record cut - * can't skew the counts. + * every import path: when the sample is a prefix sliced out of a larger file, a + * possibly-partial trailing line is dropped before parsing so a mid-record cut + * can't skew the counts. Pass `complete: true` when the sample is the entire file + * (nothing was sliced off) so a final row with no trailing newline still counts — + * dropping it would silently discard the only distinguishing data row of a tiny + * file and let a header-widening separator win. */ export async function detectCsvDelimiter( input: Buffer | Uint8Array | string, - fallback: CsvDelimiter = ',' + fallback: CsvDelimiter = ',', + { complete = false }: { complete?: boolean } = {} ): Promise { const { parse } = await import('csv-parse/sync') const decoded = decodeCsvText(input) - // Drop a partial final line when the sample is a prefix of a larger file; keep the - // whole thing when it's a single line (no newline) so a tiny full file still parses. + // Drop a partial final line only when the sample is a truncated prefix; keep every row when + // the sample is the whole file (or a single line) so a trailing-newline-less last row counts. const lastNewline = decoded.lastIndexOf('\n') - const text = lastNewline > 0 ? decoded.slice(0, lastNewline + 1) : decoded + const text = !complete && lastNewline > 0 ? decoded.slice(0, lastNewline + 1) : decoded if (text.trim() === '') return fallback let best: { delimiter: CsvDelimiter; fields: number; score: number } | null = null @@ -648,7 +652,8 @@ export async function parseFileRows( if (ext === 'csv' || ext === 'tsv' || contentType === 'text/csv') { const delimiter = await detectCsvDelimiter( buffer.subarray(0, CSV_DELIMITER_SNIFF_BYTES), - ext === 'tsv' ? '\t' : ',' + ext === 'tsv' ? '\t' : ',', + { complete: buffer.length <= CSV_DELIMITER_SNIFF_BYTES } ) return parseCsvBuffer(buffer, delimiter) } From 4312df9c936ca9d0ad1e31dd601a827991b9f7ae Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 10:51:05 -0700 Subject: [PATCH 5/7] fix(tables): keep the double-cast annotation adjacent to its cast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hoist the csv-parse options out of the multi-line parse() call so the `// double-cast-allowed` annotation sits directly above the `as unknown as` token — the strict boundary audit only matches an annotation within three lines of the cast. Also simplify the stream sniffer's completeness check to `exhausted` (the loop only ends early on end-of-stream, so it already means the whole file fit the sniff window). --- apps/sim/lib/table/csv-delimiter-stream.ts | 11 +++++------ apps/sim/lib/table/import.ts | 10 ++++------ 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/apps/sim/lib/table/csv-delimiter-stream.ts b/apps/sim/lib/table/csv-delimiter-stream.ts index fa3b5491957..8dfef189e3f 100644 --- a/apps/sim/lib/table/csv-delimiter-stream.ts +++ b/apps/sim/lib/table/csv-delimiter-stream.ts @@ -48,12 +48,11 @@ export async function sniffCsvDelimiterFromStream( // Copy at most the sniff window for detection — `Buffer.concat`'s length arg truncates, // so an oversized final chunk can't inflate this allocation past CSV_DELIMITER_SNIFF_BYTES. - // `exhausted` means the whole file fit in the window, so its last row must count even - // without a trailing newline; otherwise the sample is a prefix and its tail may be partial. - const capped = Math.min(size, CSV_DELIMITER_SNIFF_BYTES) - const sample = Buffer.concat(chunks, capped) - const complete = exhausted && size <= CSV_DELIMITER_SNIFF_BYTES - const delimiter = await detectCsvDelimiter(sample, fallback, { complete }) + const sample = Buffer.concat(chunks, Math.min(size, CSV_DELIMITER_SNIFF_BYTES)) + // `exhausted` (the loop stopped on end-of-stream, never on the size cap) means the whole file + // fit in the window, so its last row must count even without a trailing newline. Otherwise the + // sample is a truncated prefix whose final line may be partial and should be dropped. + const delimiter = await detectCsvDelimiter(sample, fallback, { complete: exhausted }) const stream = Readable.from( (async function* replay() { diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts index cae20c4875d..94b4dc2272d 100644 --- a/apps/sim/lib/table/import.ts +++ b/apps/sim/lib/table/import.ts @@ -254,14 +254,12 @@ export async function parseCsvBuffer( const text = decodeCsvText(input) let headers: string[] = [] + const options = csvParseOptions(delimiter, (h) => { + headers = h + }) // double-cast-allowed: shared csvParseOptions() loses the `columns` literal that drives // csv-parse's record-vs-string[][] overload, but `columns` is always set so records are objects - const parsed = parse( - text, - csvParseOptions(delimiter, (h) => { - headers = h - }) - ) as unknown as Record[] + const parsed = parse(text, options) as unknown as Record[] if (parsed.length === 0) { throw new Error('CSV file has no data rows') From e6422740b0a0669eb785e914d5ea6ffd8000fdc3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 10:54:20 -0700 Subject: [PATCH 6/7] fix(tables): observe EOF at the exact sniff-window boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stream sniffer's read loop stopped as soon as the buffered size reached the sniff window, so a file whose size is exactly the window never triggered the extra read that observes end-of-stream — it was judged a truncated prefix and dropped its final newline-less row, disagreeing with the buffered callers (which mark that size complete). Read while size <= window so the boundary case sees EOF and `exhausted` (hence `complete`) is set correctly. Regression test added. --- apps/sim/lib/table/csv-delimiter-stream.ts | 12 ++++++++---- apps/sim/lib/table/import.test.ts | 16 ++++++++++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/apps/sim/lib/table/csv-delimiter-stream.ts b/apps/sim/lib/table/csv-delimiter-stream.ts index 8dfef189e3f..1c4c5c2eb70 100644 --- a/apps/sim/lib/table/csv-delimiter-stream.ts +++ b/apps/sim/lib/table/csv-delimiter-stream.ts @@ -35,7 +35,11 @@ export async function sniffCsvDelimiterFromStream( let size = 0 let exhausted = false - while (size < CSV_DELIMITER_SNIFF_BYTES) { + // Read until the buffered size *exceeds* the window (not just reaches it) or the stream ends. + // The `<=` is deliberate: a file whose size is exactly the window must still trigger one more + // read so EOF is observed and `exhausted` becomes true — otherwise it would be misjudged as a + // truncated prefix and disagree with the buffered callers (which pass complete for that size). + while (size <= CSV_DELIMITER_SNIFF_BYTES) { const next = await reader.next() if (next.done) { exhausted = true @@ -49,9 +53,9 @@ export async function sniffCsvDelimiterFromStream( // Copy at most the sniff window for detection — `Buffer.concat`'s length arg truncates, // so an oversized final chunk can't inflate this allocation past CSV_DELIMITER_SNIFF_BYTES. const sample = Buffer.concat(chunks, Math.min(size, CSV_DELIMITER_SNIFF_BYTES)) - // `exhausted` (the loop stopped on end-of-stream, never on the size cap) means the whole file - // fit in the window, so its last row must count even without a trailing newline. Otherwise the - // sample is a truncated prefix whose final line may be partial and should be dropped. + // `exhausted` (the loop stopped on end-of-stream, never on exceeding the window) means the whole + // file fit in the window, so its last row must count even without a trailing newline. Otherwise + // the sample is a truncated prefix whose final line may be partial and should be dropped. const delimiter = await detectCsvDelimiter(sample, fallback, { complete: exhausted }) const stream = Readable.from( diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index 076314043f5..de461e37ec3 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -6,6 +6,7 @@ import { describe, expect, it } from 'vitest' import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { buildAutoMapping, + CSV_DELIMITER_SNIFF_BYTES, CsvImportValidationError, coerceRowsForTable, coerceValue, @@ -465,6 +466,21 @@ describe('import', () => { expect((await collect(stream)).equals(full)).toBe(true) }) + it('treats a file exactly the sniff-window size as complete (observes EOF at the boundary)', async () => { + // Header widens under comma; a single giant data row pads the file to exactly the window + // with no trailing newline. If the loop stopped at the boundary without seeing EOF, the row + // would be dropped and comma would win — so this passing proves the boundary EOF is observed. + const header = 'a,b;c,d;e,f\n' + const suffix = ';3' + const pad = 'x'.repeat( + CSV_DELIMITER_SNIFF_BYTES - header.length - '1;'.length - suffix.length + ) + const full = Buffer.from(`${header}1;${pad}${suffix}`) + expect(full.length).toBe(CSV_DELIMITER_SNIFF_BYTES) + const { delimiter } = await sniffCsvDelimiterFromStream(Readable.from([full])) + expect(delimiter).toBe(';') + }) + it('counts the final row of a fully-buffered file with no trailing newline', async () => { // The whole file fits the sniff window (exhausted), so complete:true flows through and the // last row is scored — semicolon must win despite comma widening the header row. From 35156cc4c36ee17e26cf34397c60f33d19131af8 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 23 Jul 2026 10:58:53 -0700 Subject: [PATCH 7/7] test(tables): use sleep helper instead of raw setTimeout promise --- apps/sim/lib/table/import.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts index de461e37ec3..91baff2b94e 100644 --- a/apps/sim/lib/table/import.test.ts +++ b/apps/sim/lib/table/import.test.ts @@ -2,6 +2,7 @@ * @vitest-environment node */ import { Readable } from 'node:stream' +import { sleep } from '@sim/utils/helpers' import { describe, expect, it } from 'vitest' import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' import { @@ -522,7 +523,7 @@ describe('import', () => { }) const { stream } = await sniffCsvDelimiterFromStream(source) stream.destroy() - await new Promise((resolve) => setTimeout(resolve, 20)) + await sleep(20) expect(destroyed).toBe(true) })