[dbsp] Split large outputs in chain_aggregate. #5376
Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds logic to split large outputs into multiple steps for the chain_aggregate operator, completing the implementation of output splitting across operators in the codebase.
Key Changes:
- Converts
ChainAggregatefrom aBinaryOperatorto aStreamingBinaryOperatorto enable chunked output processing - Implements batching logic that yields results in configurable chunk sizes instead of processing entire datasets at once
- Updates test configuration to verify chunking behavior with a small chunk size
ryzhyk
force-pushed
the
chain-aggregate-splitter
branch
from
51cab13 to
dbc94d8
Compare
| self: Rc<Self>, | ||
| delta: &Option<Z>, | ||
| output_trace: &Spine<OZ>, | ||
| ) -> impl futures::Stream<Item = (OZ, bool, Option<Position>)> + 'static { |
There was a problem hiding this comment.
why not use AsyncStream here as in other places?
| delta_cursor.step_key(); | ||
| delta_cursor.step_key(); | ||
|
|
||
| if builder.num_tuples() >= chunk_size { |
There was a problem hiding this comment.
I should have known this, but this means that the output can be larger than the chunk size?
The current builder does not seem to be reduced to chunk_size and the left-over saved for later.
There was a problem hiding this comment.
the builder can grow up to chunk_size + 1, since it can produce two output records per key.
Once that happens, we will output everything in the current builder and create a fresh builder for remaining outputs.
Not sure if this answers the question.
github-merge-queue
bot
removed this pull request from the merge queue due to a conflict with the base branch
I think this is the last operator that was missing the logic to split large outputs into multiple steps. Signed-off-by: Leonid Ryzhyk <[email protected]>
ryzhyk
force-pushed
the
chain-aggregate-splitter
branch
from
dbc94d8 to
8deb916
Compare
I think this is the last operator that was missing the logic to split large outputs into multiple steps.