Потоковая выгрузка из PostgreSQL в CSV: серверный курсор и COPY

В прошлый раз я разобрал самые такие три жизненных примера работы с выгрузками.

Наивный способ через память, постраничку через OFFSET/LIMIT и через keyset курсор. Но было бы несправедливо остановиться на этом и не протестировать другие оставшиеся варианты. Два способа, которые я хочу рассмотреть - это серверный курсор pg-query-stream и выгрузка напрямую из базы через COPY. В этой статье дальше будут примеры с кодом, но сам по себе код не информативен, самое главное - что я прогоню эти два новых варианта через тот же бенчмарк, чтобы сравнить нагрузку на память и время выполнения.

Стек тот же: Node.js, Fastify, Knex поверх PostgreSQL.

Три прикладных способа — коротко

Для контекста и замеров напомню, с чем сравниваем.

Всё в памяти. Прочитать весь результат в массив, собрать одну большую строку, отдать. Просто и неприемлемо на больших объемах данных.

Поток с keyset-пагинацией. Читать блоками по тысяче строк, каждый следующий блок — WHERE (createdAt, id) < (курсор) по уникальному ключу. Память постоянная: в каждый момент живёт один блок. Каждый блок — атомарный короткий запрос. соединение между ними освобождается. Строки могут измениться между блоками и важно выбрать такой курсор, чтобы это не мешало.

Поток с OFFSET/LIMIT. Та же потоковая отдача, но постраничка работает через OFFSET. Память тоже постоянная, зато время растёт: чем дальше мы идем по строкам, тем больше на это времени уходит у базы.

Общее у всех трёх способов - что все разруливается сугубо в рантайме приложения. А вот ниже будет два способа, где основную работу берет на себя Postgres.

Серверный курсор

Вместо сотен независимых запросов за блоками можно открыть один серверный курсор и тянуть из него строки по мере готовности. В Postgres это делается через pg-query-stream. Он использует серверный курсор PostgreSQL и получает результат порциями через Readable-поток. На время чтения курсор удерживает соединение с базой.

Через чистый pg-query-stream пришлось бы писать SQL руками и потерять конструктор запросов. Но в Knex для этого есть .stream(), который на pg использует тот же pg-query-stream под капотом, а фильтры и сортировка работают те же. (https://knexjs.org/guide/interfaces.html#streams)

import { Readable } from 'node:stream';

const rows = db('records')
    .where(filters)
    .orderBy('createdAt', 'desc')
    .orderBy('id', 'desc')
    .stream();

async function* toCsv() {
    yield headerLine();
    try {
        for await (const row of rows) {
            yield formatRow(row);
        }
    } finally {
        rows.destroy();
    }
}

reply.send(Readable.from(toCsv()));

for await с потоком даёт backpressure, который ограничивает скорость чтения и накопление данных в буферах. Следующая порция строк из курсора запрашивается только если буфер не переполнен. CSV по-прежнему собирается в JS теми же formatRow/headerLine, поэтому вывод совпадает с ручными потоковыми вариантами байт в байт.

Плата — транзакция и соединение держатся открытыми всю выгрузку. У keyset соединение освобождается между блоками, и на пуле веб-сервера это заметно. Сотня параллельных долгих экспортов через курсор займёт сотню соединений на всё время скачивания (хотя если у вас проект с 100 параллельных выгрузок, это уже что-то очень очень серьезное). Самый главный плюс - что курсор читает результат одного запроса из согласованного снимка. Изменения, сделанные после открытия курсора, не попадут в его выдачу. Это свойство курсора, а не просто факт наличия транзакции: при обычном READ COMMITTED отдельные запросы в одной транзакции могут видеть разные снимки. В конце кода finally с destroy() обязателен - без него оборванное на середине скачивание оставит курсор и соединение висеть до таймаута.

COPY TO STDOUT

И остается один совершенно иной вариант. Собственно если нужно получить из Postgres CSV, то пусть Postgre сам его и формирует. Команда COPY (SELECT ...) TO STDOUT WITH (FORMAT csv) форматирует результат внутри базы, без единого объекта-строки в Node. Поток байт идёт из сокета БД as is. Библиотека pg-copy-streams оборачивает это в Readable:

import { to as copyTo } from 'pg-copy-streams';

const selectSql = db('records')
    .select(COLUMNS.map((c) => `records.${c}`))
    .where(filters)
    .orderBy('createdAt', 'desc')
    .orderBy('id', 'desc')
    .toString();

const copySql = `COPY (${selectSql}) TO STDOUT WITH (FORMAT csv, HEADER true, DELIMITER ';')`;

const connection = await db.client.acquireConnection();
const stream = connection.query(copyTo(copySql));

let released = false;
stream.once('close', () => {
    if (released) return;
    released = true;
    void db.client.releaseConnection(connection);
});

reply.raw.once('close', () => {
    if (!reply.raw.writableFinished) stream.destroy();
});

reply.send(stream);

COPY работает на сыром pg-соединении - берём его из пула Knex и возвращаем после. И в конце освобождаем соединение после закрытия COPY-потока. Это будет ожидаемо быстрее, в конце я еще замеряю обязательно через бенч. И важно отметить, что COPY тоже дает согласованный целостный экспорт данных. Но несмотря на скорость есть ряд негативных моментов. Во первых здесь тоже будет одно долгое соединение, как и в предыдущем варианте.

Билдер остаётся только для сборки SELECT, дальше идет сырой SQL. COPY не принимает bind-параметры, поэтому значения фильтров попадают в текст команды через .toString(). Knex экранирует их, но это ровно то место, где нужно следить за sql иньекциями и правильным форматированием запроса. За сырым SQL и toString важно следить, чтобы в него не попало ничего постороннего с фронта.

Формат выгрузки - на стороне Postgre. Разделитель, кавычки и заголовок задаются опциями COPY, но BOM для Excel, даты в нужном виде, нейтрализацию формул (=, +, -, @) и любые преобразования значений придётся делать отдельным преобразованием поверх потока. Вывод COPY не совпадает с ручным. Boolean приходят как t/f, время — в формате PostgreSQL со смещением, а не ISO с Z. Если файл открывают люди в Excel, одними опциями COPY не обойтись.

Ну и сырое соединение, с которым нужно аккуратно работать. Например, если connection.query() выбросит синхронную ошибку, соединение не будет возвращено в пул. То есть напрямую брать этот код в production явно не стоит.

Бенч

Я собрал все пять способов в одном API и прогнал на таблице примерно в 340 тысяч строк - те что накопились в локальной базе проекта. CSV файл выгрузки весит около 37 МБ. Чтобы память одного метода не тянулась в замер следующего, каждый вариант запускался в отдельном свежем процессе: поднял инстанс, сделал один запрос, погасил. Пик памяти снимаю на сервере - heapUsed и rss каждые 25 мс, с global.gc() перед каждым прогоном (--expose-gc).

Все замеры можно считать примерными - я их прогонял буквально несколько раз, но пропорции при каждом прогоне были примерно одинаковыми. В этом прогоне COPY быстрее всех с большим отрывом- и при этом легче всех по памяти: формирование CSV не нагружает память Node. Серверный курсор идёт вровень с keyset по памяти и чуть чуть быстрее него по времени. OFFSET отстаёт по времени в пять раз, оставаясь при этом экономным по памяти.

Несколько оговорок, без которых таблицу легко прочитать неверно.

Четыре первых метода, собирающие CSV в JS, дали байт в байт одинаковый файл - можно ради интереса взять чексумму - он будет одинаковой. COPY же отличается: другое форматирование, нет BOM, свой формат булевых и дат. Поэтому его я сравнивал по времени и памяти, а не по хэшу; «тот же CSV» про него говорить нельзя. Можно отдельным этапом пробовать его приводить к тому же формату - но это будут еще десятки строк кода и не совсем по теме статьи.

Что же выбрать? Если CSV уходит в машинную обработку и своё форматирование и красоты в csv не нужны - можно брать COPY. Есть определенные минусы по коду, но этот вариант самый быстрый. Если нужен согласованный снимок данных, подходят и серверный курсор, и COPY. Курсор удобнее, когда требуется собственное форматирование CSV в Node.js, а COPY - когда достаточно возможностей Postgres. Много параллельных выгрузок будут создавать много одновременных долгих соединений. keyset остаётся разумной серединой, когда снимок не нужен, а соединение в пуле держать не хочется. Например при выгрузке по ID или по Времени + ID если данные не меняются и вы выгружаете какие-нибудь логи или иные временные данные. OFFSET проигрывает keyset по времени но как раз такие его обычно реализуют первым из-за простоты и прозрачности.