still TODO for me: summarize the impact on the end-to-end test with 1 million 110 byte rows.

it looks like the encoded batch sizes went up. I can see how we'd have more overhead for very small batches with this (e.g. 4 vecs vs 2 per record). let's look into whether there's an inflection point where a sufficiently large batch is smaller in the columnar representation. if not, we should at least understand why before we move forward with using this in the storage format

also, I've sufficiently forgotten the context of my WIP that you shouldn't feel the need to structure commits as things on top of it. once we're confidant in the perf worthiness of this and you're ready to get it reviewed for merge, let's structure the PR as "introduce a columnar batch thingy" in one commit and "use it for unsealed" in another

once we're confidant in the perf worthiness of this and you're ready to get it reviewed for merge, let's structure the PR as "introduce a columnar batch thingy" in one commit and "use it for unsealed" in another

yup -- that was how I was hoping to merge it. Kept it this way in case it was easier for review, but am happy to switch to this structure

This change is RFAL! I broke it up into logical commits as requested, and I have the following answers for the questions around encoding size, and impact on end-to-end tests, which I also included in the second commit messages (under the benchmark results)

For very small batches this change results in a noticeable increase in the encoded
size but the effect is small, and entirely unnoticeable once data volumes exceed
1,000 bytes (measured with inserts of 1.1KB, 110 KB, and 110 MB, and the sizes of
unsealed batches produced before and after this change were identical).

Unfortunately, this change doesn't register any improvement in the end-to-end test
sending 1 million 110 byte rows via a COPY INTO, because we save on allocations
during `write_atomic`, but end up having to allocate when sending rows to the listener.
However, there's clear positives from this change. Serialization times decrease
from 230 ms to 170 ms (out of a total of ~360 - 370 ms CPU time spent in drain_pending).
This represents a ~20% improvement that is roughly commeasurate with what we observe
in the benchmarks. On the coordinator side, we go from spending ~175 ms in
`write_atomic` to spending about 140 ms in `write_atomic`, which is a smaller than
expected change. Curiously, the trace seems to be dominated by `free_tiny` which
leads us to suspect that perhaps the cost we're now bottlenecked on is the `drop`
on the owned vec that gets passed to `write_atomic`.

It seems very likely that the end-to-end tests will show improvement when we move updating listeners to a background task.

Hmm, I'd like to do some more experiments then before we merge this. From a results perspective, I'm highly confidant that this is the right direction, but from a process perspective, I want us to do our due diligence before we commit to such a big change in the name of perf. In particular, actually seeing the needle move on the end-to-end test.

For very small batches this change results in a noticeable increase in the encoded size but the effect is small, and entirely unnoticeable once data volumes exceed 1,000 bytes (measured with inserts of 1.1KB, 110 KB, and 110 MB, and the sizes of unsealed batches produced before and after this change were identical).

Huh. The columnar representation was the same size? That's surprising, I'd expect at some point for them to be smaller, but maybe my mental model here is wrong. I have zero idea how bincode encodes things, but let's walk through a generalized example of what might be happening. For the following:

<key1><val1><ts1><diff1>
...
<keyN><valN><tsN><diffN>

The row encoding might be <N_rows: u64> followed by N repetitions of <key_len1: u64><key_bytes1: [u8: key_len1]><val_len1: u64><val_bytes1: [u8: key_len1]><ts1: u64><diff1: i64> for a total of 8 + 8*N + all_key_bytes 8*N + all_val_bytes + 8*N + 8*N = 8 + 32N + all_key_val_bytes.

The col encoding might be <N_keys: u64> followed by N repetitions of <key_len1: u64><key_bytes1: [u8: key_len1]> followed by the same for vals, then <N_timestamps: u64><timestamps: [u64; N_timestamps]><N_diffs: u64><diffs: [u64; N_diffs]> for a total of 8 + 8 * N + all_key_bytes + 8 + 8 * N + all_val_bytes + 8 + 8 * N + 8 + 8 * N = 32 + 32 * N + all_key_val_bytes.

Yeah, I can see how bincode might behind this. The usual places for columnar to pull ahead of row encoding is if the format is self-describing (bincode isn't AFAIK) or if the encoding doesn't store things like [u8] inline, but instead has a pointer out to the data (e.g. capnp). And it's losing a bit of efficiency on the columnar side by having to encode the length of the 4 vecs separately, even though they're always the same in practice.

I'm sufficiently convinced that this behavior is expected.

we save on allocations during write_atomic, but end up having to allocate when sending rows to the listener.

Let's explore some prototypes of how we can avoid this. On one hand, we could move sending to the listener out of the hot path, which is something we'll likely want to do at some point, but in a way it's just shuffling the problem to some other thread, which will still become a throughput bottleneck for us at some point. What about this: instead of sending a ((Vec<u8>, Vec<u8>), usize, i64) out to the listener, we could send a (idx: usize, batch: Arc<ColumnarBatch>) and the listener is responsible for indexing the indicated row out. That would avoid the col->row transformation entirely.

To be clear, my last suggestion isn't that we should expand the scope of this PR to include that change, just to build a quick prototype of it on top of this PR and see if the needle moves on end-to-end in a way that we expect.

Thanks for the review Dan! I agree that being more rigorous with the performance analysis would be better.

Re: the size on disk - I actually think you've raised a really interesting question and its actually non-obvious why the data in storage are the same between the row-oriented, and the columnar representations, and thinking about this has helped me understand what we're actually writing down better.

Concretely, we have a 100 byte string + 2 4 byte ints in each row inserted with COPY FROM. In the Row encoding that turns into 100 bytes for the string + 1 byte for the length of the string (because its small), + 3 tag bytes (one per column) for a total of 112 bytes. When we call Codec::encode() on this Row we add 8 bytes for the row length + 1 byte of magic, so 121 for the encoded Row.

Thats key_bytes. val_bytes is just empty because its (). That + 2 x 8 bytes (for time and diff) comes out to 137 bytes per ((key, val), time, diff), which seems like it has to be constant between the previous representation, and the columnar one, because we're not really doing any compression on any of these bytes, we're just re-arranging how they are laid out.

BUT, it feels like, given your reasonable description of how bincode might be doing things -- bincode doesn't know a priori that we're encoding the length in the first 9 bytes of each key, and would have to encode the length of the Vec<u8> provided per key and value in a way that the columnar representation does not at first glance. Specifically, I don't think "The col encoding might be <N_keys: u64> followed by N repetitions of <key_len1: u64><key_bytes1: [u8: key_len1]> followed by the same for vals," is correct because it doesn't make sense to me that the key_lengths would get interleaved with each key's bytes. However, the columnar representation pays the same additional cost in the offset arrays (which is something I didn't realize until thinking through this!). So, thats an additional 16 bytes, stored either inline with the row-oriented representation, or in separate arrays in columnar, for a total of 153 bytes per row inserted, which is almost exactly whats observed (148 MB == 148 * 2^20 bytes ~= 155 million bytes)

Perhaps one takeaway here is that the length prefix we add in Codec::encode for Row could/should be removed?

Re: prototype -- I'll do that right now.

I also was thinking about breaking the last commit that introduces the columnar representation into two commits, one that introduces the idea of the WriteReqBuilder and changes writes to happen in terms of ColumnarRecords, but doesn't change the on-disk encoding (this commit would be a small perf regression - because we'd be allocating a large ColumnarRecords object, and then breaking it up to construct a BlobUnsealedBatch), and then another commit that handles actually changing the encoding.

My main motivation here is to avoid having to rebase the write path code multiple times between now and when we can do a meta version bump, but I don't have super strong feelings about this. wdyt?

I also was thinking about breaking the last commit that introduces the columnar representation into two commits, one that introduces the idea of the WriteReqBuilder and changes writes to happen in terms of ColumnarRecords, but doesn't change the on-disk encoding (this commit would be a small perf regression - because we'd be allocating a large ColumnarRecords object, and then breaking it up to construct a BlobUnsealedBatch), and then another commit that handles actually changing the encoding.

My main motivation here is to avoid having to rebase the write path code multiple times between now and when we can do a meta version bump, but I don't have super strong feelings about this. wdyt?

Yeah that sgtm, exactly the sort of thing I want to start practicing in advance of when we do backward compatibility for real. I was thinking yesterday that I could do something similar with that WIP inc meta branch I linked you the other day

Alright! Took a while to de-noise the end-to-end benchmark -- for future ref

cargo run --release -- -w 1 --log-file=stderr --experimental --persistent-user-tables --disable-persistent-system-tables-test --timestamp-frequency 100ms

turns off the system table test and correctly sets the step interval to 100ms. Going from there, a nice repeatable testing protocol is to wipe mzdata, and then do 5 - 10 inserts of 1 million batches, without Instruments attached. Instruments adds another ~15% of overhead. Also, I commented out drain_unsealed and compact to focus on the write latency.

With all of that -- the best case latency before the columnar change is 2.2 seconds, and the best case latency after the columnar change (changing the implementation to send ColumnarRecords directly to listeners as discussed) is 2.1 seconds. Of that time, ~1.5 seconds is spent decoding the copy text format.

The 100 ms improvement is roughly what we had expected based on the microbenchmarks / more detailed info from Instruments (we go from spending approximately 700 ms in coord + persist to 600 ms in coord + persist, driven by speedups in serialization and small speedups in write_atomic)

For me this is pretty conclusive proof that this change does indeed improve performance end-to-end. I'll go ahead and split up the commits so that we can merge everything that threads ColumnarRecords through, without actually changing the encoding, and then leave the encoding change for our breaking change window.

RFAL! This pr now threads ColumnarRecords through as much of persist as possible without changing the encoding, and #9239 handles actually changing the encoding

tftr! merging on green!