mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
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
This commit is contained in:
@@ -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<string, unknown>[] = []
|
||||
for await (const record of parser as AsyncIterable<Record<string, unknown>>) {
|
||||
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
|
||||
|
||||
@@ -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<string, unknown>[]): Promise<ImportState> => {
|
||||
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(
|
||||
|
||||
+22
-5
@@ -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<string, unknown>[]
|
||||
}
|
||||
|
||||
/** 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,13 @@ async function parseCsvPreview(file: File, delimiter: ',' | '\t') {
|
||||
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,
|
||||
{ complete: !sliced && bytes.length <= CSV_DELIMITER_SNIFF_BYTES }
|
||||
)
|
||||
return parseCsvBuffer(bytes, delimiter)
|
||||
}
|
||||
|
||||
@@ -180,8 +198,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,
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
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.
|
||||
*
|
||||
* 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,
|
||||
fallback: CsvDelimiter = ','
|
||||
): Promise<SniffedCsvStream> {
|
||||
const reader = source[Symbol.asyncIterator]()
|
||||
const chunks: Buffer[] = []
|
||||
let size = 0
|
||||
let exhausted = false
|
||||
|
||||
// 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
|
||||
break
|
||||
}
|
||||
const chunk = Buffer.isBuffer(next.value) ? next.value : Buffer.from(next.value as Uint8Array)
|
||||
chunks.push(chunk)
|
||||
size += chunk.length
|
||||
}
|
||||
|
||||
// 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 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(
|
||||
(async function* replay() {
|
||||
// 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()
|
||||
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 }
|
||||
}
|
||||
@@ -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<void>
|
||||
// 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<void>
|
||||
},
|
||||
})
|
||||
|
||||
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<string, string> | null = null
|
||||
@@ -142,7 +151,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
|
||||
* 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)
|
||||
|
||||
@@ -2,14 +2,19 @@
|
||||
* @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 {
|
||||
buildAutoMapping,
|
||||
CSV_DELIMITER_SNIFF_BYTES,
|
||||
CsvImportValidationError,
|
||||
coerceRowsForTable,
|
||||
coerceValue,
|
||||
createCsvParser,
|
||||
csvParseOptions,
|
||||
dedupeHeaders,
|
||||
detectCsvDelimiter,
|
||||
inferColumnType,
|
||||
inferSchemaFromCsv,
|
||||
parseCsvBuffer,
|
||||
@@ -316,8 +321,225 @@ 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'])
|
||||
})
|
||||
|
||||
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', () => {
|
||||
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('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('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.
|
||||
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(';')
|
||||
})
|
||||
})
|
||||
|
||||
describe('sniffCsvDelimiterFromStream', () => {
|
||||
async function collect(stream: Readable): Promise<Buffer> {
|
||||
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('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.
|
||||
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.
|
||||
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))
|
||||
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 sleep(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)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
+168
-20
@@ -17,13 +17,43 @@ import { type NormalizeDateCellOptions, normalizeDateCellValue } from '@/lib/tab
|
||||
import type { ColumnDefinition, RowData, TableSchema } from '@/lib/table/types'
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* 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 function csvParseOptions(delimiter = ','): CsvParseOptions {
|
||||
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` 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 = ',',
|
||||
onHeaders?: (headers: string[]) => void
|
||||
): CsvParseOptions {
|
||||
return {
|
||||
columns: true,
|
||||
columns: (header: string[]) => {
|
||||
// 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,
|
||||
trim: true,
|
||||
relax_column_count: true,
|
||||
@@ -40,9 +70,127 @@ 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 = ','): Parser {
|
||||
return parseCsvStream(csvParseOptions(delimiter))
|
||||
export function createCsvParser(delimiter = ',', onHeaders?: (headers: string[]) => void): Parser {
|
||||
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<string>()
|
||||
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
|
||||
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 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.
|
||||
*
|
||||
* 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: 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 = ',',
|
||||
{ complete = false }: { complete?: boolean } = {}
|
||||
): Promise<CsvDelimiter> {
|
||||
const { parse } = await import('csv-parse/sync')
|
||||
const decoded = decodeCsvText(input)
|
||||
// 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 = !complete && lastNewline > 0 ? decoded.slice(0, lastNewline + 1) : decoded
|
||||
if (text.trim() === '') return fallback
|
||||
|
||||
let best: { delimiter: CsvDelimiter; fields: number; score: 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
|
||||
|
||||
// Modal row width: the most frequent column count, tie-broken toward the wider one.
|
||||
const widthCounts = new Map<number, number>()
|
||||
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
|
||||
|
||||
// Reward a split that is both wide and uniform; ties break toward the wider split.
|
||||
const score = fields * (modalFreq / records.length)
|
||||
|
||||
if (!best || score > best.score || (score === best.score && fields > best.fields)) {
|
||||
best = { delimiter, fields, score }
|
||||
}
|
||||
}
|
||||
|
||||
return best?.delimiter ?? fallback
|
||||
}
|
||||
|
||||
/** Narrower type than `COLUMN_TYPES` used internally for coercion. */
|
||||
@@ -103,24 +251,20 @@ export async function parseCsvBuffer(
|
||||
): Promise<{ headers: string[]; rows: Record<string, unknown>[] }> {
|
||||
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<string, unknown>[]
|
||||
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, options) as unknown as Record<string, unknown>[]
|
||||
|
||||
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 +648,11 @@ 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' : ',',
|
||||
{ complete: buffer.length <= CSV_DELIMITER_SNIFF_BYTES }
|
||||
)
|
||||
return parseCsvBuffer(buffer, delimiter)
|
||||
}
|
||||
throw new Error(`Unsupported file format: "${ext ?? fileName}". Supported: csv, tsv, json`)
|
||||
|
||||
Reference in New Issue
Block a user