Document/Improve: Continuous batching/chunking of a stream

[jmealo]

I'm creating this issue to discuss solutions (documentation or otherwise) to help folks avoid some of the non-memory related pitfalls on #116 when continuously batching COPY commands.

See:
#116 (comment)

@jeromew: I found out what happens if you always have a COPY in progress. It works great until something finally manages to get a lock on the table and then you have a bunch of COPY operations pending. From a locking perspective, it's really not ideal to start a COPY, stream data into it, close that stream, start a new COPY stream an immediately pipe into it, you're effectively never yielding the table for an exclusive lock.

🔥 If you do not give PostgreSQL some breathing room between COPY operations; you're going to run into 🔒 locking issues 🔒.

I'm interested if anyone has any ideas for the best ways to handle this entirely. I'm going to dump some tips for anyone working on this in the future.

Pro-tip: Safely create async batches

Under high load, a new event may arrive before the previous event's call to getBatch() resolves. This can lead to orphaned batches (memory leak) and lost messages.

const getBatch = async () => {
  // If another batch is already pending; we return the promise that will resolve to that batch.
  // You can .then() or await the same promise multiple times; you cannot resolve()/reject() multiple values
  if (nextBatch) {
    return nextBatch
  }

  try {
    nextBatch = producer.createBatch({ maxSizeInBytes })
    const batch = await nextBatch
    return batch
  } finally {
    nextBatch = undefined
  }
}

Anti-pattern: Immediately unpipe/piping another stream into rotating COPY streams

// Ooops! If you immediately rotate streams you will always have a lock on the table (until you don't and then COPY will pile up in a `WAITING` state, especially if you have control/metadata queries interleaved between)
const rotateCopyStream = async (logStream) => {
  // disconnecting the csv stream from our log stream, causes log stream to pause
  if (csvStream) logStream.unpipe(csvStream)

  if (pgStream) {
    // disconnecting the pg stream from the csv stream causes the csv stream to end
    csvStream.unpipe(pgStream)
    pgStream.end() // calling this ends the COPY command
  }

  // Get connection and execute our copy command
  const client = await pool.connect()
  pgStream = client.query(copyFrom(COPY_QUERY))

  // Debugging helps
  pgStream.on('error', (error) => {
    if (client) client.release()
    console.log('pgSteam -> stream errored')
    console.log(error)
  })

  pgStream.on('finish', () => {
    if (client) client.release()
    console.log('pgStream -> stream complete')
  })

  csvStream = format({headers: false, includeEndRowDelimiter: true, delimiter: '\t'})
  logStream.pipe(csvStream, {end: false}).pipe(pgStream)
}

@jbtobar: I'm using 100% streams in my workflow, but I've found a Node.js foot canon when batching things coming from an event emitter/event handler/subscription. If batch creation is async, you need to make sure if a second event comes in while the batches are being rotated, that you don't end up creating two or more batches. An example of how to accomplish this is below:

I considered putting an advisory lock at the beginning of the COPY command. This would give us a chance to keep our input stream paused and resume flow when the last COPY is complete with the possibility of application-level logic. I haven't tried it yet, but, that's my next plan of attack.

[jeromew]

Regarding the "antipattern" that is mentioned in #118 (comment)

pgStream.end() // calling this ends the COPY command

Calling end() only starts the process of ending the COPY operation (it sends CopyDone). You probably should not call

pgStream = client.query(copyFrom(COPY_QUERY))

and pipe into the new rotated stream until the previous stream emits 'finish'.

In the code, what happens is :

COPY -> COMMITING
     -> COPY

when what you want in order to leave more breathing room to the backend may be

COPY -> COMMITING -> COPY

[jeromew]

Just to make it clear this issue is for discussing a pattern that attracts some people because of the possibilities given by pg-copy-stream

Some people want to create a long-lived ingestion mechanism to ingest data into postgresql. For example, you could imagine that you have a never ending stream of logs that you want to ingest into postgresql.

this could be done by

1. inserting log lines one by one via INSERT commands
2. inserting batched volumes of log lines (let's say 100s) via MULTI-INSERT commands. cf "insert multiple rows" on https://www.postgresql.org/docs/14/dml-insert.html

there are probably other ways to do that with postgresql, including mechanisms linked with replication.

but still it can be interesting to see if and how COPY can help for this kind of scenarios.

first it is necessary to understand that a COPY FROM operation ingest rows. During the operation, the rows are not visible. They will become visible only once the COPY operation is terminated.

So pg-copy-stream cannot be used to open a long-lived stream, push data into it, and expect that this data will be queryable. For the data to be visible and queryable, you need to finish the COPY operation.

Once this is said, the idea of rotated COPY operations appear : ingest N rows via COPY and then create a new COPY operation.

It could be interesting to agree on advantages and disadvantages of the different solutions that can be used to do that and see how it impacts the performance of postgresql depending on the throughput of the ingested data, and how the throughput of out-of-band queries that need to be executed on the data.

I will leave this issue open for several months to see if people want to share their experience, have ideas around this, and discuss this further.

[jmealo]

@jeromew: Howdy, here I am bulk ingesting data into PostgreSQL again 😃 ... Multi-INSERT commands seems slow compared to copy but I haven't benchmarked it yet.

I'm realizing that a library that wraps pg-copy-streams that adds the logic for safely handling batches of bulk inserts, that flush for visibility and explains everything this issue does could be a good compromise.

If there was any interest in linking to it from this library, I'd gladly consider implementing it. Thoughts? @jeromew @brianc

[jeromew]

Hello,

I understand the need to find a nice streaming ingestion mechanism for long-lived streams and mega-huge ingested filed with regular visibility on the data beeing ingested (using one COPY operation only gives visibility after the COPY is commited).

it would indeed be interesting to benchmark COPY vs Multi-Insert.

by Multi-Insert for N rows, I do not mean N INSERT operations, "1 insert per row", but I mean the ability that postgres has to insert several rows as a batch in one operation - cf https://www.postgresql.org/docs/current/dml-insert.html and thus send 1 INSERT with N rows

You can insert multiple rows in a single command:

INSERT INTO products (product_no, name, price) VALUES
    (1, 'Cheese', 9.99),
    (2, 'Bread', 1.99),
    (3, 'Milk', 2.99);

The TIP on this page says

When inserting a lot of data at the same time, consider using the COPY command. It is not as flexible as the INSERT command, but is more efficient.

I fully agree for 1 COPY versus INSERTs but if the regular visibility is important to you I would be surprised that the time spent setting up regular COPY operation for N rows will outperform a regular INSERT with N rows, and even more so if the INSERT N rows is a prepared statement

A specificity also when using COPY is that copy-from does not require the incoming stream to be respectful of row boundaries (Postgres has everything it needs to detect row boundaries and reconstruct rows from streamed network chunks) but in your solution you would need to either add a row boundary detection mechanism or require incoming streams to be respectful of row boundaries.

Note also that if the stream is an object stream with 1 row per chunk, it is important performance-wise to recombine the rows into bigger network chunks as per #127

[Inviz]

I'm seeing copy stream library spending a lot of time flushing stream synchronously. Is there any way to make it async?

[jeromew]

Hello,
I am not sure about what you observe exactly. I guess it depends on your setup so an example would be welcome.
nevertheless it could be related to what was discussed earlier in #127
adding a batching stream before postgres can avoid too many network requests

[Inviz]

Thanks @jeromew . I am already batching the queries. I think my questions is more philosophical - is there a way to write into PG buffer without blocking JS world? Probably not? Would native bindings improve it in any way? (I am faintly aware that node-pg-copy-streams may not support it, but in principle?)

Then would spawning streamer in a worker with parallelism, - do you think stream-COPYing into the same table makes sense in parallel?

[jeromew]

I don't think pg-copy-stream is currently blocking JS world when writing into PG (well maybe it is but this would need a precise test script highlighting this). we're writing into a stream. There could be scenarios where if the upstream source itself is synchronous maybe it has precedence or something but here again, without a simple & precise test script hightlighing the issue there is not much i can do.

regarding multiple clients streaming into the same table, I haven't tested it. If postgres supports it, it could probably speed things up.
but according to source code it seems COPY FROM acquires a Row Exclusive Mode lock on the table so I don't think you can do 2 copy operations in parallel on the same table ; you will have to test.

[Inviz]

@jeromew Thank you. Given that .write() calls are synchronous, it's fair to say that there's going to be some blocking somewhere down the line, as it's the only way to provide back-pressure to the caller. PG driver does flushing of stream, blocking further write()s, and the world. At least that's what i gathered doing profiling and overall poking. There's no other way to have while (true) stream.write() loop if it's not blocking, right?

You are indeed right, it seems PG doesnt parallelize COPY, other places in internet also support that interpretation.

How about native bindings? Would there theoretically anything to be gained if pg-copy-stream would support those?

So far it looks like only low-hanging improvement i can do is to try to issue pg-copy-stream from within a worker, but it needs some

Thank you for answers and help.

[jeromew]

@Inviz I agree that the node stream machinery takes some toll on the main thread so having pg-copy-streams on a huge file in parallel with another application sharing the main thread can probably be seen ; but I think the main drivers on the total ingestion time are the network calls (which is probably why the batching stream in front of from improves things a lot)

why do you talk about 'worker' ? is it because you share the ingestion with another application like a web application ? For sure if you push stream pressure too far it can have an effect on the main thread. I think pg-copy-streams handle the back pressure correctly but never tested it thouroughly. maybe playing with the watermarks can help. Maybe there are other ways to give some breathing air to the main thread. Also not having postgres on the same machine will help if that's your case.

Can I ask what is your use case ? I would help me understand the scenario.

For now I am not in favor of going the 'native' way :

• 1/ would the 'record' throughput really be implemented in javascript ? I now there are other implementations out there in other languages.
• 2/ do you already handle your source stream as binary ? This can both shrink to total size to ingest & give less work to postgres ; but it will give more work on the javascript side if you need to convert from plain to binary
• 3/ depending on the scenario and the data, there are other quick wins that can be implemented: suppress indices before ingestion ; ingest in many tables in parallel and then send a reconciliation sql request that merges those tables or have some kind of view on all the ingested tables, ..

but this is all about the speed of ingestion, not about the impact on the main thread which appears to be your issue here (?)

[Inviz]

@jeromew I really appreciate your thorough reply, thank you. I hope that doesn't hijack the thread of discussion, as i think it's semi-related to the OP.

why do you talk about 'worker' ? is it because you share the ingestion with another application like a web application ? For sure if you push stream pressure too far it can have an effect on the main thread.

So as you very helpfully pointed out, COPY into a single table doesnt work in parallel. In my case however I am streaming into multiple tables at once (as a result of processing one input stream), so potentially putting different streams into different workers could increase the overall throughput, by making multiple streams not block each other when flushing chunks into PG. This seems to be the most promising way so far to unlock higher speeds.

I think you are right that there could be alternative ways to give breathing space to the main thread, through balancing act of finding the right buffer size.

Thanks for providing your comments on the native bindings approach. I think you are right, there's not a whole lot to be gained without making big changes, like going binary, suppressing indecies, streaming into temp tables, etc. I am using continuous aggregations/hypertables in timescaledb (PG extension), so it's desirable to find approach that keeps streaming realtime. But i think second best approach is streaming into temp table and then force PG to do the merging.

I think in my workloads I am not at the point that makes it feasible. I wanted to figure out what's the best I could do without losing flexibility.

I'll post up my results of experimenting with workers.

[jeromew]

@Inviz in the one-stream => multiple COPY stream to different tables scenario, and depending on the work, having the ability to use several CPUs via workers can probably increase the throughput.
I guess you have already found https://github.com/timescale/timescaledb-parallel-copy (go language) in which timescaledb probably put some thought. it seems to be using workers each responsible with batches of around ~ 5000 lines all inserting in the same hypertable. probably timescaledb has modified the COPY mechanism to benefit from hypertables so that could bypass the LOCK in some way if the batches fall into different physical tables).

Depending on the scenario you can also look into removing the wall logging on these tables. you could also maybe look into redis streams to have the initial PUSH into redis and then workers PULLing from redis streams but that's just an idea.

you can also take a look at pg-copy-streams-binary . it is not optimized for high throuput because it does systematic byte<->object conversions but it can give you an example about binary handling and one streams => many tables. When I wrote that I used https://www.npmjs.com/package/multi-fork for the partitioning of the streams / this module has interesting ideas but I realize that it would probably need a refresh with newer versions of node / streams.

optimizing node pipelines is hard ! I am still looking for a node pipeline "live viewer" if you happen to find one !

[Inviz]

Yes I've seen that timescaledb parallel copy thing. When I dug deeper into pgx dependency, i didnt quite find where it achieves the concurrency while writing into a table, as it seems to be using COPY FROM faithfully all the way down. Timescaledb however is known for adding features that dont require turning them on explicitly, so it could be something inside, as you are theorizing. The tool has some regressions in losing reporting capability, so i couldnt at the time make faithful test. I'll give it more thought, there's probably to be gained something here.

Thanks for your ideas and thanks for the pointer for binary version of the library and some context around it. I'll check it out!

Personally i am working on my own async generator-based library https://github.com/Inviz/iterators which allows for easy concurrency, partitioning and such. But i definitely feel what you mean about live viewer. Understanding and tracing streams is hard. I had to use flamecharts mostly. But i was considering to build better tracing/viewing primitives.

[jeromew]

note you may also want to look at alternative js COPY implementations like in https://github.com/porsager/postgres
I did not test to see how they behave compared to pg/pg-copy-stream but maybe there is less overhead for your use case