Refactor and optimize distributed COPY

[erimatnor]

Refactor the code path that handles remote distributed COPY. The
main changes are:

• Use a hash table to lookup data node connections instead of a list.
• Reduce the number of data copies by accumulating bigger CopyData
  messages directly in the libpq connection buffer instead of doing it
  in a separate buffer which later requires copying the data. To
  achieve this, some internal libpq functions are imported that gives
  more fine-grain control over buffers.
• Add an foreign data wrapper option to set the number of rows to
  send in a CopyData message.

[codecov]

Codecov Report

Merging #5417 (a669236) into main (c6b9f50) will decrease coverage by 0.05%.
The diff coverage is 85.30%.

❗ Current head a669236 differs from pull request most recent head 4ee69d6. Consider uploading reports for the commit 4ee69d6 to get more accurate results

@@            Coverage Diff             @@
##             main    #5417      +/-   ##
==========================================
- Coverage   90.76%   90.71%   -0.05%     
==========================================
  Files         229      230       +1     
  Lines       53705    53761      +56     
==========================================
+ Hits        48747    48771      +24     
- Misses       4958     4990      +32     

Impacted Files	Coverage Δ
tsl/src/remote/connection.c	88.66% <0.00%> (-0.21%)	⬇️
tsl/src/fdw/option.c	58.18% <20.00%> (-1.82%)	⬇️
tsl/src/remote/libpq-ts.c	71.83% <71.83%> (ø)
tsl/src/remote/dist_copy.c	88.01% <92.26%> (-1.47%)	⬇️
tsl/src/nodes/data_node_copy.c	95.23% <100.00%> (+0.53%)	⬆️

... and 8 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[akuzm]

there seems little to gain from accumulating a bigger data buffer before calling PQputCopyData()

The reason to batch rows before PQputCopyData is that it is a pretty complex function that adds several kinds of overhead:

1. Overhead in the size of data that is sent to the remote server. Each buffer is sent as a separate message, which has overhead (header/length/trailer etc).
2. CPU overhead on data node to parse these extra messages.
3. CPU overhead on access node. This function does a lot of things.
4. Possibly network overhead because it tries to flush the pending output buffers.

Removing the batching increases all these kinds of overhead 1000 times (by current batch size).

Point 3 is probably the most critical, because COPY throughput is limited by CPU usage on the access node (Igor can comment on that).

[erimatnor]

there seems little to gain from accumulating a bigger data buffer before calling PQputCopyData()

The reason to batch rows before PQputCopyData is that it is a pretty complex function that adds several kinds of overhead:

1. Overhead in the size of data that is sent to the remote server. Each buffer is sent as a separate message, which has overhead (header/length/trailer etc).

2. CPU overhead on data node to parse these extra messages.

3. CPU overhead on access node. This function does a lot of things.

4. Possibly network overhead because it tries to flush the pending output buffers.

Removing the batching increases all these kinds of overhead 1000 times (by current batch size).

Point 3 is probably the most critical, because COPY throughput is limited by CPU usage on the access node (Igor can comment on that).

CPU overhead is actually what I am most concerned about. I am testing different things in this PR (which is why it is still in draft), and removing buffering actually is faster on my local machine. Presumably this is because the buffering done in the current code is actually very CPU intensive given how much string wrangling happens: for each data node, we need to first construct the row index arrays, then loop through them to create a data node unique buffer/data string. That's potentially a lot of CPU to do that and it gets much worse with higher replication factors. Perhaps the network overhead is worth it if CPU/latency is the bottleneck and not network bandwidth.

Now, my local (single machine) setup is not realistic so we should basically benchmark this code on a real cluster and see what effect it has, then change accordingly. I might be wrong, and, in that case, adding back some buffering to this PR is not difficult.

[erimatnor]

@akuzm I added the ability to accumulate bigger CopyData messages directly in the libpq send buffer instead of maintaining our own buffers. The number of rows to accumulate before sending a CopyData message can be set using a foreign data wrapper option. This allows us to test and evaluate different buffering levels.

I had to import some libpq code because, although it is possible to include the internal libpq API (actually the header explicitly says apps can do this at their own risk), the internal functions aren't exported by the linker. But the send functions needed isn't much code and is likely to stay quite stable across PG versions.

[erimatnor]

Here are some performance numbers for ingest with this change on a 5-DN cluster:

[akuzm]

Do we still do the periodic flush somewhere? We need some way to control the outgoing queue size. I expect we're going to be bottlenecked at the data nodes in the normal case, so we need to avoid growing the output buffers on access node indefinitely.

[erimatnor]

So, I removed this because I don't think it is necessary.

The reason I don't think it is necessary is because even if we'd use standard libpq functions, it actually flushes automatically once a msg end marker is written and the buffer has reached 8k.

Further, now that I changed this to use the internal API, we have better control on a per socket basis. So, every socket is flushed once we write the CopyData end marker, which happens after N rows are buffered. So, basically, we accumulate up to N rows of data on a socket, then write msg end and flush. If the flush fails, the code will do a WaitEventSetWait on all sockets that could not be flushed.

Previously, we flushed all sockets after having sent N rows across all sockets, which meant that some sockets might not require a flush if they didn't get many of those N rows.

I kept the flush_active_connections() at the end of the transfer, but it only does something if any of the sockets failed the their last flush for some reason. Then it will wait until the flush succeeds.

[akuzm]

So, every socket is flushed once we write the CopyData end marker, which happens after N rows are buffered. So, basically, we accumulate up to N rows of data on a socket, then write msg end and flush. If the flush fails, the code will do a WaitEventSetWait on all sockets that could not be flushed.

Does this guarantee that we're flushing at the same rate we're receiving the new data, and the buffer size doesn't grow? Before it just flushed to zero outgoing buffers, now that you imported the internal header, probably we can just examine the buffer size.

[akuzm]

After reading the code more, looks like send_row_to_data_nodes flushes to zero output buffer, right?

[erimatnor]

@akuzm yes, send_row_to_data_nodes() is supposed to flush. However, admittedly this wasn't entirely clear and I reworked the code a bit to reuse flush_active_connections() since I had unnecessarily duplicated some of that code. Do you think it is more clear now?

[akuzm]

Thanks, yes, it's more clear.

[akuzm]

Good to see the improvement in the benchmarks.

[erimatnor]

@akuzm @pmwkaa @nikkhils I had issues getting the use of internal APIs to play nice with the Windows builds and tests, so I reverted the use of those APIs. Multi-row message are now accumulated in per-datanode buffers, similar to before. This means additional copying, but it doesn't seem to affect the performance in my local tests. I think the rest of the refactor still has value, however.