pg-copy-streams
Advanced tools
Low-Level COPY TO and COPY FROM streams for PostgreSQL in JavaScript using
The pg-copy-streams npm package provides a way to perform high-performance data import and export operations with PostgreSQL using the COPY command. It allows you to stream data to and from PostgreSQL databases efficiently.
COPY FROM
This feature allows you to import data into a PostgreSQL table from a file using the COPY FROM command. The code sample demonstrates how to stream data from a CSV file into a PostgreSQL table.
const { Client } = require('pg');
const copyFrom = require('pg-copy-streams').from;
const fs = require('fs');
const client = new Client();
client.connect();
const stream = client.query(copyFrom('COPY my_table FROM STDIN'));
const fileStream = fs.createReadStream('data.csv');
fileStream.pipe(stream).on('finish', () => {
console.log('Data imported successfully');
client.end();
});COPY TO
This feature allows you to export data from a PostgreSQL table to a file using the COPY TO command. The code sample demonstrates how to stream data from a PostgreSQL table into a CSV file.
const { Client } = require('pg');
const copyTo = require('pg-copy-streams').to;
const fs = require('fs');
const client = new Client();
client.connect();
const stream = client.query(copyTo('COPY my_table TO STDOUT'));
const fileStream = fs.createWriteStream('data.csv');
stream.pipe(fileStream).on('finish', () => {
console.log('Data exported successfully');
client.end();
});The 'pg' package is the official PostgreSQL client for Node.js. While it does not provide streaming capabilities for the COPY command out of the box, it can be used in conjunction with other libraries to achieve similar functionality. It is a more general-purpose library for interacting with PostgreSQL databases.
The 'pg-promise' package is a PostgreSQL interface for Node.js that provides a more powerful and flexible API compared to the 'pg' package. It supports a wide range of features, including transactions and connection pooling, but does not natively support streaming for the COPY command. However, it can be extended to work with streaming libraries.
The 'pg-boss' package is a job queue built on top of PostgreSQL. While it is not directly related to streaming data with the COPY command, it provides a way to manage background jobs and tasks using PostgreSQL as the backend. It is useful for scenarios where you need to process large amounts of data asynchronously.
COPY FROM / COPY TO for node-postgres. Stream from one database to another, and stuff.
Did you know that PostgreSQL supports streaming data directly into and out of a table? This means you can take your favorite CSV or TSV file and pipe it directly into an existing PostgreSQL table.
PostgreSQL supports text, csv/tsv and binary data. If you have data in another format (say for example JSON) convert it to one of the supported format and pipe it directly into an existing PostgreSQL table !
You can also take a table and pipe it directly to a file, another database, stdout, even to /dev/null if you're crazy!
What this module gives you is a Readable or Writable stream directly into/out of a table in your database. This mode of interfacing with your table is very fast and very brittle. You are responsible for properly encoding and ordering all your columns. If anything is out of place PostgreSQL will send you back an error. The stream works within a transaction so you wont leave things in a 1/2 borked state, but it's still good to be aware of.
If you're not familiar with the feature (I wasn't either) you can read this for some good helps: http://www.postgresql.org/docs/9.6/static/sql-copy.html
var { Pool } = require('pg')
var { to as copyTo } = require('pg-copy-streams')
var pool = new Pool()
pool.connect(function (err, client, done) {
var stream = client.query(copyTo('COPY my_table TO STDOUT'))
stream.pipe(process.stdout)
stream.on('end', done)
stream.on('error', done)
})
// async/await
import { pipeline } from 'node:stream/promises'
import pg from 'pg'
import { to as copyTo } from 'pg-copy-streams'
const pool = new pg.Pool()
const client = await pool.connect()
try {
const stream = client.query(copyTo('COPY my_table TO STDOUT'))
await pipeline(stream, process.stdout)
} finally {
client.release()
}
await pool.end()
Important: When copying data out of postgresql, postgresql will chunk the data on 64kB boundaries. You should expect rows to be cut across the boundaries of these chunks (the end of a chunk will not always match the end of a row). If you are piping the csv output of postgres into a file, this might not be a problem. But if you are trying to analyse the csv output on-the-fly, you need to make sure that you correctly discover the lines of the csv output across the chunk boundaries. We are not recommending any specific streaming csv parser but csv-parser and csv-parse seem to correctly handle this.
var fs = require('node:fs')
var { Pool } = require('pg')
var { from as copyFrom } = require('pg-copy-streams')
var pool = new Pool()
pool.connect(function (err, client, done) {
var stream = client.query(copyFrom('COPY my_table FROM STDIN'))
var fileStream = fs.createReadStream('some_file.tsv')
fileStream.on('error', done)
stream.on('error', done)
stream.on('finish', done)
fileStream.pipe(stream)
})
// async/await
import { pipeline } from 'node:stream/promises'
import fs from 'node:fs'
import pg from 'pg'
import { from as copyFrom } from 'pg-copy-streams'
const pool = new pg.Pool()
const client = await pool.connect()
try {
const ingestStream = client.query(copyFrom('COPY my_table FROM STDIN'))
const sourceStream = fs.createReadStream('some_file.tsv')
await pipeline(sourceStream, ingestStream)
} finally {
client.release()
}
await pool.end()
Note: In version prior to 4.0.0, when copying data into postgresql, it was necessary to wait for the 'end' event of pg-copy-streams.from to correctly detect the end of the COPY operation. This was necessary due to the internals of the module but non-standard. This is not true for versions including and after 4.0.0. The end of the COPY operation must now be detected via the standard 'finish' event. Users of 4.0.0+ should not wait for the 'end' event because it is not fired anymore.
In version 6.0.0+, If you have not yet finished ingesting data into a copyFrom stream and you want to ask postgresql to abort the process, you can call destroy() on the stream (or let pipeline do it for you if it detects an error in the pipeline). This will send a CopyFail message to the backend that will rollback the operation. Please take into account that this will not revert the operation if the CopyDone message has already been sent and is being processed by the backend.
This is a more advanded topic. Check the test/copy-both.js file for an example of how this can be used.
Note regarding logical decoding: Parsers for logical decoding scenarios are easier to write when copy-both.js pushes chunks that are aligned on the copyData protocol frames. This is not the default mode of operation of copy-both.js in order to increase the streaming performance. If you need the pushed chunks to be aligned on copyData frames, use the alignOnCopyDataFrame: true option.
$ npm install pg-copy-streams
This module only works with the pure JavaScript bindings. If you're using require('pg').native please make sure to use normal require('pg') or require('pg.js') when you're using copy streams.
Before you set out on this magical piping journey, you really should read this: http://www.postgresql.org/docs/current/static/sql-copy.html, and you might want to take a look at the tests to get an idea of how things work.
Take note of the following warning in the PostgreSQL documentation:
COPY stops operation at the first error. This should not lead to problems in the event of a COPY TO, but the target table will already have received earlier rows in a COPY FROM. These rows will not be visible or accessible, but they still occupy disk space. This might amount to a considerable amount of wasted disk space if the failure happened well into a large copy operation. You might wish to invoke VACUUM to recover the wasted space.
The COPY command is commonly used to move huge sets of data. This can put some pressure on the node.js loop, the amount of CPU or the amount of memory used.
There is a bench/ directory in the repository where benchmark scripts are stored. If you have performance issues with pg-copy-stream do not hesitate to write a new benchmark that highlights your issue. Please avoid to commit huge files (PR won't be accepted) and find other ways to generate huge datasets.
If you have a local instance of postgres on your machine, you can start a benchmark for example with
$ cd bench
$ PGPORT=5432 PGDATABASE=postgres node copy-from.js
In order to launch the test suite, you need to have a local instance of postgres running on your machine.
Since version 5.1.0 and the implementation of copy-both.js for logical decoding scenarios, your local postgres instance will need to be configured to accept replication scenarios :
postgresql.conf
wal_level = logical
max_wal_senders > 0
max_replication_slots > 0
pg_hba.conf
make sure your user can connect using the replication mode
$ PGPORT=5432 PGDATABASE=postgres make test
Instead of adding a bunch more code to the already bloated node-postgres I am trying to make the internals extensible and work on adding edge-case features as 3rd party modules. This is one of those.
Please, if you have any issues with this, open an issue.
Better yet, submit a pull request. I love pull requests.
Generally how I work is if you submit a few pull requests and you're interested I'll make you a contributor and give you full access to everything.
Since this isn't a module with tons of installs and dependent modules I hope we can work together on this to iterate faster here and make something really useful.
pg optional timeout mechanismpipeline will automatically send a CopyFail message to the backend is a source triggers an error. cf #115This version is a major change because some users of the library may have been using other techniques in order to ask the backend to rollback the current operation.
Bugfix release handling a corner case when an empty stream is piped into copy-from
This version adds a Duplex stream implementation of the PostgreSQL copyBoth mode described on https://www.postgresql.org/docs/9.6/protocol-flow.html. This mode opens the possibility of dealing with replication and logical decoding scenarios.
This version's major change is a modification in the COPY TO implementation. The new implementation now extends Readable while previous version where extending Transform. This should not have an effect on how users use the module but was considered to justify a major version number because even if the test suite coverage is wide, it could have an impact on the streaming dynamics in certain edge cases that are not yet captured by the tests.
Readable instead of TransformThis version's major change is a modification in the COPY FROM implementation. In previous version, copy-from was internally designed as a Transform duplex stream. The user-facing API was writable, and the readable side of the Transform was piped into the postgres connection stream to copy the data inside the database.
This led to an issue because Transform was emitting its 'finish' too early after the writable side was ended. Postgres had not yet read all the data on the readable side and had not confirmed that the COPY operation was finished. The recommendation was to wait for the 'end' event on the readable side which correcly detected the end of the COPY operation and the fact that the pg connection was ready for new queries.
This recommendation worked ok but this way of detecting the end of a writable is not standard and was leading to different issues (interaction with the finished and pipeline API for example)
The new copy-from implementation extends writable and now emits 'finish' with the correct timing : after the COPY operation and after the postgres connection has reached the readyForQuery state.
Another big change in this version is that copy-to now shortcuts the core pg parsing during the COPY operation. This avoids double-parsing and avoids the fact that pg buffers whole postgres protocol messages.
Writable instead of TransformThis version's major change is a modification in the COPY TO implementation. In the previous versions, a row could be pushed downstream only after the full row was gathered in memory. In many cases, rows are small and this is not an issue. But there are some use cases where rows can grow bigger (think of a row containing a 1MB raw image in a BYTEA field. cf issue #91). In these cases, the library was constantly trying to allocate very big buffers and this could lead to severe performance issues. In the new implementation, all the data payload received from a postgres chunk is sent downstream without waiting for full row boundaries.
Some users may in the past have relied on the fact the the copy-to chunk boundaries exactly matched row boundaries. A major difference in the 3.x version is that the module does not offer any guarantee that its chunk boundaries match row boundaries. A row data could (and you have to realize that this will happen) be split across 2 or more chunks depending on the size of the rows and on postgres's own chunking decisions.
As a consequence, when the copy-to stream is piped into a pipeline that does row/CSV parsing, you need to make sure that this pipeline correcly handles rows than span across chunk boundaries. For its tests, this module uses the csv-parser module
prettier configuration following discussion on brianc/node-postgres#2172This version's major change is a modification in the COPY TO implementation. In the previous version, when a chunk was received from the database, it was analyzed and every row contained within that chunk was pushed individually down the stream pipeline. Small rows could lead to a "one chunk" / "thousands of row pushed" performance issue in node. Thanks to @rafatower & CartoDB for the patch. This is considered to be a major change since some people could be relying on the fact that each outgoing chunk is an individual row.
Other changes in this version
The MIT License (MIT)
Copyright (c) 2013 Brian M. Carlson
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
FAQs
Low-Level COPY TO and COPY FROM streams for PostgreSQL in JavaScript using
We found that pg-copy-streams demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 3 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
A malicious npm package disguised as a WhatsApp client is exploiting authentication flows with a remote kill switch to exfiltrate data and destroy files.
Security News
PyPI now supports digital attestations, enhancing security and trust by allowing package maintainers to verify the authenticity of Python packages.
Security News
GitHub removed 27 malicious pull requests attempting to inject harmful code across multiple open source repositories, in another round of low-effort attacks.