Skip to content

Commit efe1de3

Browse files
authored
fix(tables): sniff CSV delimiter and capture the real header row on import (#5888)
* 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) * 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. * 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. * fix(tables): keep the final row when sniffing a complete CSV 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. * fix(tables): keep the double-cast annotation adjacent to its cast 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). * fix(tables): observe EOF at the exact sniff-window boundary 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. * test(tables): use sleep helper instead of raw setTimeout promise
1 parent 44749b8 commit efe1de3

7 files changed

Lines changed: 533 additions & 40 deletions

File tree

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

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ import {
3636
validateMapping,
3737
wouldExceedRowLimit,
3838
} from '@/lib/table'
39+
import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream'
3940
import { importAppendRows, importReplaceRows } from '@/lib/table/import-data'
4041
import { getUserSettings } from '@/lib/users/queries'
4142
import {
@@ -176,19 +177,26 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
176177
timezone = timezoneValidation.data
177178
}
178179

179-
const delimiter = extensionValidation.data === 'tsv' ? '\t' : ','
180-
const parser = createCsvParser(delimiter)
180+
// The extension only picks the fallback — the separator is sniffed from the file's
181+
// head so semicolon/pipe exports (European-locale Excel) don't land in one column.
182+
const { delimiter, stream: csvStream } = await sniffCsvDelimiterFromStream(
183+
file.stream,
184+
extensionValidation.data === 'tsv' ? '\t' : ','
185+
)
186+
let headers: string[] = []
187+
const parser = createCsvParser(delimiter, (parsedHeaders) => {
188+
headers = parsedHeaders
189+
})
181190
// `.pipe` doesn't forward source errors; forward them so the iterator throws.
182-
file.stream.on('error', (streamErr) => parser.destroy(streamErr))
183-
file.stream.pipe(parser)
191+
csvStream.on('error', (streamErr) => parser.destroy(streamErr))
192+
csvStream.pipe(parser)
184193
const rows: Record<string, unknown>[] = []
185194
for await (const record of parser as AsyncIterable<Record<string, unknown>>) {
186195
rows.push(record)
187196
}
188197
if (rows.length === 0) {
189198
return NextResponse.json({ error: 'CSV file has no data rows' }, { status: 400 })
190199
}
191-
const headers = Object.keys(rows[0])
192200

193201
let effectiveMapping = mapping ?? buildAutoMapping(headers, table.schema)
194202
let prospectiveTable: TableDefinition = table

apps/sim/app/api/table/import-csv/route.ts

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import {
2626
type TableDefinition,
2727
type TableSchema,
2828
} from '@/lib/table'
29+
import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream'
2930
import { getUserSettings } from '@/lib/users/queries'
3031
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
3132
import {
@@ -107,12 +108,20 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
107108
{ status: 400 }
108109
)
109110
}
110-
const delimiter = extensionResult.data === 'tsv' ? '\t' : ','
111+
// The extension only picks the fallback — the separator is sniffed from the file's
112+
// head so semicolon/pipe exports (European-locale Excel) don't land in one column.
113+
const { delimiter, stream: csvStream } = await sniffCsvDelimiterFromStream(
114+
file.stream,
115+
extensionResult.data === 'tsv' ? '\t' : ','
116+
)
111117

112-
const parser = createCsvParser(delimiter)
118+
let csvHeaders: string[] = []
119+
const parser = createCsvParser(delimiter, (headers) => {
120+
csvHeaders = headers
121+
})
113122
// `.pipe` doesn't forward source errors; forward them so the iterator throws.
114-
file.stream.on('error', (err) => parser.destroy(err))
115-
file.stream.pipe(parser)
123+
csvStream.on('error', (err) => parser.destroy(err))
124+
csvStream.pipe(parser)
116125

117126
interface ImportState {
118127
table: TableDefinition
@@ -139,7 +148,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
139148

140149
/** Infer the schema from the buffered sample and create the (empty) table. */
141150
const buildTable = async (sampleRows: Record<string, unknown>[]): Promise<ImportState> => {
142-
const inferred = inferSchemaFromCsv(Object.keys(sampleRows[0]), sampleRows)
151+
const inferred = inferSchemaFromCsv(csvHeaders, sampleRows)
143152
const schema: TableSchema = { columns: inferred.columns.map(normalizeColumn) }
144153
const planLimits = await getWorkspaceTableLimits(workspaceId)
145154
const tableName = sanitizeName(file.filename.replace(/\.[^.]+$/, ''), 'imported_table').slice(

apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,13 @@ import { createLogger } from '@sim/logger'
2525
import { getErrorMessage } from '@sim/utils/errors'
2626
import { truncate } from '@sim/utils/string'
2727
import { CSV_ASYNC_IMPORT_THRESHOLD_BYTES } from '@/lib/table/constants'
28-
import { buildAutoMapping, parseCsvBuffer } from '@/lib/table/import'
28+
import {
29+
buildAutoMapping,
30+
CSV_DELIMITER_SNIFF_BYTES,
31+
type CsvDelimiter,
32+
detectCsvDelimiter,
33+
parseCsvBuffer,
34+
} from '@/lib/table/import'
2935
import type { TableDefinition } from '@/lib/table/types'
3036
import {
3137
type CsvImportMode,
@@ -107,15 +113,27 @@ interface ParsedCsv {
107113
sampleRows: Record<string, unknown>[]
108114
}
109115

110-
/** Parses the head of a CSV/TSV for the mapping + sample, dropping any truncated final line. */
111-
async function parseCsvPreview(file: File, delimiter: ',' | '\t') {
116+
/**
117+
* Parses the head of a CSV/TSV for the mapping + sample, dropping any truncated final line.
118+
*
119+
* The separator is sniffed from the same leading bytes the server sniffs, so the mapping shown
120+
* here always matches the columns the import will actually produce.
121+
*/
122+
async function parseCsvPreview(file: File, fallbackDelimiter: CsvDelimiter) {
112123
const sliced = file.size > CSV_PREVIEW_BYTES
113124
const blob = sliced ? file.slice(0, CSV_PREVIEW_BYTES) : file
114125
let bytes = new Uint8Array(await blob.arrayBuffer())
115126
if (sliced) {
116127
const lastNewline = bytes.lastIndexOf(0x0a)
117128
if (lastNewline > 0) bytes = bytes.subarray(0, lastNewline + 1)
118129
}
130+
// The sniff sample is the whole file only when nothing was sliced off and it fits the window;
131+
// otherwise it's a truncated prefix whose last line may be partial.
132+
const delimiter = await detectCsvDelimiter(
133+
bytes.subarray(0, CSV_DELIMITER_SNIFF_BYTES),
134+
fallbackDelimiter,
135+
{ complete: !sliced && bytes.length <= CSV_DELIMITER_SNIFF_BYTES }
136+
)
119137
return parseCsvBuffer(bytes, delimiter)
120138
}
121139

@@ -180,8 +198,7 @@ export function ImportCsvDialog({
180198
setParsing(true)
181199
setParseError(null)
182200
try {
183-
const delimiter: ',' | '\t' = ext === 'tsv' ? '\t' : ','
184-
const { headers, rows } = await parseCsvPreview(file, delimiter)
201+
const { headers, rows } = await parseCsvPreview(file, ext === 'tsv' ? '\t' : ',')
185202
const autoMapping = buildAutoMapping(headers, table.schema)
186203
setParsed({
187204
file,
Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
import { Readable } from 'node:stream'
2+
import {
3+
CSV_DELIMITER_SNIFF_BYTES,
4+
type CsvDelimiter,
5+
detectCsvDelimiter,
6+
} from '@/lib/table/import'
7+
8+
export interface SniffedCsvStream {
9+
delimiter: CsvDelimiter
10+
/**
11+
* The full file contents, replayed from byte zero. Use this in place of the
12+
* source stream — the source has already been partially consumed.
13+
*/
14+
stream: Readable
15+
}
16+
17+
/**
18+
* Reads the head of a CSV/TSV stream, sniffs its field separator, then returns a
19+
* stream that replays the buffered head followed by the untouched remainder.
20+
*
21+
* Memory is bounded: it buffers only the source chunks needed to reach the
22+
* {@link CSV_DELIMITER_SNIFF_BYTES} window (one chunk past it, at worst), and the
23+
* detection sample it copies is capped at exactly that window regardless of how
24+
* large a single upstream chunk is. Those buffered chunks are then replayed
25+
* *by reference* — never re-copied — so a multi-GB import stays O(sniff window)
26+
* plus the single in-flight chunk. {@link detectCsvDelimiter} drops any partial
27+
* trailing line, so no newline trimming is needed here.
28+
*/
29+
export async function sniffCsvDelimiterFromStream(
30+
source: Readable,
31+
fallback: CsvDelimiter = ','
32+
): Promise<SniffedCsvStream> {
33+
const reader = source[Symbol.asyncIterator]()
34+
const chunks: Buffer[] = []
35+
let size = 0
36+
let exhausted = false
37+
38+
// Read until the buffered size *exceeds* the window (not just reaches it) or the stream ends.
39+
// The `<=` is deliberate: a file whose size is exactly the window must still trigger one more
40+
// read so EOF is observed and `exhausted` becomes true — otherwise it would be misjudged as a
41+
// truncated prefix and disagree with the buffered callers (which pass complete for that size).
42+
while (size <= CSV_DELIMITER_SNIFF_BYTES) {
43+
const next = await reader.next()
44+
if (next.done) {
45+
exhausted = true
46+
break
47+
}
48+
const chunk = Buffer.isBuffer(next.value) ? next.value : Buffer.from(next.value as Uint8Array)
49+
chunks.push(chunk)
50+
size += chunk.length
51+
}
52+
53+
// Copy at most the sniff window for detection — `Buffer.concat`'s length arg truncates,
54+
// so an oversized final chunk can't inflate this allocation past CSV_DELIMITER_SNIFF_BYTES.
55+
const sample = Buffer.concat(chunks, Math.min(size, CSV_DELIMITER_SNIFF_BYTES))
56+
// `exhausted` (the loop stopped on end-of-stream, never on exceeding the window) means the whole
57+
// file fit in the window, so its last row must count even without a trailing newline. Otherwise
58+
// the sample is a truncated prefix whose final line may be partial and should be dropped.
59+
const delimiter = await detectCsvDelimiter(sample, fallback, { complete: exhausted })
60+
61+
const stream = Readable.from(
62+
(async function* replay() {
63+
// Replay the already-read chunks by reference (no re-copy), then drain the rest.
64+
for (const chunk of chunks) yield chunk
65+
if (exhausted) return
66+
while (true) {
67+
const next = await reader.next()
68+
if (next.done) return
69+
yield next.value
70+
}
71+
})()
72+
)
73+
74+
// `Readable.from` closes the generator on destroy, which returns the source
75+
// iterator — but an early destroy of the wrapper before the generator is
76+
// pulled would otherwise leak the source's socket.
77+
stream.on('close', () => source.destroy())
78+
79+
return { delimiter, stream }
80+
}

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import {
1919
} from '@/lib/table'
2020
import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
2121
import { withGeneratedColumnIds } from '@/lib/table/column-keys'
22+
import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream'
2223
import { appendTableEvent } from '@/lib/table/events'
2324
import {
2425
addImportColumns,
@@ -103,6 +104,11 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
103104
// Stream the file rather than buffering it — a ~1M-row import must never be held in memory.
104105
source = await downloadFileStream({ key: fileKey, context: 'workspace' })
105106

107+
// The kickoff route's extension-derived delimiter is only the fallback — the separator is
108+
// sniffed from the file's head so semicolon/pipe exports don't collapse into one column.
109+
const sniffed = await sniffCsvDelimiterFromStream(source, delimiter)
110+
const csvStream = sniffed.stream
111+
106112
// Append must continue after the existing rows; create/replace start empty. Read once up
107113
// front (the import is the table's sole writer) and assign contiguous positions / threaded
108114
// order keys from it.
@@ -123,11 +129,14 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
123129
},
124130
})
125131

126-
const parser = createCsvParser(delimiter)
132+
let csvHeaders: string[] = []
133+
const parser = createCsvParser(sniffed.delimiter, (headers) => {
134+
csvHeaders = headers
135+
})
127136
// `.pipe` doesn't forward source errors; forward so the iterator throws.
128-
source.on('error', (err) => parser.destroy(err))
137+
csvStream.on('error', (err) => parser.destroy(err))
129138
byteCounter.on('error', (err) => parser.destroy(err))
130-
source.pipe(byteCounter).pipe(parser)
139+
csvStream.pipe(byteCounter).pipe(parser)
131140

132141
let schema: TableSchema | null = null
133142
let headerToColumn: Map<string, string> | null = null
@@ -142,7 +151,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
142151
* map onto the existing schema, optionally auto-creating `createColumns` first.
143152
*/
144153
const resolveSetup = async () => {
145-
const headers = Object.keys(sample[0])
154+
const headers = csvHeaders
146155

147156
if (mode === 'create') {
148157
const inferred = inferSchemaFromCsv(headers, sample)

0 commit comments

Comments
 (0)