0
# pg-copy-streams
1

2
pg-copy-streams provides low-level streaming interfaces for PostgreSQL's COPY TO and COPY FROM operations. It enables high-performance data transfer directly between Node.js applications and PostgreSQL databases through three specialized stream types: readable streams for exporting data, writable streams for importing data, and duplex streams for bidirectional operations.
3

4
## Package Information
5

6
- **Package Name**: pg-copy-streams
7
- **Package Type**: npm
8
- **Language**: JavaScript
9
- **Installation**: `npm install pg-copy-streams`
10

11
## Core Imports
12

13
CommonJS (Node.js default):
14

15
```javascript
16
const { to, from, both } = require('pg-copy-streams');
17

18
// Alternative destructuring with aliases
19
const { to: copyTo, from: copyFrom, both: copyBoth } = require('pg-copy-streams');
20

21
// Import entire module
22
const pgCopyStreams = require('pg-copy-streams');
23
const copyTo = pgCopyStreams.to;
24
```
25

26
ES Modules (when using "type": "module" in package.json):
27

28
```javascript
29
import { to, from, both } from 'pg-copy-streams';
30

31
// Alternative with aliases
32
import { to as copyTo, from as copyFrom, both as copyBoth } from 'pg-copy-streams';
33

34
// Default import (not available - module uses named exports only)
35
// import pgCopyStreams from 'pg-copy-streams'; // ❌ This won't work
36
```
37

38
## Basic Usage
39

40
```javascript
41
const { Pool } = require('pg');
42
const { to: copyTo, from: copyFrom } = require('pg-copy-streams');
43

44
const pool = new Pool();
45

46
// Export data from PostgreSQL (COPY TO)
47
pool.connect((err, client, done) => {
48
  const stream = client.query(copyTo('COPY my_table TO STDOUT'));
49
  stream.pipe(process.stdout);
50
  stream.on('end', done);
51
  stream.on('error', done);
52
});
53

54
// Import data into PostgreSQL (COPY FROM)
55
const fs = require('fs');
56
pool.connect((err, client, done) => {
57
  const stream = client.query(copyFrom('COPY my_table FROM STDIN'));
58
  const fileStream = fs.createReadStream('data.csv');
59
  fileStream.pipe(stream);
60
  stream.on('finish', done);
61
  stream.on('error', done);
62
});
63
```
64

65
## Architecture
66

67
pg-copy-streams is built around three core components:
68

69
- **Stream Factory Functions**: Entry-point functions (`to`, `from`, `both`) that create specialized stream instances
70
- **Stream Classes**: Three stream implementations extending Node.js stream classes (Readable, Writable, Duplex)
71
- **Protocol Integration**: Low-level PostgreSQL protocol handling for COPY operations with message parsing and connection management
72
- **Node-postgres Integration**: Seamless integration with the `pg` library connection and query system
73

74
## Capabilities
75

76
### COPY TO Operations (Data Export)
77

78
Creates readable streams for exporting data from PostgreSQL tables to external destinations. Ideal for backup operations, data extraction, and ETL processes.
79

80
```javascript { .api }
81
/**
82
 * Creates a readable stream for COPY TO operations
83
 * @param {string} text - SQL COPY TO statement
84
 * @param {object} [options] - Optional stream configuration 
85
 * @returns {CopyToQueryStream} Readable stream instance
86
 */
87
function to(text, options);
88
```
89

90
[COPY TO Operations](./copy-to.md)
91

92
### COPY FROM Operations (Data Import)
93

94
Creates writable streams for importing data from external sources into PostgreSQL tables. Perfect for bulk data loading, CSV imports, and ETL pipelines.
95

96
```javascript { .api }
97
/**
98
 * Creates a writable stream for COPY FROM operations
99
 * @param {string} text - SQL COPY FROM statement
100
 * @param {object} [options] - Optional stream configuration
101
 * @returns {CopyFromQueryStream} Writable stream instance
102
 */
103
function from(text, options);
104
```
105

106
[COPY FROM Operations](./copy-from.md)
107

108
### COPY BOTH Operations (Bidirectional Streaming)
109

110
Creates duplex streams for bidirectional data operations, primarily used for replication and logical decoding scenarios.
111

112
```javascript { .api }
113
/**
114
 * Creates a duplex stream for COPY BOTH operations
115
 * @param {string} text - SQL COPY statement
116
 * @param {object} [options] - Optional stream configuration
117
 * @returns {CopyBothQueryStream} Duplex stream instance
118
 */
119
function both(text, options);
120
```
121

122
[COPY BOTH Operations](./copy-both.md)
123

124
## Types
125

126
```javascript { .api }
127
/**
128
 * @typedef {object} StreamOptions
129
 * @property {number} [highWaterMark] - Stream high water mark
130
 * @property {boolean} [objectMode] - Object mode flag
131
 * @property {string} [encoding] - Stream encoding
132
 */
133

134
/**
135
 * @typedef {StreamOptions} CopyBothOptions
136
 * @property {boolean} [alignOnCopyDataFrame] - Whether to align data on COPY data frame boundaries
137
 */
138

139
/**
140
 * CopyToQueryStream - Readable stream for COPY TO operations
141
 * @class
142
 * @extends {Readable}
143
 */
144
class CopyToQueryStream {
145
  /**
146
   * @param {string} text - The SQL COPY TO statement
147
   * @param {StreamOptions} [options] - Stream options
148
   */
149
  constructor(text, options) {}
150
  
151
  /** @type {string} The SQL COPY statement */
152
  text;
153
  
154
  /** @type {number} Number of rows processed */
155
  rowCount;
156
  
157
  /** @type {object} PostgreSQL connection reference */
158
  connection;
159
  
160
  /**
161
   * Submits the query to a PostgreSQL connection
162
   * @param {object} connection - pg connection object
163
   */
164
  submit(connection) {}
165
  
166
  /**
167
   * Handles errors during stream processing
168
   * @param {Error} err - Error object
169
   */
170
  handleError(err) {}
171
  
172
  /**
173
   * Processes CommandComplete message and extracts row count
174
   * @param {object} msg - Message object with text property
175
   */
176
  handleCommandComplete(msg) {}
177
  
178
  /**
179
   * Called when connection is ready for next query
180
   */
181
  handleReadyForQuery() {}
182
}
183

184
/**
185
 * CopyFromQueryStream - Writable stream for COPY FROM operations
186
 * @class
187
 * @extends {Writable}
188
 */
189
class CopyFromQueryStream {
190
  /**
191
   * @param {string} text - The SQL COPY FROM statement
192
   * @param {StreamOptions} [options] - Stream options
193
   */
194
  constructor(text, options) {}
195
  
196
  /** @type {string} The SQL COPY statement */
197
  text;
198
  
199
  /** @type {number} Number of rows processed */
200
  rowCount;
201
  
202
  /** @type {object} PostgreSQL connection reference */
203
  connection;
204
  
205
  /**
206
   * Submits the query to a PostgreSQL connection
207
   * @param {object} connection - pg connection object
208
   */
209
  submit(connection) {}
210
  
211
  /**
212
   * Callback placeholder for pg timeout mechanism
213
   */
214
  callback() {}
215
  
216
  /**
217
   * Handles errors during stream processing
218
   * @param {Error} e - Error object
219
   */
220
  handleError(e) {}
221
  
222
  /**
223
   * Handles CopyInResponse from PostgreSQL backend
224
   * @param {object} connection - pg connection object
225
   */
226
  handleCopyInResponse(connection) {}
227
  
228
  /**
229
   * Processes CommandComplete message and extracts row count
230
   * @param {object} msg - Message object with text property
231
   */
232
  handleCommandComplete(msg) {}
233
  
234
  /**
235
   * Called when connection is ready for next query
236
   */
237
  handleReadyForQuery() {}
238
}
239

240
/**
241
 * CopyBothQueryStream - Duplex stream for COPY BOTH operations
242
 * @class
243
 * @extends {Duplex}
244
 */
245
class CopyBothQueryStream {
246
  /**
247
   * @param {string} text - The SQL COPY statement
248
   * @param {CopyBothOptions} [options] - Stream options
249
   */
250
  constructor(text, options) {}
251
  
252
  /** @type {string} The SQL COPY statement */
253
  text;
254
  
255
  /** @type {object} PostgreSQL connection reference */
256
  connection;
257
  
258
  /** @type {boolean} Whether to align data on COPY data frame boundaries */
259
  alignOnCopyDataFrame;
260
  
261
  /**
262
   * Submits the query to a PostgreSQL connection
263
   * @param {object} connection - pg connection object
264
   */
265
  submit(connection) {}
266
  
267
  /**
268
   * Handles errors during stream processing
269
   * @param {Error} err - Error object
270
   */
271
  handleError(err) {}
272
  
273
  /**
274
   * Handles out-of-band CopyData messages
275
   * @param {Buffer} chunk - Data chunk
276
   */
277
  handleCopyData(chunk) {}
278
  
279
  /**
280
   * Called when command completes
281
   */
282
  handleCommandComplete() {}
283
  
284
  /**
285
   * Called when connection is ready for next query
286
   */
287
  handleReadyForQuery() {}
288
}
289
```
290

291
## Error Handling
292

293
pg-copy-streams provides comprehensive error handling for various failure scenarios:
294

295
### Common Error Types
296

297
- **Syntax Errors**: Invalid SQL COPY statements
298
- **Connection Errors**: Database connection failures
299
- **Data Format Errors**: Malformed data that PostgreSQL cannot process
300
- **Permission Errors**: Insufficient database permissions
301
- **Stream Errors**: Network interruptions or stream corruption
302

303
### Error Recovery
304

305
- **Automatic Rollback**: Failed COPY operations are automatically rolled back by PostgreSQL
306
- **CopyFail Messages**: Stream destruction automatically sends CopyFail to abort operations
307
- **Transaction Safety**: Operations within transactions maintain ACID properties
308

309
## Compatibility and Performance
310

311
### PostgreSQL Compatibility
312

313
- **PostgreSQL Versions**: Compatible with PostgreSQL 9.0+
314
- **Protocol Support**: Uses PostgreSQL's native COPY protocol for optimal performance
315
- **Pure JavaScript**: Works only with pure JavaScript `pg` driver (not `pg.native`)
316

317
### Performance Characteristics
318

319
- **High Throughput**: Direct protocol integration bypasses query parsing overhead
320
- **Memory Efficient**: Streaming operations avoid loading entire datasets into memory
321
- **Chunk Boundaries**: PostgreSQL chunks data on 64kB boundaries (may split rows)
322
- **Backpressure Handling**: Proper Node.js stream backpressure support
323

324
### Node.js Requirements
325

326
- **Node.js Version**: Requires Node.js 8.0+ (async/await support recommended)
327
- **Stream API**: Built on Node.js streams for maximum compatibility
328
- **Buffer Handling**: Uses modern Buffer allocation methods for security