Потоковая выгрузка из 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).
Метод
Время
peak heapUsed
peak RSS
всё в памяти
2.4 с
263 МБ
761 МБ
keyset-поток
2.7 с
66 МБ
206 МБ
OFFSET-поток
13.8 с
63 МБ
201 МБ
серверный курсор
2.5 с
60 МБ
185 МБ
COPY
0.56 с
39 МБ
179 МБ
Все замеры можно считать примерными — я их прогонял буквально несколько раз, но пропорции при каждом прогоне были примерно одинаковыми. В этом прогоне COPY быстрее всех с большим отрывом- и при этом легче всех по памяти: формирование CSV не нагружает память Node. Серверный курсор идёт вровень с keyset по памяти и чуть чуть быстрее него по времени. OFFSET отстаёт по времени в пять раз, оставаясь при этом экономным по памяти.
Несколько оговорок, без которых таблицу легко прочитать неверно.
Четыре первых метода, собирающие CSV в JS, дали байт в байт одинаковый файл — можно ради интереса взять чексумму — он будет одинаковой. COPY же отличается: другое форматирование, нет BOM, свой формат булевых и дат. Поэтому его я сравнивал по времени и памяти, а не по хэшу; «тот же CSV» про него говорить нельзя. Можно отдельным этапом пробовать его приводить к тому же формату — но это будут еще десятки строк кода и не совсем по теме статьи.
Что же выбрать? Если CSV уходит в машинную обработку и своё форматирование и красоты в csv не нужны — можно брать COPY. Есть определенные минусы по коду, но этот вариант самый быстрый. Если нужен согласованный снимок данных, подходят и серверный курсор, и COPY. Курсор удобнее, когда требуется собственное форматирование CSV в Node.js, а COPY — когда достаточно возможностей Postgres. Много параллельных выгрузок будут создавать много одновременных долгих соединений. keyset остаётся разумной серединой, когда снимок не нужен, а соединение в пуле держать не хочется. Например при выгрузке по ID или по Времени + ID если данные не меняются и вы выгружаете какие-нибудь логи или иные временные данные. OFFSET проигрывает keyset по времени но как раз такие его обычно реализуют первым из-за простоты и прозрачности.
Источник: habr.com