refactor(streaming): better source executor & split change #3669
Conversation
Signed-off-by: Bugen Zhao <[email protected]>
Signed-off-by: Bugen Zhao <[email protected]>
Signed-off-by: Bugen Zhao <[email protected]>
Codecov Report
@@ Coverage Diff @@
## main #3669 +/- ##
==========================================
- Coverage 74.23% 74.23% -0.01%
==========================================
Files 789 789
Lines 111366 111368 +2
==========================================
+ Hits 82676 82677 +1
- Misses 28690 28691 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more |
Signed-off-by: Bugen Zhao <[email protected]>
There was a problem hiding this comment.
LGTM! Most of the comments are unrelated to your change, I'm just revisiting the previous source implementation.
cc @KeXiangWang I think we should not make the source executor part harder to read and maintain (as seen in #3471). With this PR's change, it would be easier to follow and implement pause barrier. Would you please base your changes on this? What do you think?
| while let Some(barrier) = rx.recv().await { | ||
| yield barrier; | ||
| } | ||
| bail!("barrier reader closed unexpectedly"); |
There was a problem hiding this comment.
may also pending here. Not sure whether replacing a terminated stream would cause some confusing result...
There was a problem hiding this comment.
May fix in further refactors.
| for ele in &mut boot_state { | ||
| if let Some(recover_state) = self | ||
| .split_state_store | ||
| .try_recover_from_state_store(ele, epoch) |
There was a problem hiding this comment.
should wait_epoch here in the future, need revisiting this part of code.
| { | ||
| if let Some(target_splits) = mapping.get(&self.actor_id).cloned() { | ||
| if let Some(target_state) = self.get_diff(target_splits) { | ||
| log::info!( |
| split_offset_mapping, | ||
| } = chunk_with_state?; | ||
|
|
||
| if let Some(mapping) = split_offset_mapping { |
There was a problem hiding this comment.
So every time we receive a chunk, we will have such many things to do... I'm thinking of how to improve this part's efficiency.
There was a problem hiding this comment.
Would it be better if we store stream_source_splits as a HashMap?
| } | ||
| self.state_cache.clear(); |
There was a problem hiding this comment.
Why do we need to clear state cache here?
There was a problem hiding this comment.
we write state into state store every time a barrier comes. if the state is none we can skip the process.
…elabs#3669) * remove lock replace Signed-off-by: Bugen Zhao <[email protected]> * further cleanup Signed-off-by: Bugen Zhao <[email protected]> * fix clippy Signed-off-by: Bugen Zhao <[email protected]> * refine docs Signed-off-by: Bugen Zhao <[email protected]> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.
What's changed and what's your intention?
We need to construct a new
SourceStreamReaderImplon split change. The previous implementation uses a hacky way to do this and may lead to bugs (#3411), however, we can directly mutate the source chunk arm of the stream.This PR also refines some code styles.
Checklist
./risedev check(or alias,./risedev c)Refer to a related PR or issue link (optional)
nextlast forever and cannot process barrier #3411