From cd0023bb329179136e1461e1c3c6f4cf3c3c7ff8 Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Fri, 31 Jul 2026 15:45:47 -0400 Subject: [PATCH 1/9] feat: streaming bulk-ingest bounded by flush_bytes * concurrency --- src/es/helpers/bulk-ingest.ts | 322 ++++++++++++++++++++-------------- 1 file changed, 195 insertions(+), 127 deletions(-) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index dc54a5d2..1dd75007 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -4,20 +4,18 @@ */ import { z } from 'zod' -import { readFileSync } from 'node:fs' +import { createReadStream } from 'node:fs' +import { createInterface } from 'node:readline' +import type { Readable } from 'node:stream' +import { parse as parseCsvStream } from 'csv-parse' import type { EsClient } from '../../lib/es-client.ts' import { defineCommand } from '../../factory.ts' import type { OpaqueCommandHandle, JsonValue } from '../../factory.ts' import { getEsClient } from '../../lib/es-client.ts' import { missingConfigError, transportError } from '../errors.ts' import { - parseInput, - parseCsvInput, - readRawInput, globFiles, - buildBulkNdjsonBody, retryWithBackoff, - runWithConcurrency, ProgressReporter } from './shared.ts' @@ -51,44 +49,23 @@ const inputSchema = z.object({ type BulkIngestInput = z.infer -/** - * Splits an array of documents into batches where each batch's serialized - * size does not exceed the byte threshold. - */ -function splitIntoBatches (docs: unknown[], flushBytes: number): unknown[][] { - const batches: unknown[][] = [] - let currentBatch: unknown[] = [] - let currentSize = 0 - - for (const doc of docs) { - const docSize = JSON.stringify(doc).length + 1 // +1 for newline - if (currentBatch.length > 0 && currentSize + docSize > flushBytes) { - batches.push(currentBatch) - currentBatch = [] - currentSize = 0 - } - currentBatch.push(doc) - currentSize += docSize - } - if (currentBatch.length > 0) { - batches.push(currentBatch) +/** Bounded concurrency via a counting semaphore. */ +class Semaphore { + private count: number + private readonly waiters: Array<() => void> = [] + + constructor (n: number) { this.count = n } + + async acquire (): Promise { + if (this.count > 0) { this.count--; return } + await new Promise(r => this.waiters.push(r)) } - return batches -} -/** Parses raw file content according to the selected source format. */ -function parseByFormat (raw: string, opts: BulkIngestInput): unknown[] { - if (opts.source_format === 'csv') { - const csvColumns = opts.csv_columns != null - ? opts.csv_columns.split(',').map((c) => c.trim()).filter(Boolean) - : undefined - return parseCsvInput(raw, { - ...(opts.csv_delimiter != null && { delimiter: opts.csv_delimiter }), - ...(csvColumns != null && { columns: csvColumns }), - ...(opts.skip_header != null && { skipHeader: opts.skip_header }), - }) + release (): void { + const next = this.waiters.shift() + if (next != null) next() + else this.count++ } - return parseInput(raw) } /** Returns the default glob pattern for the given source format. */ @@ -97,46 +74,6 @@ function defaultGlob (format: SourceFormat): string { return '**/*.{json,ndjson,jsonl}' } -/** Collects documents from the resolved input source. */ -function collectDocuments (opts: BulkIngestInput): { docs: unknown[], filesProcessed: number } { - const { data_file, data_dir } = opts - - if (data_file != null && data_dir != null) { - throw new Error('Provide only one input source: --data-file or --data-dir (not both)') - } - - if (data_dir != null) { - const pattern = opts.glob ?? defaultGlob(opts.source_format) - const recursive = opts.no_recursive !== true - const resolvedPattern = recursive ? pattern : pattern.replace(/^\*\*\//, '') - const files = globFiles(data_dir, resolvedPattern) - if (files.length === 0) { - throw new Error(`No files matched pattern "${resolvedPattern}" in ${data_dir}`) - } - const allDocs: unknown[] = [] - for (const file of files) { - const raw = readFileSync(file, 'utf-8') - allDocs.push(...parseByFormat(raw, opts)) - } - return { docs: allDocs, filesProcessed: files.length } - } - - if (data_file != null) { - const raw = readRawInput(data_file) - if (raw == null || raw.trim().length === 0) { - throw new Error('No input data received from file') - } - return { docs: parseByFormat(raw, opts), filesProcessed: 1 } - } - - // Fall back to stdin - const raw = readRawInput() - if (raw == null || raw.trim().length === 0) { - throw new Error('No input provided. Use --data-file, --data-dir, or pipe data to stdin') - } - return { docs: parseByFormat(raw, opts), filesProcessed: 0 } -} - /** Sends a single bulk batch to Elasticsearch. Returns the count of errors. */ async function sendBatch ( transport: EsClient, @@ -161,6 +98,167 @@ async function sendBatch ( return { errors: errorCount, total } } +/** + * Streams documents from all input sources and sends them as bulk batches. + * + * Peak memory is bounded by flush_bytes * concurrency regardless of input size. + * The semaphore provides backpressure: the producer blocks once concurrency + * slots are exhausted, preventing unbounded batch accumulation. + */ +async function streamBulkIngest ( + opts: BulkIngestInput, + transport: EsClient, + reporter: ProgressReporter +): Promise { + const { flush_bytes, concurrency, retries, retry_delay, index, pipeline, routing } = opts + + const actionLine = JSON.stringify({ + index: { + ...(index != null && { _index: index }), + ...(pipeline != null && { pipeline }), + ...(routing != null && { routing }), + } + }) + + const sem = new Semaphore(concurrency) + const errors: unknown[] = [] + + let buf = '' + let bufBytes = 0 + + const submitBatch = async (body: string): Promise => { + await sem.acquire() + // Fire-and-forget: producer continues reading while this batch is in flight. + retryWithBackoff( + async () => { + const res = await sendBatch(transport, body, index) + if (res.errors > 0 && res.errors === res.total) { + throw new Error(`Bulk batch failed: ${res.errors}/${res.total} errors`) + } + return res + }, + { retries, delay: retry_delay } + ).then(res => { + reporter.report(res.total, res.errors) + }).catch(err => { + errors.push(err) + }).finally(() => sem.release()) + } + + const flush = async (): Promise => { + if (bufBytes === 0) return + const body = buf + buf = '' + bufBytes = 0 + await submitBatch(body) + } + + const addDoc = async (docJson: string): Promise => { + const pair = actionLine + '\n' + docJson + '\n' + buf += pair + bufBytes += pair.length + if (bufBytes >= flush_bytes) await flush() + } + + // Resolve file list + const { data_file, data_dir, source_format } = opts + + if (data_file != null && data_dir != null) { + throw Object.assign(new Error('Provide only one input source: --data-file or --data-dir (not both)'), { code: 'input_error' }) + } + + let filePaths: Array + + if (data_dir != null) { + const pattern = opts.glob ?? defaultGlob(source_format) + const recursive = opts.no_recursive !== true + const resolvedPattern = recursive ? pattern : pattern.replace(/^\*\*\//, '') + const found = globFiles(data_dir, resolvedPattern) + if (found.length === 0) { + throw Object.assign(new Error(`No files matched pattern "${resolvedPattern}" in ${data_dir}`), { code: 'input_error' }) + } + reporter.filesProcessed = found.length + filePaths = found + } else if (data_file != null) { + reporter.filesProcessed = 1 + filePaths = [data_file] + } else { + if (process.stdin.isTTY === true) { + throw Object.assign(new Error('No input provided. Use --data-file, --data-dir, or pipe data to stdin'), { code: 'input_error' }) + } + filePaths = [undefined] + } + + for (const filePath of filePaths) { + const stream: Readable = filePath != null ? createReadStream(filePath, { encoding: 'utf-8' }) : process.stdin + + if (source_format === 'csv') { + const csvColumns = opts.csv_columns != null + ? opts.csv_columns.split(',').map(c => c.trim()).filter(Boolean) + : undefined + const parser = parseCsvStream({ + delimiter: opts.csv_delimiter ?? ',', + columns: csvColumns != null && csvColumns.length > 0 ? csvColumns : true, + from_line: opts.skip_header === true ? 2 : 1, + skip_empty_lines: true, + trim: true, + cast (value) { + if (value === 'true') return true + if (value === 'false') return false + if (value !== '' && !isNaN(Number(value))) return Number(value) + return value + } + }) + stream.pipe(parser) + for await (const record of parser) { + await addDoc(JSON.stringify(record)) + } + } else { + // ndjson / json: line-by-line for NDJSON, buffered fallback for JSON arrays + const rl = createInterface({ input: stream, crlfDelay: Infinity }) + let isJsonArray: boolean | null = null // null = not yet determined + let arrayBuf = '' + + for await (const line of rl) { + const trimmed = line.trim() + if (trimmed.length === 0) continue + + if (isJsonArray === null) { + isJsonArray = trimmed.startsWith('[') + } + + if (isJsonArray) { + arrayBuf += line + '\n' + continue + } + + try { + await addDoc(JSON.stringify(JSON.parse(trimmed))) + } catch { + throw new Error(`Failed to parse NDJSON line: ${trimmed.slice(0, 80)}`) + } + } + + if (isJsonArray === true && arrayBuf.trim().length > 0) { + const parsed: unknown = JSON.parse(arrayBuf) + if (!Array.isArray(parsed)) throw new Error('Expected a JSON array') + for (const doc of parsed) { + await addDoc(JSON.stringify(doc)) + } + } + } + } + + await flush() + + // Drain: acquire all slots to confirm every in-flight batch has finished. + for (let i = 0; i < concurrency; i++) { + await sem.acquire() + } + + if (errors.length > 0) throw errors[0] +} + function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { return async (parsed: { input?: BulkIngestInput; options: Record }): Promise => { const opts = parsed.input! @@ -172,56 +270,26 @@ function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { return missingConfigError(err) } - let docs: unknown[] - let filesProcessed: number - try { - const result = collectDocuments(opts) - docs = result.docs - filesProcessed = result.filesProcessed - } catch (err) { - return { - error: { - code: 'input_error', - message: err instanceof Error ? err.message : String(err) - } - } - } - - if (docs.length === 0) { - return { total: 0, succeeded: 0, failed: 0, retries: 0, elapsed_ms: 0 } - } - - const batches = splitIntoBatches(docs, opts.flush_bytes) - const reporter = new ProgressReporter() - reporter.filesProcessed = filesProcessed - - const { retries, retry_delay, index, pipeline, routing } = opts try { - await runWithConcurrency(batches, opts.concurrency, async (batch) => { - const ndjsonBody = buildBulkNdjsonBody(batch, { index, pipeline, routing }) - - const result = await retryWithBackoff( - async () => { - const res = await sendBatch(transport, ndjsonBody, index) - if (res.errors > 0) { - // Only retry if all items failed (likely a transient cluster issue) - // Partial failures are reported as-is - if (res.errors === res.total) { - throw new Error(`Bulk batch failed: ${res.errors}/${res.total} errors`) - } - } - return res - }, - { retries, delay: retry_delay } - ) - - reporter.report(result.total, result.errors) - return result - }) + await streamBulkIngest(opts, transport, reporter) } catch (err) { - // If retries exhausted, report what we have so far + const code = (err as { code?: string }).code + if (code === 'input_error' || (err instanceof Error && ( + err.message.startsWith('No files matched') || + err.message.startsWith('Provide only one') || + err.message.startsWith('No input provided') || + err.message.startsWith('Failed to parse') || + err.message.startsWith('Expected a JSON') + ))) { + return { + error: { + code: 'input_error', + message: err instanceof Error ? err.message : String(err) + } + } + } return transportError(err) } From 9b2992f0593de445c4133efc79363d8be68987ce Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Fri, 31 Jul 2026 15:45:50 -0400 Subject: [PATCH 2/9] test: fix runCommand helper to handle Node test runner IPC on stdout --- test/es/helpers/bulk-ingest.test.ts | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index 304f6271..ddc3c721 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -75,17 +75,25 @@ async function runCommand (args: string[], deps: BulkIngestDeps): Promise 0 ? errOutput : stdOutput - if (output.trim().length > 0) { - try { - return JSON.parse(output.trim()) - } catch { - return output.trim() + + if (errOutput.trim().length > 0) { + try { return JSON.parse(errOutput.trim()) } catch { return errOutput.trim() } + } + + for (let i = stdoutChunks.length - 1; i >= 0; i--) { + const t = stdoutChunks[i]!.trim() + if ((t.startsWith('{') || t.startsWith('[')) && t.length > 0) { + try { return JSON.parse(t) } catch {} } } + const stdOutput = stdoutChunks.join('') + if (stdOutput.trim().length > 0) return stdOutput.trim() return undefined } From 4071fa9042a5927540fbe7d1cf4ed699575c468a Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 13:42:44 -0400 Subject: [PATCH 3/9] fix: avoid empty catch block in bulk-ingest test --- test/es/helpers/bulk-ingest.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index ddc3c721..51781ec9 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -89,7 +89,7 @@ async function runCommand (args: string[], deps: BulkIngestDeps): Promise= 0; i--) { const t = stdoutChunks[i]!.trim() if ((t.startsWith('{') || t.startsWith('[')) && t.length > 0) { - try { return JSON.parse(t) } catch {} + try { return JSON.parse(t) } catch { /* not JSON, try the next chunk */ } } } const stdOutput = stdoutChunks.join('') From a95d1acc518783ba196163d7f0c825173e3c0277 Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 13:42:44 -0400 Subject: [PATCH 4/9] refactor: drop dead JSON/CSV helpers left by streaming --- src/es/helpers/shared.ts | 88 ---------------------------------- test/es/helpers/shared.test.ts | 74 ---------------------------- 2 files changed, 162 deletions(-) diff --git a/src/es/helpers/shared.ts b/src/es/helpers/shared.ts index 0db74e70..d2b9bf6a 100644 --- a/src/es/helpers/shared.ts +++ b/src/es/helpers/shared.ts @@ -5,75 +5,8 @@ import { readFileSync, globSync } from 'node:fs' import { resolve } from 'node:path' -import { parse as parseCsv } from 'csv-parse/sync' import type { JsonValue } from '../../factory.ts' -/** - * Parses raw text input as either a JSON array or NDJSON (newline-delimited JSON). - * Auto-detects the format: if the trimmed input starts with `[`, it's parsed as JSON array; - * otherwise each non-empty line is parsed as a separate JSON object. - */ -export function parseInput (raw: string): unknown[] { - const trimmed = raw.trim() - if (trimmed.length === 0) return [] - - if (trimmed.startsWith('[')) { - const parsed = JSON.parse(trimmed) - if (!Array.isArray(parsed)) { - throw new Error('Expected a JSON array, got: ' + typeof parsed) - } - return parsed - } - - const lines = trimmed.split('\n') - const docs: unknown[] = [] - for (let i = 0; i < lines.length; i++) { - const line = lines[i]!.trim() - if (line.length === 0) continue - try { - docs.push(JSON.parse(line)) - } catch { - throw new Error(`Failed to parse NDJSON at line ${i + 1}: ${line.slice(0, 80)}`) - } - } - return docs -} - -export interface CsvParseOptions { - delimiter?: string - /** Explicit column names. If provided, the first data row is treated as data (not headers). */ - columns?: string[] - /** Skip the first row of the file (useful when the file has a header you want to discard). */ - skipHeader?: boolean -} - -/** - * Parses CSV text into an array of objects. - * By default the first row is used as column headers. - * If `columns` is provided, those names are used instead and every row is treated as data. - * `skipHeader: true` discards the first row (combine with `columns` to rename headers). - */ -export function parseCsvInput (raw: string, opts: CsvParseOptions = {}): unknown[] { - const { delimiter = ',', columns, skipHeader = false } = opts - - const fromLine = skipHeader ? 2 : 1 - const columnsOpt: string[] | true = columns != null && columns.length > 0 ? columns : true - - return parseCsv(raw, { - delimiter, - columns: columnsOpt, - from_line: fromLine, - skip_empty_lines: true, - trim: true, - cast (value) { - if (value === 'true') return true - if (value === 'false') return false - if (value !== '' && !isNaN(Number(value))) return Number(value) - return value - }, - }) as unknown[] -} - /** * Reads raw text content from a file path or stdin. * Returns `undefined` when no input is available (interactive TTY with no file). @@ -99,27 +32,6 @@ export function globFiles (dir: string, pattern: string): string[] { return matches.map((f) => resolve(absDir, f)).sort() } -/** - * Builds an NDJSON body for the Elasticsearch `_bulk` API. - * Each document is wrapped in an `{"index": {...}}` action line followed by the document line. - */ -export function buildBulkNdjsonBody ( - docs: unknown[], - opts: { index?: string | undefined, pipeline?: string | undefined, routing?: string | undefined } -): string { - const lines: string[] = [] - for (const doc of docs) { - const action: Record = {} - if (opts.index != null) action._index = opts.index - if (opts.pipeline != null) action.pipeline = opts.pipeline - if (opts.routing != null) action.routing = opts.routing - lines.push(JSON.stringify({ index: action })) - lines.push(JSON.stringify(doc)) - } - // bulk API requires a trailing newline - return lines.join('\n') + '\n' -} - /** * Retries an async function with exponential backoff. * On each failure the delay doubles. Rethrows the last error after exhausting retries. diff --git a/test/es/helpers/shared.test.ts b/test/es/helpers/shared.test.ts index 09b6e31b..9df835eb 100644 --- a/test/es/helpers/shared.test.ts +++ b/test/es/helpers/shared.test.ts @@ -6,85 +6,11 @@ import { describe, it } from 'node:test' import assert from 'node:assert/strict' import { - parseInput, - buildBulkNdjsonBody, retryWithBackoff, runWithConcurrency, ProgressReporter } from '../../../src/es/helpers/shared.ts' -describe('parseInput', () => { - it('parses a JSON array', () => { - const result = parseInput('[{"a":1},{"b":2}]') - assert.deepStrictEqual(result, [{ a: 1 }, { b: 2 }]) - }) - - it('parses NDJSON', () => { - const result = parseInput('{"a":1}\n{"b":2}\n') - assert.deepStrictEqual(result, [{ a: 1 }, { b: 2 }]) - }) - - it('skips empty lines in NDJSON', () => { - const result = parseInput('{"a":1}\n\n{"b":2}\n\n') - assert.deepStrictEqual(result, [{ a: 1 }, { b: 2 }]) - }) - - it('returns empty array for empty input', () => { - assert.deepStrictEqual(parseInput(''), []) - assert.deepStrictEqual(parseInput(' \n '), []) - }) - - it('parses a single JSON object as NDJSON', () => { - const result = parseInput('{"a":1}') - assert.deepStrictEqual(result, [{ a: 1 }]) - }) - - it('throws on malformed NDJSON line', () => { - assert.throws(() => parseInput('{"a":1}\nnot json\n'), /Failed to parse NDJSON at line 2/) - }) - - it('handles JSON array with whitespace', () => { - const result = parseInput(' \n [{"a":1}] \n ') - assert.deepStrictEqual(result, [{ a: 1 }]) - }) -}) - -describe('buildBulkNdjsonBody', () => { - it('wraps documents in index actions', () => { - const body = buildBulkNdjsonBody([{ title: 'doc1' }, { title: 'doc2' }], { index: 'my-index' }) - const lines = body.split('\n') - assert.equal(lines.length, 5) // 4 content lines + trailing empty line - assert.deepStrictEqual(JSON.parse(lines[0]), { index: { _index: 'my-index' } }) - assert.deepStrictEqual(JSON.parse(lines[1]), { title: 'doc1' }) - assert.deepStrictEqual(JSON.parse(lines[2]), { index: { _index: 'my-index' } }) - assert.deepStrictEqual(JSON.parse(lines[3]), { title: 'doc2' }) - assert.equal(lines[4], '') // trailing newline - }) - - it('includes pipeline and routing in action metadata', () => { - const body = buildBulkNdjsonBody([{ a: 1 }], { - index: 'idx', - pipeline: 'my-pipe', - routing: 'shard-1' - }) - const action = JSON.parse(body.split('\n')[0]) - assert.deepStrictEqual(action, { - index: { _index: 'idx', pipeline: 'my-pipe', routing: 'shard-1' } - }) - }) - - it('returns trailing newline for empty docs', () => { - const body = buildBulkNdjsonBody([], { index: 'idx' }) - assert.equal(body, '\n') - }) - - it('omits undefined metadata fields', () => { - const body = buildBulkNdjsonBody([{ a: 1 }], {}) - const action = JSON.parse(body.split('\n')[0]) - assert.deepStrictEqual(action, { index: {} }) - }) -}) - describe('retryWithBackoff', () => { it('returns result on first success', async () => { const result = await retryWithBackoff(() => Promise.resolve(42), { retries: 3, delay: 1 }) From 086d06425df6c5b2a018d8f8da1abc0c9f6b7bd6 Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 19:00:34 -0400 Subject: [PATCH 5/9] feat: stream json array parsing in bulk ingest --- src/es/helpers/bulk-ingest.ts | 100 +++++++++++++++++++++++++--- test/es/helpers/bulk-ingest.test.ts | 91 +++++++++++++++++++++++++ 2 files changed, 181 insertions(+), 10 deletions(-) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index 1dd75007..b70ae8dc 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -68,6 +68,81 @@ class Semaphore { } } +/** + * Extracts top-level elements from a streamed JSON array (`[doc, doc, ...]`) + * without ever holding the whole array in memory. Feed it text chunks in + * order via {@link feed}; it returns each complete top-level element (as + * unparsed JSON text) as soon as its closing delimiter is seen. Only the + * current in-progress element is buffered, so memory is bounded by the + * largest single element, not the file size. + */ +class JsonArraySplitter { + private started = false + private closed = false + private depth = 0 + private inString = false + private escaped = false + private buf = '' + private hasContent = false + + feed (chunk: string): string[] { + const elements: string[] = [] + for (let i = 0; i < chunk.length && !this.closed; i++) { + const c = chunk[i]! + + if (!this.started) { + if (c === '[') this.started = true + continue + } + + if (this.inString) { + this.buf += c + if (this.escaped) this.escaped = false + else if (c === '\\') this.escaped = true + else if (c === '"') this.inString = false + continue + } + + if (this.depth > 0) { + if (c === '"') this.inString = true + else if (c === '{' || c === '[') this.depth++ + else if (c === '}' || c === ']') this.depth-- + this.buf += c + continue + } + + // depth === 0: between elements, or inside an unbracketed scalar (number/bool/null/string) + if (c === '"') { + this.inString = true + this.buf += c + this.hasContent = true + } else if (c === ' ' || c === '\n' || c === '\r' || c === '\t' || c === ',') { + if (this.hasContent) elements.push(this.emit()) + } else if (c === ']') { + if (this.hasContent) elements.push(this.emit()) + this.closed = true + } else { + if (c === '{' || c === '[') this.depth++ + this.buf += c + this.hasContent = true + } + } + return elements + } + + /** True once the closing `]` of the array has been consumed. */ + isClosed (): boolean { + return this.closed + } + + private emit (): string { + const el = this.buf + this.buf = '' + this.hasContent = false + return el + } +} + /** Returns the default glob pattern for the given source format. */ function defaultGlob (format: SourceFormat): string { if (format === 'csv') return '**/*.csv' @@ -214,10 +289,11 @@ async function streamBulkIngest ( await addDoc(JSON.stringify(record)) } } else { - // ndjson / json: line-by-line for NDJSON, buffered fallback for JSON arrays + // ndjson: line-by-line. json (JSON array): streamed element-by-element via + // JsonArraySplitter, so a multi-GB array never gets buffered whole. const rl = createInterface({ input: stream, crlfDelay: Infinity }) let isJsonArray: boolean | null = null // null = not yet determined - let arrayBuf = '' + const arraySplitter = new JsonArraySplitter() for await (const line of rl) { const trimmed = line.trim() @@ -228,7 +304,15 @@ async function streamBulkIngest ( } if (isJsonArray) { - arrayBuf += line + '\n' + for (const element of arraySplitter.feed(line + '\n')) { + let doc: unknown + try { + doc = JSON.parse(element) + } catch { + throw new Error(`Failed to parse JSON array element: ${element.slice(0, 80)}`) + } + await addDoc(JSON.stringify(doc)) + } continue } @@ -239,12 +323,8 @@ async function streamBulkIngest ( } } - if (isJsonArray === true && arrayBuf.trim().length > 0) { - const parsed: unknown = JSON.parse(arrayBuf) - if (!Array.isArray(parsed)) throw new Error('Expected a JSON array') - for (const doc of parsed) { - await addDoc(JSON.stringify(doc)) - } + if (isJsonArray === true && !arraySplitter.isClosed()) { + throw new Error('Unexpected end of input: JSON array was not closed') } } } @@ -281,7 +361,7 @@ function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { err.message.startsWith('Provide only one') || err.message.startsWith('No input provided') || err.message.startsWith('Failed to parse') || - err.message.startsWith('Expected a JSON') + err.message.startsWith('Unexpected end of input') ))) { return { error: { diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index 51781ec9..94997841 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -279,6 +279,97 @@ describe('bulk-ingest command', () => { assert.equal(requests[0]!.opts, undefined) }) + it('streams a pretty-printed, multi-line JSON array', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), JSON.stringify([{ title: 'doc1' }, { title: 'doc2' }], null, 2)) + + const { transport, requests } = mockTransport([successResponse(2)]) + + await runCommand(['--index', 'test-idx', '--data-file', join(tmpDir, 'data.json'), '--json'], makeDeps(transport)) + + assert.equal(requests.length, 1) + const body = requests[0]!.params.body as string + assert.ok(body.includes('"doc1"')) + assert.ok(body.includes('"doc2"')) + }) + + it('streams a minified JSON array with multiple documents on one line', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), '[{"a":1},{"b":2},{"c":3}]') + + const { transport, requests } = mockTransport([successResponse(3)]) + + await runCommand(['--index', 'test-idx', '--data-file', join(tmpDir, 'data.json'), '--json'], makeDeps(transport)) + + assert.equal(requests.length, 1) + const body = requests[0]!.params.body as string + assert.ok(body.includes('"a"') && body.includes('"b"') && body.includes('"c"')) + }) + + it('correctly splits JSON array elements containing commas and brackets inside strings', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const docs = [ + { note: 'contains a, comma and a ] bracket' }, + { note: 'contains an escaped \\" quote' }, + { nested: { arr: [1, 2, { deep: true }] } } + ] + writeFileSync(join(tmpDir, 'data.json'), JSON.stringify(docs)) + + const { transport, requests } = mockTransport([successResponse(3)]) + + await runCommand(['--index', 'test-idx', '--data-file', join(tmpDir, 'data.json'), '--json'], makeDeps(transport)) + + assert.equal(requests.length, 1) + const body = requests[0]!.params.body as string + const lines = body.trim().split('\n').filter((_, i) => i % 2 === 1) // doc lines only + assert.deepStrictEqual(lines.map((l) => JSON.parse(l)), docs) + }) + + it('rejects a JSON array that is never closed', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), '[{"a":1},{"b":2}') + + const { transport } = mockTransport([successResponse(1)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', join(tmpDir, 'data.json'), + '--json' + ], makeDeps(transport)) as Record + + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /not closed/) + }) + + it('rejects an unparseable element inside a JSON array', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), '[{"a":1}, not-json, {"b":2}]') + + const { transport } = mockTransport([successResponse(1)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', join(tmpDir, 'data.json'), + '--json' + ], makeDeps(transport)) as Record + + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /Failed to parse JSON array element/) + }) + + it('streams an empty JSON array without sending any batch', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), '[]') + + const { transport, requests } = mockTransport([successResponse(0)]) + + await runCommand(['--index', 'test-idx', '--data-file', join(tmpDir, 'data.json'), '--json'], makeDeps(transport)) + + assert.equal(requests.length, 0) + }) + it('returns empty summary for zero documents', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) writeFileSync(join(tmpDir, 'data.json'), '[]') From 9051cfd093ffa1e1476e072ccca74ab923bde609 Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 19:30:37 -0400 Subject: [PATCH 6/9] fix: count failed batches in bulk ingest summary --- src/es/helpers/bulk-ingest.ts | 34 +++++++-- test/es/helpers/bulk-ingest.test.ts | 113 ++++++++++++++++++++++++---- 2 files changed, 123 insertions(+), 24 deletions(-) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index b70ae8dc..8b497c85 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -196,12 +196,18 @@ async function streamBulkIngest ( }) const sem = new Semaphore(concurrency) - const errors: unknown[] = [] + const errors: string[] = [] let buf = '' let bufBytes = 0 - - const submitBatch = async (body: string): Promise => { + let bufDocs = 0 + + // A batch can fail two ways: retryWithBackoff exhausts retries after a + // network/transport throw (no response, so we don't know per-doc status), + // or every item in the ES response errored. Either way we still know + // exactly how many docs were in the batch (docCount), so that count is + // always reflected in the summary instead of silently vanishing. + const submitBatch = async (body: string, docCount: number): Promise => { await sem.acquire() // Fire-and-forget: producer continues reading while this batch is in flight. retryWithBackoff( @@ -215,23 +221,28 @@ async function streamBulkIngest ( { retries, delay: retry_delay } ).then(res => { reporter.report(res.total, res.errors) - }).catch(err => { - errors.push(err) + }).catch((err: unknown) => { + const message = err instanceof Error ? err.message : String(err) + errors.push(`${docCount} doc(s) dropped: ${message}`) + reporter.report(docCount, docCount) }).finally(() => sem.release()) } const flush = async (): Promise => { if (bufBytes === 0) return const body = buf + const docCount = bufDocs buf = '' bufBytes = 0 - await submitBatch(body) + bufDocs = 0 + await submitBatch(body, docCount) } const addDoc = async (docJson: string): Promise => { const pair = actionLine + '\n' + docJson + '\n' buf += pair bufBytes += pair.length + bufDocs++ if (bufBytes >= flush_bytes) await flush() } @@ -336,7 +347,9 @@ async function streamBulkIngest ( await sem.acquire() } - if (errors.length > 0) throw errors[0] + if (errors.length > 0) { + throw new Error(`${errors.length} batch(es) failed:\n${errors.join('\n')}`) + } } function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { @@ -355,6 +368,10 @@ function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { try { await streamBulkIngest(opts, transport, reporter) } catch (err) { + // Include whatever progress was made before the failure, so the caller + // knows how much data actually landed and doesn't have to guess before + // deciding whether/how to re-run (e.g. with an upsert). + const summary = reporter.summary() as Record const code = (err as { code?: string }).code if (code === 'input_error' || (err instanceof Error && ( err.message.startsWith('No files matched') || @@ -364,13 +381,14 @@ function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { err.message.startsWith('Unexpected end of input') ))) { return { + ...summary, error: { code: 'input_error', message: err instanceof Error ? err.message : String(err) } } } - return transportError(err) + return { ...summary, ...(transportError(err) as Record) } } return reporter.summary() diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index 94997841..511a72bd 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -43,6 +43,14 @@ function successResponse (count: number): { errors: boolean, items: Array> } { + return { + errors: true, + items: Array.from({ length: count }, () => ({ index: { status: 500 } })) + } +} + /** Runs the bulk-ingest command programmatically and returns handler result. */ async function runCommand (args: string[], deps: BulkIngestDeps): Promise { const cmd = createBulkIngestCommand(deps) @@ -76,24 +84,32 @@ async function runCommand (args: string[], deps: BulkIngestDeps): Promise 0) { - try { return JSON.parse(errOutput.trim()) } catch { return errOutput.trim() } - } - - for (let i = stdoutChunks.length - 1; i >= 0; i--) { - const t = stdoutChunks[i]!.trim() - if ((t.startsWith('{') || t.startsWith('[')) && t.length > 0) { - try { return JSON.parse(t) } catch { /* not JSON, try the next chunk */ } + // Each process.stdout/stderr.write() call is its own array entry, and each + // logical write (a progress update, a binary IPC frame from Node's test + // runner, or the final JSON result) is a complete, self-contained chunk. + // So scan chunks (not a joined blob, which can merge binary-frame trailing + // bytes into the JSON chunk and break the startsWith('{') check) in + // reverse for the last one that parses as JSON. + const findJsonChunk = (chunks: string[]): unknown => { + for (let i = chunks.length - 1; i >= 0; i--) { + const t = chunks[i]!.trim() + if ((t.startsWith('{') || t.startsWith('[')) && t.length > 0) { + try { return JSON.parse(t) } catch { /* not JSON, try the previous chunk */ } + } } + return undefined } - const stdOutput = stdoutChunks.join('') - if (stdOutput.trim().length > 0) return stdOutput.trim() + + const errResult = findJsonChunk(stderrChunks) + if (errResult !== undefined) return errResult + + const outResult = findJsonChunk(stdoutChunks) + if (outResult !== undefined) return outResult + + const errOutput = stderrChunks.join('').trim() + if (errOutput.length > 0) return errOutput + const stdOutput = stdoutChunks.join('').trim() + if (stdOutput.length > 0) return stdOutput return undefined } @@ -370,6 +386,71 @@ describe('bulk-ingest command', () => { assert.equal(requests.length, 0) }) + it('counts a fully-failed batch instead of dropping it from the summary', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + writeFileSync(join(tmpDir, 'data.json'), JSON.stringify([{ a: 1 }, { a: 2 }])) + + const { transport } = mockTransport([failureResponse(2)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', join(tmpDir, 'data.json'), + '--retries', '0', + '--json' + ], makeDeps(transport)) as Record + + assert.equal(result.total, 2) + assert.equal(result.failed, 2) + assert.equal(result.succeeded, 0) + const error = result.error as Record + assert.equal(error.code, 'transport_error') + assert.match(error.message as string, /2 doc\(s\) dropped/) + }) + + it('surfaces every failed batch, not just the first', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const docs = Array.from({ length: 4 }, (_, i) => ({ id: i, data: 'x'.repeat(100) })) + writeFileSync(join(tmpDir, 'data.json'), JSON.stringify(docs)) + + // Small flush-bytes forces multiple batches; every batch fails. + const { transport, requests } = mockTransport([failureResponse(1)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', join(tmpDir, 'data.json'), + '--flush-bytes', '50', + '--retries', '0', + '--json' + ], makeDeps(transport)) as Record + + assert.ok(requests.length > 1, `expected multiple batches, got ${requests.length}`) + const error = result.error as Record + const dropMentions = (error.message as string).split('doc(s) dropped').length - 1 + assert.equal(dropMentions, requests.length, 'expected every failed batch to be mentioned, not just the first') + }) + + it('reports partial progress when a later batch fails', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const docs = Array.from({ length: 2 }, (_, i) => ({ id: i, data: 'x'.repeat(100) })) + writeFileSync(join(tmpDir, 'data.json'), JSON.stringify(docs)) + + // First batch succeeds, second fails. + const { transport } = mockTransport([successResponse(1), failureResponse(1)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', join(tmpDir, 'data.json'), + '--flush-bytes', '50', + '--retries', '0', + '--json' + ], makeDeps(transport)) as Record + + assert.equal(result.total, 2) + assert.equal(result.succeeded, 1) + assert.equal(result.failed, 1) + assert.ok(result.error != null) + }) + it('returns empty summary for zero documents', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) writeFileSync(join(tmpDir, 'data.json'), '[]') From 1b5102a884fa159a99a4140a569602e1163244bd Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 20:54:11 -0400 Subject: [PATCH 7/9] fix: report enoent eacces as input error not transport error --- src/es/helpers/bulk-ingest.ts | 5 ++++- test/es/helpers/bulk-ingest.test.ts | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index 8b497c85..df0ff030 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -143,6 +143,9 @@ class JsonArraySplitter { } } +/** Local filesystem error codes that mean "the input is broken", not "the cluster is broken". */ +const LOCAL_FS_ERROR_CODES = new Set(['ENOENT', 'EACCES', 'EISDIR']) + /** Returns the default glob pattern for the given source format. */ function defaultGlob (format: SourceFormat): string { if (format === 'csv') return '**/*.csv' @@ -373,7 +376,7 @@ function createBulkIngestHandler (deps: BulkIngestDeps = defaultDeps) { // deciding whether/how to re-run (e.g. with an upsert). const summary = reporter.summary() as Record const code = (err as { code?: string }).code - if (code === 'input_error' || (err instanceof Error && ( + if (code === 'input_error' || (code != null && LOCAL_FS_ERROR_CODES.has(code)) || (err instanceof Error && ( err.message.startsWith('No files matched') || err.message.startsWith('Provide only one') || err.message.startsWith('No input provided') || diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index 511a72bd..987eae95 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -451,6 +451,20 @@ describe('bulk-ingest command', () => { assert.ok(result.error != null) }) + it('reports input_error, not transport_error, for a missing --data-file', async () => { + const { transport } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', '/tmp/does-not-exist-bulk-ingest-test.ndjson', + '--json' + ], makeDeps(transport)) as Record + + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /ENOENT/) + }) + it('returns empty summary for zero documents', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) writeFileSync(join(tmpDir, 'data.json'), '[]') From 4ae80e0bce4d28cb4542d6069b096970ee0d5c1f Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 20:54:55 -0400 Subject: [PATCH 8/9] fix: forward csv stream errors instead of crashing --- src/es/helpers/bulk-ingest.ts | 4 ++++ test/es/helpers/bulk-ingest.test.ts | 15 +++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index df0ff030..b6feb4af 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -298,6 +298,10 @@ async function streamBulkIngest ( return value } }) + // .pipe() doesn't forward source errors to the destination, so a missing + // or unreadable file would otherwise crash the process as an unhandled + // 'error' event instead of surfacing as an input_error. + stream.on('error', (e) => { parser.destroy(e) }) stream.pipe(parser) for await (const record of parser) { await addDoc(JSON.stringify(record)) diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index 987eae95..f8e6ab06 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -594,6 +594,21 @@ describe('bulk-ingest command', () => { assert.ok(body.includes('"bar"')) }) + it('reports input_error instead of crashing on a missing CSV file', async () => { + const { transport } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', '/tmp/does-not-exist-bulk-ingest-test.csv', + '--source-format', 'csv', + '--json' + ], makeDeps(transport)) as Record + + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /ENOENT/) + }) + it('casts numeric and boolean values from CSV', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-csv-')) const filePath = join(tmpDir, 'data.csv') From 46020a28e2e79ec73555a63b35f5d85c20d0acf8 Mon Sep 17 00:00:00 2001 From: margaretjgu Date: Mon, 3 Aug 2026 20:58:02 -0400 Subject: [PATCH 9/9] fix: reject empty data file or stdin as input error --- src/es/helpers/bulk-ingest.ts | 20 ++++++++ test/es/helpers/bulk-ingest.test.ts | 72 +++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) diff --git a/src/es/helpers/bulk-ingest.ts b/src/es/helpers/bulk-ingest.ts index b6feb4af..638e2148 100644 --- a/src/es/helpers/bulk-ingest.ts +++ b/src/es/helpers/bulk-ingest.ts @@ -278,9 +278,22 @@ async function streamBulkIngest ( filePaths = [undefined] } + // A single --data-file or stdin with no non-whitespace bytes is almost + // certainly a mistake (wrong path, empty pipe), so it's an explicit + // input_error rather than a silent "processed 0 docs" success. --data-dir + // has no equivalent check: individual empty files within a batch are fine. + const singleSourceInput = data_dir == null + let sawContent = false + for (const filePath of filePaths) { const stream: Readable = filePath != null ? createReadStream(filePath, { encoding: 'utf-8' }) : process.stdin + if (singleSourceInput) { + stream.on('data', (chunk: string | Buffer) => { + if (!sawContent && chunk.toString().trim().length > 0) sawContent = true + }) + } + if (source_format === 'csv') { const csvColumns = opts.csv_columns != null ? opts.csv_columns.split(',').map(c => c.trim()).filter(Boolean) @@ -347,6 +360,13 @@ async function streamBulkIngest ( } } + if (singleSourceInput && !sawContent) { + const message = data_file != null + ? 'No input data received from file' + : 'No input provided. Use --data-file, --data-dir, or pipe data to stdin' + throw Object.assign(new Error(message), { code: 'input_error' }) + } + await flush() // Drain: acquire all slots to confirm every in-flight batch has finished. diff --git a/test/es/helpers/bulk-ingest.test.ts b/test/es/helpers/bulk-ingest.test.ts index f8e6ab06..7af54dc9 100644 --- a/test/es/helpers/bulk-ingest.test.ts +++ b/test/es/helpers/bulk-ingest.test.ts @@ -465,6 +465,60 @@ describe('bulk-ingest command', () => { assert.match(error.message as string, /ENOENT/) }) + it('rejects an empty --data-file as input_error, not a silent success', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const filePath = join(tmpDir, 'empty.ndjson') + writeFileSync(filePath, '') + + const { transport, requests } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', filePath, + '--json' + ], makeDeps(transport)) as Record + + assert.equal(requests.length, 0) + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /No input data received/) + }) + + it('rejects whitespace-only --data-file as input_error', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const filePath = join(tmpDir, 'whitespace.ndjson') + writeFileSync(filePath, ' \n\n \n') + + const { transport } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', filePath, + '--json' + ], makeDeps(transport)) as Record + + const error = result.error as Record + assert.equal(error.code, 'input_error') + assert.match(error.message as string, /No input data received/) + }) + + it('does not treat a valid empty JSON array `[]` as an input_error', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) + const filePath = join(tmpDir, 'data.json') + writeFileSync(filePath, '[]') + + const { transport } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', filePath, + '--json' + ], makeDeps(transport)) as Record + + assert.equal(result.error, undefined) + assert.equal(result.total, 0) + }) + it('returns empty summary for zero documents', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-')) writeFileSync(join(tmpDir, 'data.json'), '[]') @@ -609,6 +663,24 @@ describe('bulk-ingest command', () => { assert.match(error.message as string, /ENOENT/) }) + it('does not treat a header-only CSV (zero data rows) as input_error', async () => { + const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-csv-')) + const filePath = join(tmpDir, 'headers-only.csv') + writeFileSync(filePath, 'name,age\n') + + const { transport } = mockTransport([successResponse(0)]) + + const result = await runCommand([ + '--index', 'test-idx', + '--data-file', filePath, + '--source-format', 'csv', + '--json' + ], makeDeps(transport)) as Record + + assert.equal(result.error, undefined) + assert.equal(result.total, 0) + }) + it('casts numeric and boolean values from CSV', async () => { const tmpDir = mkdtempSync(join(tmpdir(), 'bulk-test-csv-')) const filePath = join(tmpDir, 'data.csv')