compute/correction-v2: introduce a Stage to speed up small inserts

[teskje]

This PR extends the CorrectionV2 data structure with a Stage that accumulates inserted updates before they get inserted into the sorted chains. This significantly reduces the amount of chain merges that need to be done in workloads that trickle in large amounts of updates in small batches, greatly speeding up these workloads.

For some reason, the memory tests introduced in #31267 are such a workload. Brief testing reveals that this PR speeds up the MV hydration in the -800mb test, when pointed against CorrectionV2, by roughly an order of magnitude, from ~300s to ~30s.

Note that this change doesn't have much impact when running locally on macOS. For some reason the batch sizes entering the MV sink are much smaller when running in mzcompose, not sure why that is.

Motivation

• This PR adds a known-desirable feature.

Part of https://github.com/MaterializeInc/database-issues/issues/8464

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

[antiguru]

The implementation looks very similar to a ConsolidatingContainerBuilder, is there a reason why it couldn't reuse the implementation? It has a slightly different API, but maybe it doesn't need to. In Timely, a good rule-of-thumb is that during an operator invocation, it's great to leave state across input batches, but once the operator ceases to work, it's better to finish pending work. The builder API gives you this in the form of the extract and finish API, but it's harder to track this with the APIs implemented by Stage.

Following the builder pattern, the Stage wouldn't need to hook into logic to account for size and be aware of frontiers.

[teskje]

The implementation looks very similar to a ConsolidatingContainerBuilder, is there a reason why it couldn't reuse the implementation?

Looks like the ConsolidatingContainerBuilder produces containers up to a preferred size, which makes it awkward to use in the context of the correction buffer. For example, in the current implementation if the Stage gets enough data to fill two chunks at once, it can consolidate all this data together and then build a chain out of two chunks directly. Whereas with the ConsolidatingContainerBuilder I think I would get two consolidated Vecs and to build a chain from them I'd have to merge them into one Vec and then consolidate them again. So we'd spend more time consolidating than we do now.

it's great to leave state across input batches, but once the operator ceases to work, it's better to finish pending work

Why would it be preferable to finish pending work when the operator becomes idle?

Keep in mind that the correction buffer is used in an async operator which, afaick, doesn't have a good way to know when it will become idle. It just does loop { select! { ... } } on a bunch of input channels. We could probably twiddle that control flow to do something special when we find out that non of the channels is ready, but that would make the code more complex, so we should have a good reason for doing so.

[teskje]

I tested using the ConsolidatingContainerBuilder instead in #31456. It reduces the amount of code, which is nice. But for some reason it also has way worse performance, compared to the hand-rolled Stage implementation in this PR.

[antiguru]

LGTM, let's take the discussion around the consolidating container builder offline!

[teskje]

TFTR!