RFC: Breaking `Inserter` API Changes

[abonander]

Use case

Writing to Inserter from multiple threads

Applications that are trying to maximize client-side batching might want to buffer data from multiple threads simultaneously. As of now, they'd have to wrap Inserter in a mutex which would force all writes to be serial.

Async flushes

When the period elapses, Inserter has no way to flush the batch asynchronously, i.e. without the user calling .write() or .commit(). This is because the timeout checks and flushing are done directly in these methods.

Additionally, because these methods have to "block" on finishing the current INSERT statement and receiving a response, it's not possible to begin the next INSERT statement until we get a response for the previous one, resulting in a stall in ingestion rate (#41).

.commit() method naming is misleading

ClickHouse doesn't have transactions (not technically; there's a preview feature for transactions in Cloud but it's not generally available or recommended yet), and data is generally inserted as soon as it's received server-side. .commit() should perhaps more accurately be called .flush() because all it does is ensure the data in the buffer is sent to the server and finishes the current INSERT statement and underlying HTTP request.

Enabling retries

Currently, when an INSERT statement fails in the middle of writing data, it's impossible to tell how much data was successfully flushed to the server and the client-side buffer is completely discarded when it could be buffered and sent with the next request instead (#212).

Describe the solution you'd like

Move the Inserter state machine to a background task and send data chunks via a channel. (Alternatively, some sort of ring buffer? I don't remember there being a good high-quality crate for ringbufs though.)

Since channels are generally MPSC, .write() can change to take &self.

Deprecate or just delete .commit() and .force_commit(). Add .flush() as a replacement for .force_commit(). Threshold-based committing/flushing can be done asynchronously on the background task without an explicit call.

Maybe add a .compress_buffer() call (#233)?

Describe the alternatives you've considered

Additional context

[loyd]

The classic pattern for high reliability is to dump data to disk and remove it only when CH responds "Ok". So, the new API should also provide the required information, which is now returned by commit(), at least Quantities::rows. How is it supposed to work with the suggested design?

Applications that are trying to maximize client-side batching might want to buffer data from multiple threads simultaneously
As of now, they'd have to wrap Inserter in a mutex which would force all writes to be serial.

Why don't these applications use async inserts then? It's trivial to have multiple tasks running while Some(d) = rx.recv().await { inserter.write(d); inserter.commit().await; } with enabled async inserts. This approach scales very well (both in terms of CPU and network utilization).

.commit() method naming is misleading

Yes, totally agree! The original name was from the pattern I used many years ago: one large table transaction_log with multiple MV into dedicated tables (like trade_log, balance_log, etc.) and the table with the name that should be the latest (~commit_log). So, inserter.commit().await is called after multiple related events have been written, and a successful return means the "transaction" has ended. The reading side checks ~commit_log and understands which "transactions" actually ended. That's also the reason behind Quantities::transactions. But, again, the name is misleading. However, the guarantee that rows inserted between commit() calls will go into the same INSERT is still in use and important.

can be done asynchronously on the background task without an explicit call

One wish: give the way to wrap (or even accept some trait Executor) so that the future can make it possible to track memory usage / cpu usage / etc under valid names. The current design supports it unnoticed because the user must call methods to make the progress instead of

Threshold-based committing/flushing can be done asynchronously on the background task without an explicit call

Also, I don't understand how a new Inserter can be used with borrowed data. Can you explain it?

[abonander]

Why don't these applications use async inserts then? It's trivial to have multiple tasks running while Some(d) = rx.recv().await { inserter.write(d); inserter.commit().await; } with enabled async inserts. This approach scales very well (both in terms of CPU and network utilization).

It can be much more expensive to do the merging on the server-side at high ingest rates. ClickHouse really likes having larger part files.