There's a subtle distinction between "batch size" and "flush threshold" -- I suspect they are related but it's not entirely explicit here. A bit more explanation could be helpful.

I'm not sure what you're referring to? What's "batch size"? There's no config with that name 🤔

No, but this thing is a "Batcher" and so it produces batches and those batches have a size. In particular the thing I think I was curious about was whether the batch would have a size > flush_num_bytes (I believe the answer to this is yes), but I wanted to be explicit that when we flush, all of the outstanding bytes (including those above the flush threshold) will end up in the batch.

Ah yes, that's what the equal or greater than this alludes to. I'll make it extra clear.

Out of scope for this PR but I'm a bit unsure myself how we should be deciding when we use tokio vs when we just spawn threads.

My general motto is: if you can avoid async, avoid async.

I think we want to drop self.rx_tables before we call this send to avoid a deadlock?

Basically, if we are using a bounded channels and rx_tables isn't being drained by any listeners, then tx_table.send(table) could be blocking the batching thread, preventing it from ever processing the Shutdown. Dropping rx_tables first should at least cause tx_table to either error or it means some other outstanding sender thread is still holding a receiver and then at least it isn't our problem.

Yeah the docs mention this:

/// Shutting down cannot ever block, unless the output channel is bounded and happens to be full
/// (see [DataTableBatcherConfig::max_tables_in_flight]).

My thought process being that if the user deliberately configures the channel sizes to be bounded, then they should expect that the system can and will block at any (inconvenient) time if they don't consume as needed.

Eh, I guess we can be extra polite...

Any particular reason to make this shared? Is there value in a new receiver jumping in mid-stream? I worry about the usefulness of the data in that case.

The channel is already mpmc by nature, so I didn't see the point of umping through extra hoops to turn it back into an mpsc one. Also parallel consumers might come in handy at some point.

How expensive is this? Wasn't this a bottle-neck before?

It was and still is and that's what's so nice about it: it's now done in a background thread on the clients...

But now we still need to send that information to the server to make things really fantastic.

it's now done in a background thread on the clients

We should be careful about that though... we don't want to make this a bottleneck or a CPU drain on the clients either.

It really isn't: it's extremely costly compared to the rest of the operations that we do on the very fast paths in the store, but it's still order of magnitude faster that most of what goes on on the client... especially if it's a python client 😒

Rather than returning a bool it seems like it would be clearer to call do_flush_all here.

I find it clearer to be able to see all flush triggers in the main loop

In that case why not move the check into the main loop as well?

do_push_row(&mut acc, row);
if acc.pending_num_rows >= config.flush_num_rows || acc.pending_num_bytes >= config.flush_num_bytes
{
    do_flush_all(&mut acc, &tx_table, "bytes|rows");
    acc.reset();
}

But, at the very least add a comment to do_push_row indicating that it returns a bool with the assumption that the caller will flush the data if it returns true?

Moving the check sounds good to me 👍