DOCSEARCHbyAlgolia
Submit Docs

pg-copy-streams

repository·master·Indexed 43 days ago

https://github.com/brianc/node-pg-copy-streams

Low-level COPY TO and COPY FROM streams for PostgreSQL in JavaScript. Version 7.0.0 provides high-performance Node.js Readable, Writable, and Duplex streams to interface with PostgreSQL's COPY command for streaming text, CSV/TSV, or binary data directly into or out of tables. Requires pure JavaScript bindings for pg.

Tokens
1.6K
Snippets
6
Records
9
Agent score
6%

What's inside pg-copy-streams

  1. Use duplex streams for replication (copyBoth)

    master

    For advanced replication or logical decoding scenarios, use the copyBoth functionality. This provides a duplex stream.

    Logical Decoding Note: By default, copy-both pushes chunks to maximize streaming performance, which may not align with copyData protocol frames. If your parser requires chunks to be aligned on copyData frames, use the following option:

    • alignOnCopyDataFrame: true

  2. How pg-copy-streams works

    master

    The module provides Readable or Writable streams to interface with PostgreSQL's COPY command. This allows for high-speed streaming of data (text, CSV/TSV, or binary) directly into or out of a table.

    Key considerations:

    • Speed vs. Brittleness: This method is extremely fast but brittle. You are responsible for the correct encoding and column ordering. Any mismatch will cause PostgreSQL to return an error.
    • Transactions: The stream works within a transaction, so partial failures won't leave the database in a half-broken state.
    • Chunking: When copying data out of PostgreSQL, data is chunked on 64kB boundaries. Rows may be split across these boundaries. If parsing CSV on-the-fly, use a parser like csv-parser or csv-parse that handles boundary splitting correctly.

  5. Pipe from a file to a table (copyIn)

    master

    Use the from function (aliased as copyFrom) to create a Writable stream to ingest data into a PostgreSQL table.

    Important Notes:

    • Event Handling: For versions 4.0.0 and later, detect the end of the COPY operation using the standard 'finish' event. Do not wait for the 'end' event, as it is no longer fired.
    • Aborting: In version 6.0.0+, you can call stream.destroy() to abort the process. This sends a CopyFail message to the backend to rollback the operation (provided the CopyDone message hasn't already been processed).

    Callback Pattern

    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 Pattern

    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()
    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()

  6. Pipe from a table to stdout (copyOut)

    master

    Use the to function (aliased as copyTo) to create a Readable stream from a PostgreSQL table.

    Important: Do not use pool.query(). You must acquire a client via pool.connect() and call client.query() on that client to support the custom query objects required for streaming.

    Callback Pattern

    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 Pattern

    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()
    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()

  7. Create a duplex stream for replication with copyBoth()

    master

    Use exports.both(txt, options) to create a CopyBothQueryStream. This creates a duplex stream suitable for scenarios like logical decoding or replication where data flows in both directions via a single COPY command.

    const { both } = require('pg-copy-streams');
const stream = both('COPY my_table FROM STDIN TO STDOUT');
// Use the stream with a database client

  8. Create a stream to copy data from a table to a destination with copyTo()

    master

    Use exports.to(txt, options) to create a CopyToQueryStream. This stream is used for COPY ... TO operations, allowing you to pipe data from a PostgreSQL table to a writable stream (like process.stdout or a file).

    const { to } = require('pg-copy-streams');
const stream = to('COPY my_table TO STDOUT');
// Use the stream with a database client

  9. Create a stream to copy data from a source to a table with copyFrom()

    master

    Use exports.from(txt, options) to create a CopyFromQueryStream. This stream is used for COPY ... FROM operations, allowing you to pipe data from a readable stream (like a file or network socket) into a PostgreSQL table.

    const { from } = require('pg-copy-streams');
const stream = from('COPY my_table FROM STDIN');
// Use the stream with a database client