Node.js: Data Warehouse – Streaming ETL Rows Without Buffering Them

export SOURCE_DATABASE_URL="postgres://etl_reader:<password>@<source-host>:5432/app"
export DEST_DATABASE_URL="postgres://etl_writer:<password>@<warehouse-host>:5432/warehouse"
node stream_etl.js --source-table public.orders --dest-table staging.stg_orders --truncate --log-every 50000

Streaming public.orders -> staging.stg_orders (8 columns)
Processed 50,000 rows (heap 21 MB)
Processed 100,000 rows (heap 23 MB)
Done: 137,412 rows in 38.6s, committed

/**
 * stream_etl.js
 *
 * Streams rows from a source Postgres table through a cursor, applies a
 * per-row transform, and bulk-loads the result into a destination table with
 * COPY ... FROM STDIN, without buffering the result set in memory.
 *
 * Reads:  SOURCE_DATABASE_URL, --source-table (SELECT *).
 *         DEST_DATABASE_URL, information_schema.columns for --dest-table.
 * Writes: --dest-table via COPY, inside one transaction (optionally after TRUNCATE).
 *
 * Usage: node stream_etl.js --source-table <schema.table> --dest-table <schema.table>
 *                           [--truncate] [--batch-size 1000] [--log-every 50000]
 */

"use strict";

const { Pool } = require("pg");
const QueryStream = require("pg-query-stream");
const { from: copyFrom } = require("pg-copy-streams");
const { Transform } = require("node:stream");
const { pipeline } = require("node:stream/promises");

// Skip pg's type parsing: every non-null value arrives as the text Postgres sent.
const RAW_TEXT = { getTypeParser: () => (value) => value };

function parseArgs(argv) {
    const args = { truncate: false, "batch-size": "1000", "log-every": "50000" };
    for (let i = 0; i < argv.length; i += 1) {
        const token = argv[i];
        if (!token.startsWith("--")) {
            throw new Error(`Unexpected argument: ${token}`);
        }
        const key = token.slice(2);
        if (key === "truncate") {
            args.truncate = true;
        } else {
            args[key] = argv[i + 1];
            i += 1;
        }
    }
    if (!args["source-table"] || !args["dest-table"]) {
        throw new Error("Usage: node stream_etl.js --source-table <schema.table> --dest-table <schema.table> [--truncate]");
    }
    const batchSize = Number(args["batch-size"]);
    const logEvery = Number(args["log-every"]);
    if (!Number.isInteger(batchSize) || batchSize < 1 || !Number.isInteger(logEvery) || logEvery < 1) {
        throw new Error("--batch-size and --log-every must be positive integers");
    }
    return {
        sourceTable: args["source-table"],
        destTable: args["dest-table"],
        truncate: args.truncate,
        batchSize,
        logEvery,
    };
}

function splitName(name) {
    const parts = name.split(".");
    return parts.length === 2 ? parts : ["public", parts[0]];
}

function quoteName(client, name) {
    return splitName(name).map((part) => client.escapeIdentifier(part)).join(".");
}

async function destinationColumns(client, name) {
    const [schema, table] = splitName(name);
    const { rows } = await client.query(
        `SELECT column_name
           FROM information_schema.columns
          WHERE table_schema = $1 AND table_name = $2
            AND is_generated = 'NEVER'
          ORDER BY ordinal_position`,
        [schema, table],
    );
    if (rows.length === 0) {
        throw new Error(`Destination table ${name} not found or not visible to this role`);
    }
    return rows.map((row) => row.column_name);
}

// One row in, one row out. Values are Postgres text (or null). Extend this for
// renames, filtering, or derived columns; never hold a reference to past rows.
function transformRow(row, loadedAt) {
    return { ...row, loaded_at: loadedAt };
}

// COPY text format: tab-delimited, \N for null, and backslash, newline,
// carriage return, and the delimiter escaped with a backslash.
function toCopyField(value) {
    if (value === null || value === undefined) {
        return "\\N";
    }
    return String(value)
        .replace(/\\/g, "\\\\")
        .replace(/\t/g, "\\t")
        .replace(/\n/g, "\\n")
        .replace(/\r/g, "\\r");
}

async function run() {
    const { sourceTable, destTable, truncate, batchSize, logEvery } = parseArgs(process.argv.slice(2));

    const sourcePool = new Pool({ connectionString: process.env.SOURCE_DATABASE_URL, max: 1 });
    const destPool = new Pool({ connectionString: process.env.DEST_DATABASE_URL, max: 1 });
    const sourceClient = await sourcePool.connect();
    const destClient = await destPool.connect();

    const startedAt = Date.now();
    const loadedAt = new Date().toISOString();
    let rowCount = 0;

    try {
        const columns = await destinationColumns(destClient, destTable);
        const columnList = columns.map((c) => destClient.escapeIdentifier(c)).join(", ");
        console.log(`Streaming ${sourceTable} -> ${destTable} (${columns.length} columns)`);

        await destClient.query("BEGIN");
        if (truncate) {
            await destClient.query(`TRUNCATE ${quoteName(destClient, destTable)}`);
        }

        const source = sourceClient.query(
            new QueryStream(`SELECT * FROM ${quoteName(sourceClient, sourceTable)}`, [], {
                batchSize,
                types: RAW_TEXT,
            }),
        );

        const toCopyLines = new Transform({
            writableObjectMode: true,
            readableObjectMode: false,
            transform(row, _encoding, callback) {
                const out = transformRow(row, loadedAt);
                if (rowCount === 0) {
                    const missing = columns.filter((c) => !(c in out));
                    if (missing.length > 0) {
                        callback(new Error(`Transform does not produce destination column(s): ${missing.join(", ")}`));
                        return;
                    }
                }
                rowCount += 1;
                if (rowCount % logEvery === 0) {
                    const heapMb = Math.round(process.memoryUsage().heapUsed / 1024 / 1024);
                    console.log(`Processed ${rowCount.toLocaleString()} rows (heap ${heapMb} MB)`);
                }
                callback(null, columns.map((c) => toCopyField(out[c])).join("\t") + "\n");
            },
        });

        const sink = destClient.query(
            copyFrom(`COPY ${quoteName(destClient, destTable)} (${columnList}) FROM STDIN`),
        );

        // Propagates backpressure and destroys every stream if any one fails;
        // destroying the COPY stream sends CopyFail so the server aborts the COPY.
        await pipeline(source, toCopyLines, sink);
        await destClient.query("COMMIT");

        const seconds = ((Date.now() - startedAt) / 1000).toFixed(1);
        console.log(`Done: ${rowCount.toLocaleString()} rows in ${seconds}s, committed`);
    } catch (error) {
        await destClient.query("ROLLBACK").catch(() => {});
        throw error;
    } finally {
        sourceClient.release();
        destClient.release();
        await Promise.all([sourcePool.end(), destPool.end()]);
    }
}

run().catch((error) => {
    console.error("ETL job failed:", error.message);
    process.exitCode = 1;
});