Streaming a Huge CSV Into a Database From Node the Fast Way

A common first design for "import this big CSV" is a Node script that reads a row, parses it and inserts it. It is easy to write, it works on a test file of 1,000 rows, and on 2,000,000 rows it takes the better part of an hour or falls over. The speed depends almost entirely on how you talk to the database, and much less on how you read the file. This article loads the same file into PostgreSQL and SQLite from Node four ways each, with timings, and covers what happens when the file has a bad row. The reading side of the stream is in the streams article, the pg-copy-streams package is described in its repository and COPY in the PostgreSQL documentation, and the same loads in the databases' own tools are in the PostgreSQL guide and the SQLite guide.

PostgreSQL: four ways to get 2,000,000 rows in

The target table has the file's eight columns with real types, int, numeric(10,2) and timestamp. The four loaders:

import fs from "node:fs";
import pg from "pg";
import { from as copyFrom } from "pg-copy-streams";
import { pipeline } from "node:stream/promises";
import { Transform } from "node:stream";
import { parse } from "csv-parse";
import { stringify } from "csv-stringify";

const mode = process.argv[2];
const limit = Number(process.argv[3] || 0);
const src = "/batches/csv-0001-0200/work/orders-2m.csv";
const client = new pg.Client({ host: "pg", user: "postgres", password: "secret", database: "lab" });
await client.connect();
await client.query("DROP TABLE IF EXISTS orders_node");
await client.query(`CREATE TABLE orders_node (order_id int, customer_id int, zip text, amount numeric(10,2), created_at timestamp, status text, country text, note text)`);
const t = performance.now();
let n = 0;
if (mode === "copy-direct") {
  await pipeline(fs.createReadStream(src), client.query(copyFrom("COPY orders_node FROM STDIN WITH (FORMAT csv, HEADER true)")));
} else if (mode === "copy-through-js") {
  // parse and validate in JS, then re-serialize into COPY
  await pipeline(
    fs.createReadStream(src), parse({ from_line: 2 }),
    new Transform({ objectMode: true, transform(r, _e, cb) {
      if (limit && n >= limit) return cb();
      if (!Number.isFinite(Number(r[3]))) return cb(new Error(`bad amount on record ${n + 1}`));
      n++; cb(null, r);
    } }),
    stringify(),
    client.query(copyFrom("COPY orders_node FROM STDIN WITH (FORMAT csv)")));
} else if (mode === "insert-per-row") {
  let i = 0;
  for await (const r of fs.createReadStream(src).pipe(parse({ from_line: 2 }))) {
    await client.query("INSERT INTO orders_node VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", r);
    if (++i === limit) break;
  }
  n = i;
} else if (mode === "insert-batched") {
  let batch = [], i = 0;
  const flush = async () => {
    if (!batch.length) return;
    const params = [], values = batch.map((r, k) => { params.push(...r); return `(${r.map((_, j) => `$${k * 8 + j + 1}`).join(",")})`; });
    await client.query(`INSERT INTO orders_node VALUES ${values.join(",")}`, params);
    batch = [];
  };
  for await (const r of fs.createReadStream(src).pipe(parse({ from_line: 2 }))) {
    batch.push(r); i++;
    if (batch.length === 1000) await flush();
    if (i === limit) break;
  }
  await flush();
  n = i;
}
const dt = (performance.now() - t) / 1000;
const { rows: [{ c, s }] } = await client.query("SELECT count(*) c, round(sum(amount),2) s FROM orders_node");
n = Number(c);
console.log(`${mode.padEnd(16)} ${String(n).padStart(8)} rows in ${dt.toFixed(2)}s (${Math.round(n / dt).toLocaleString()} rows/s), sum ${s}, rss ${(process.resourceUsage().maxRSS / 1024).toFixed(0)} MB`);
await client.end();

node load_pg.mjs insert-per-row 20000
node load_pg.mjs insert-batched
node load_pg.mjs copy-through-js
node load_pg.mjs copy-direct

insert-per-row      20000 rows in 5.05s (3,958 rows/s), sum 1601049.63, rss 102 MB
insert-batched    2000000 rows in 21.01s (95,215 rows/s), sum 160174646.36, rss 229 MB
copy-through-js   2000000 rows in 33.43s (59,830 rows/s), sum 160174646.36, rss 111 MB
copy-direct       2000000 rows in 2.32s (861,282 rows/s), sum 160174646.36, rss 107 MB

• **insert-per-row**: one INSERT with parameters for each record, awaiting each. 20,000 rows took 5.05 s: 3,958 rows per second. At that rate the 2,000,000 rows would take about 8 minutes, and I did not run the full file. Each insert is a network round trip and a transaction, and the file reading is not the bottleneck.

• **insert-batched**: 1,000 records per INSERT ... VALUES (...),(...) with numbered parameters. 2,000,000 rows in 21.01 s, 95,215 per second. The limit on a batch is the 65,535 parameters a single statement may carry, and with 8 columns that caps a batch at 8,191 rows; 1,000 is comfortably below.

• **copy-through-js**: the parser reads the file, a Transform validates the amount, csv-stringify writes the records back out, and the result goes into COPY ... FROM STDIN. 33.43 s, 59,830 per second. This is slower than the batched insert, because every row is parsed and serialized in JavaScript on its way through. What you get for the cost is control: a check per row, a chance to reshape or drop records, and an error that names the record.

• **copy-direct**: the read stream goes straight into the COPY stream, with no JavaScript touching a row. 2.32 s, 861,282 per second, 107 MB of memory. The server parses the CSV, and it is the same path psql \copy takes.

All four produced the same sum of the amount column, 160174646.36, which is the check that nothing was lost or duplicated. The direct COPY was nine times faster than batched inserts and fourteen times faster than going through JavaScript. If a file is clean enough to load as is, that is the way. It is also the place to read about COPY's CSV options, since the quoting and NULL handling are the server's.

The cost of validating in JavaScript

The slow path buys you something real, so the question is whether you need it. The server's parser is strict about structure and weak on meaning. Given a row with an extra field:

import fs from "node:fs";
import pg from "pg";
import { from as copyFrom } from "pg-copy-streams";
import { pipeline } from "node:stream/promises";

const client = new pg.Client({ host: "pg", user: "postgres", password: "secret", database: "lab" });
await client.connect();
await client.query("DROP TABLE IF EXISTS orders_node");
await client.query(`CREATE TABLE orders_node (order_id int, customer_id int, zip text, amount numeric(10,2), created_at timestamp, status text, country text, note text)`);
try {
  await pipeline(fs.createReadStream("/batches/csv-0001-0200/work/bad-orders.csv"), client.query(copyFrom("COPY orders_node FROM STDIN WITH (FORMAT csv, HEADER true)")));
  console.log("loaded");
} catch (e) {
  console.log(`${e.code} ${e.message}`);
  console.log("where:", e.where);
}
const { rows: [{ c }] } = await client.query("SELECT count(*) c FROM orders_node");
console.log("rows in table after the failure:", c);
await client.end();

node load_pg_bad.mjs

22P04 extra data after last expected column
where: COPY orders_node, line 50: "49,169480,45194,91.37,2025-04-11 03:28:57,shipped,IN,rush order,extra"
rows in table after the failure: 0

The COPY stopped with 22P04 extra data after last expected column and the server's context named the line and its content: COPY orders_node, line 50: "49,169480,...,rush order,extra". A single COPY is atomic, so the table held 0 rows afterward: nothing partial to clean up. That is a good failure mode for a data load. It also means one bad row at line 1,999,000 costs you the whole run, and the message gives the first problem and not the list.

The pattern that gets both is a staging table: load everything as text with a direct COPY, then validate and convert in SQL, with the rejected rows visible in a query. That is the approach in the article on loading CSV through a staging table, and it keeps the speed of the direct path. Use the JavaScript path when the validation is something SQL cannot say, such as a call to another service, or when the rows need reshaping.

SQLite from Node

Node 24 has a built-in SQLite module, node:sqlite, which needs no dependency. It is synchronous, so there is no network round trip, and the cost that matters is the transaction. Four loaders over the same stream, into a file on disk:

import fs from "node:fs";
import { DatabaseSync } from "node:sqlite";
import { parse } from "csv-parse";

const mode = process.argv[2];
const limit = Number(process.argv[3] || 0);
fs.rmSync("/tmp/orders.db", { force: true });
fs.rmSync("/tmp/orders.db-wal", { force: true });
fs.rmSync("/tmp/orders.db-shm", { force: true });
const db = new DatabaseSync("/tmp/orders.db");
if (mode === "per-row-wal") db.exec("PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL");
db.exec(`CREATE TABLE orders (order_id INTEGER, customer_id INTEGER, zip TEXT, amount REAL, created_at TEXT, status TEXT, country TEXT, note TEXT)`);
const insert = db.prepare("INSERT INTO orders VALUES (?,?,?,?,?,?,?,?)");
const toRow = (r) => [Number(r[0]), Number(r[1]), r[2], Number(r[3]), r[4], r[5], r[6], r[7]];
const t = performance.now();
let n = 0;
const parser = fs.createReadStream("/batches/csv-0001-0200/work/orders-2m.csv").pipe(parse({ from_line: 2 }));
if (mode === "per-row" || mode === "per-row-wal") {   // one implicit transaction per insert
  for await (const r of parser) { insert.run(...toRow(r)); if (++n === limit) break; }
} else if (mode === "one-transaction") {
  db.exec("BEGIN");
  for await (const r of parser) { insert.run(...toRow(r)); if (++n === limit) break; }
  db.exec("COMMIT");
} else if (mode === "batched") {    // commit every 50,000 rows
  db.exec("BEGIN");
  for await (const r of parser) {
    insert.run(...toRow(r)); n++;
    if (n % 50000 === 0) { db.exec("COMMIT"); db.exec("BEGIN"); }
    if (n === limit) break;
  }
  db.exec("COMMIT");
}
const dt = (performance.now() - t) / 1000;
const { c, s } = db.prepare("SELECT count(*) c, round(sum(amount),2) s FROM orders").get();
console.log(`${mode.padEnd(16)} ${String(n).padStart(8)} rows in ${dt.toFixed(2)}s (${Math.round(n / dt).toLocaleString()} rows/s), table has ${c}, sum ${s}`);

node --no-warnings load_sqlite.mjs per-row 5000
node --no-warnings load_sqlite.mjs per-row-wal 5000
node --no-warnings load_sqlite.mjs one-transaction
node --no-warnings load_sqlite.mjs batched
ls -l /tmp/orders.db | awk '{printf "database file: %.0f MB\n", $5/1e6}'

per-row              5000 rows in 10.20s (490 rows/s), table has 5000, sum 401205.61
per-row-wal          5000 rows in 0.12s (40,842 rows/s), table has 5000, sum 401205.61
one-transaction   2000000 rows in 7.20s (277,915 rows/s), table has 2000000, sum 160174646.36
batched           2000000 rows in 6.52s (306,872 rows/s), table has 2000000, sum 160174646.36
database file: 139 MB

• **per-row** (autocommit, 5,000 rows): 10.20 s, 490 rows per second. SQLite commits every statement to disk by default, and each commit waits for the file system to confirm the write.

• **per-row-wal** (5,000 rows, journal_mode = WAL, synchronous = NORMAL): 0.12 s, 40,842 rows per second. Changing how the commit is made, with the same loop, is an 83 times difference. Those settings trade some durability on power loss for speed, which is right for an import you can rerun.

• **one-transaction**: BEGIN before the loop, COMMIT after. 2,000,000 rows in 7.20 s, 277,915 per second.

• **batched**: a commit every 50,000 rows. 6.52 s, 306,872 per second. The batched form costs about the same as one transaction, and keeps the transaction (and the journal) bounded, which matters when the file is 100 times larger or the process may be killed.

The database file was 139 MB. All three full loads gave the same count and the same sum, 160174646.36. A prepared statement, created once and reused with run(), is part of the speed, and what you must not do is build the SQL string per row.

An earlier run with an in-memory database showed 109,000 rows per second for the autocommit loop, because there was no disk to wait for. That is a reminder to benchmark against the database you will deploy, not the one that is convenient.

Backpressure, in both

The reader produces rows faster than most databases accept them. In the PostgreSQL loaders, pipeline and await client.query keep the two in step, so the file is read as fast as the database accepts it and memory stays flat: 107 MB for the direct COPY, 111 MB through JavaScript, 229 MB for the batched insert, which holds a 1,000-row array and its parameter list. In the SQLite loop, for await pulls one record at a time. If you wrote the producer as an event handler instead, rows would pile up in memory while the database worked, and a 2,000,000 row file would not fit.

How to tell it worked

Compare the row count and a sum with the file: SELECT count(*), sum(amount) against qsv count and a DuckDB sum on the CSV. Check the types of the columns that matter; text loaded into a numeric column is rejected by the server, and into SQLite's untyped columns is not, as the SQLite types article explains. Count the rejected rows if you logged any. Read the table's definition, and run a query that would show a leading zero lost or a timestamp read in the wrong zone. And rerun the load on a fresh table and check the same numbers appear, since an import you cannot repeat is a liability.

When parsing in JavaScript is the wrong layer

If the file is clean and the schema is known, the database's bulk loader is faster than anything you can write, and the Node script should only open the stream and report the result. If the file needs cleaning that SQL can do, load it as text and clean it in the database. Parse in JavaScript when the logic really is in JavaScript: the rows come from an upload and must be checked against application rules, or each record triggers other work. And whatever the path, size the batch and the transaction so that a failure costs minutes and not hours, and keep the original file until the counts match.