feat(derive): New BatchStream Stage for Holocene

[refcell]

Description

Adds a new BatchStream stage for holocene.

Makes progress towards #559

[refcell]

• feat(derive): Channel Flushing #576
• feat(derive): Holocene Buffer Flushing #575
• feat(derive): BatchStream Stage Wiring #567
• feat(derive): New BatchStream Stage for Holocene #566 👈
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @refcell and the rest of your teammates on Graphite

[codecov]

Codecov Report

Attention: Patch coverage is 54.09836% with 28 lines in your changes missing coverage. Please review.

Project coverage is 78.4%. Comparing base (b2f114a) to head (52962be).
Report is 1 commits behind head on main.

✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
crates/derive/src/stages/batch_stream.rs	54.0%	28 Missing ⚠️

Additional details and impacted files

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[clabby]

This is a good start. Small preference for BatchStream rather than BatchSpan, as a dumb naming nit.

I think there's a few things we need to think about architecturally here, though. The way I think about this is that we need to be able to stream single batches from the span batch into the BatchQueue, one at a time. This means that, from holocene forward, every batch sent into the BatchQueue is a SingleBatch. We need to retain next_batch(...) -> PipelineResult<Batch>, to keep that backwards compatibility, but the new stage should enforce that.

What this looks like in my head is something like:

The things I'm trying to optimize for here is:

• The BatchQueue currently performs the check_batch call. But it does it on Batch, which wraps both SingleBatch + SpanBatch. After holocene, basically, the BatchQueue should never receive a SpanBatch. This way, we can do the span batch prefix check in the BatchStream, and just reuse the SingleBatch check in the BQ.
• The BatchStream stage basically acts as a buffer of SingleBatches, derived from a SpanBatch. So it holds all of the SingleBatches from a span batch in-memory, and gets drained by the BatchQueue as SingleBatches are read. It can eventually land on Eof, signaling that it needs to fetch a new batch from the ChannelReader.

Notably this means that the stage will always be present - no need for a active flag, just forwards batches from the ChannelReader directly pre-holocene. Otherwise, it acts as an in-memory buffer of SingleBatches, and also owns the job of validating the span batch's prefix per the spec.

[clabby]

Good start! Let's get this in and keep iterating.