refactor(batch): Use futures-async-stream to implement HashJoin executor #2119

D2Lark · 2022-04-25T08:39:09Z

What's changed and what's your intention?

Implement HashJoin using Executor2 trait.

Please explain IN DETAIL what the changes are in this PR and why they are needed:

Summarize your change (mandatory)
How does this PR work? Need a brief introduction for the changed logic (optional)
Describe clearly one logical change and avoid lazy messages (optional)
Describe any limitations of the current code (optional)

Checklist

I have written necessary docs and comments
I have added necessary unit tests and integration tests

Refer to a related PR or issue link (optional)

close #1947

Signed-off-by: d2lark <lichengamoy@gmail.com>

…tor2

D2Lark · 2022-04-25T08:41:45Z

There are still some unresolved issues, such as Calling child.execute() twice at the same time shows the problem of ownership

--> src/batch/src/executor2/join/hash_join.rs:174:17
|
170 | let mut right_child_stream = self.right_child.execute();
| --------- self.right_child partially moved due to this method call
...
174 | self.build(build_table, right_child_stream).await?
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ value borrowed here after partial move
|
note: this function takes ownership of the receiver self, which moves self.right_child
--> src/batch/src/executor2/mod.rs:67:16
|
67 | fn execute(self: Box) -> BoxedDataChunkStream;
| ^^^^
= note: partial move occurs because self.right_child has type std::boxed::Box<dyn executor2::Executor2>, which does not implement the Copy trait

error[E0382]: use of moved value: left_child_stream

liurenjie1024 · 2022-04-25T09:09:56Z

src/batch/src/executor2/join/chunked_data.rs

@@ -0,0 +1,212 @@
+// Copyright 2022 Singularity Data


We should also remove chunked_data in executor/join mod?

I tried this once but had some problems, if I delete executor/join/chunked_data, many functions and structures under executor2/join/chunked_data have to be changed to pub, so I keep executor/join/chunked_data for now and delete it after all of them have been ported?

liurenjie1024 · 2022-04-25T09:19:03Z

src/batch/src/executor2/join/hash_join.rs

+    #[try_stream(boxed, ok = DataChunk, error = RwError)]
+    async fn do_execute(mut self: Box<Self>) {
+        let mut left_child_stream = self.left_child.execute();
+        let mut right_child_stream = self.right_child.execute();


The execution of right child should be postponed , we should execute left child to build table first.

liurenjie1024 · 2022-04-25T09:22:52Z

src/batch/src/executor2/join/hash_join.rs

    /// Build side
-    right_child: BoxedExecutor,
+    right_child: BoxedExecutor2,
    state: HashJoinState<K>,


In fact, after using stream api, we no longer need this state machine. We just need to keep BuildTable and ProbeTable<K> as local varialbe in do_execute method.

From my simple understanding, the previous control flow open next close caused a lot of variables to be bound to the executor and the whole logic was very complicated. So this morning I made a version of the risinglight implementation and simplified it

Thanks for you contribution, I'll take a review.

In fact, after using stream api, we no longer need this state machine. We just need to keep BuildTable and ProbeTable as local varialbe in do_execute method.

I see what you mean, I'll optimize it

liurenjie1024 · 2022-04-25T09:26:19Z

Yes, execute will take the ownership of executor.

D2Lark · 2022-04-25T09:30:11Z

Yes, execute will take the ownership of the executor.

yes, because of that we usually put the execute at the end of func. So far there are no problems in the non-join executor. But every join executor has two children to take their father's ownership , What to do with this？

skyzh · 2022-04-25T15:01:56Z

Yes, execute will take the ownership of the executor.

yes, because of that we usually put the execute at the end of func. So far there are no problems in the non-join executor. But every join executor has two children to take their father's ownership , What to do with this？

Just set them as Option<BoxedExecutor> and use left.take().unwrap() to get the ownership.

D2Lark · 2022-04-25T23:34:22Z

Yes, execute will take the ownership of the executor.

yes, because of that we usually put the execute at the end of func. So far there are no problems in the non-join executor. But every join executor has two children to take their father's ownership , What to do with this？

Just set them as Option<BoxedExecutor> and use left.take().unwrap() to get the ownership.

Thanks for the tip~

Signed-off-by: d2lark <lichengamoy@gmail.com>

…tor2

Signed-off-by: d2lark <lichengamoy@gmail.com>

…tor2

Signed-off-by: d2lark <lichengamoy@gmail.com>

…tor2

Signed-off-by: d2lark <lichengamoy@gmail.com>

TennyZhuang · 2022-04-28T05:39:07Z

CI running https://github.com/singularity-data/risingwave/actions/runs/2237323815

TennyZhuang · 2022-04-28T05:58:56Z

rustdoc check failed @D2Lark

cargo doc --document-private-items --no-deps

https://github.com/singularity-data/risingwave/runs/6205527285?check_suite_focus=true

…tor2

Signed-off-by: d2lark <lichengamoy@gmail.com>

liurenjie1024

LGTM

TennyZhuang · 2022-04-28T09:29:20Z

https://github.com/singularity-data/risingwave/actions/runs/2238380756 CI running

codecov · 2022-04-28T09:37:08Z

Codecov Report

Merging #2119 (69b5eb4) into main (61c645e) will decrease coverage by 0.01%.
The diff coverage is 85.57%.

@@            Coverage Diff             @@
##             main    #2119      +/-   ##
==========================================
- Coverage   70.92%   70.90%   -0.02%     
==========================================
  Files         650      652       +2     
  Lines       82725    82790      +65     
==========================================
+ Hits        58671    58705      +34     
- Misses      24054    24085      +31

Flag	Coverage Δ
rust	`70.90% <85.57%> (-0.02%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/batch/src/executor/join/mod.rs	`13.15% <ø> (-50.00%)`	⬇️
src/batch/src/executor/join/nested_loop_join.rs	`82.17% <ø> (ø)`
src/batch/src/executor/mod.rs	`60.46% <ø> (ø)`
src/batch/src/executor2/join/hash_join_state.rs	`88.78% <ø> (ø)`
src/batch/src/executor2/mod.rs	`100.00% <ø> (ø)`
src/batch/src/executor2/join/mod.rs	`60.52% <60.52%> (ø)`
src/batch/src/executor2/join/hash_join.rs	`85.32% <62.85%> (ø)`
src/batch/src/executor2/join/chunked_data.rs	`99.21% <99.21%> (ø)`
src/batch/src/executor/join/chunked_data.rs	`92.96% <0.00%> (-6.25%)`	⬇️
... and 2 more

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

yuhao-su · 2022-04-28T09:40:34Z

LGTM

Signed-off-by: d2lark <lichengamoy@gmail.com>

…tor2

TennyZhuang · 2022-04-28T10:19:25Z

https://github.com/singularity-data/risingwave/actions/runs/2238638430

liurenjie1024 · 2022-04-28T10:38:03Z

https://github.com/singularity-data/risingwave/actions/runs/2238638430

The check passed, merged.

D2Lark · 2022-04-29T01:32:30Z

src/batch/src/executor2/join/hash_join.rs

-        while let Some(chunk) = self.right_child.next().await? {
+impl<K: HashKey + Send + Sync> HashJoinExecutor2<K> {
+    #[try_stream(boxed, ok = DataChunk, error = RwError)]
+    async fn do_execute(mut self: Box<Self>) {


I don't think this is an elegant implementation and I'd like to refactor it

D2Lark added 2 commits April 25, 2022 08:35

impl hash_join exexcutor2 (1/2)

03029ae

Signed-off-by: d2lark <lichengamoy@gmail.com>

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

1d0858f

…tor2

liurenjie1024 reviewed Apr 25, 2022

View reviewed changes

D2Lark added 7 commits April 25, 2022 23:57

use option wrap child

3f03bc1

Signed-off-by: d2lark <lichengamoy@gmail.com>

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

8824faa

…tor2

use local var to control process instead of state

3f39a68

Signed-off-by: d2lark <lichengamoy@gmail.com>

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

5c5dd92

…tor2

use phantomData

62c9e05

Signed-off-by: d2lark <lichengamoy@gmail.com>

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

f1457cd

…tor2

rename stat enum

00581ab

Signed-off-by: d2lark <lichengamoy@gmail.com>

TennyZhuang requested a review from yuhao-su April 28, 2022 05:57

D2Lark added 2 commits April 28, 2022 06:30

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

7175158

…tor2

doc fic

0ff7dfc

Signed-off-by: d2lark <lichengamoy@gmail.com>

liurenjie1024 approved these changes Apr 28, 2022

View reviewed changes

yuhao-su approved these changes Apr 28, 2022

View reviewed changes

D2Lark added 2 commits April 28, 2022 09:52

cargo fmt

5912d33

Signed-off-by: d2lark <lichengamoy@gmail.com>

Merge remote-tracking branch 'upstream/main' into impl_hashjoin_execu…

69b5eb4

…tor2

liurenjie1024 merged commit cb72ab3 into risingwavelabs:main Apr 28, 2022

D2Lark deleted the impl_hashjoin_executor2 branch April 28, 2022 10:38

D2Lark commented Apr 29, 2022

View reviewed changes

D2Lark mentioned this pull request May 4, 2022

refactor(batch): Use futures-async-stream to implement NestedLoopJoin executor #2217

Merged

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(batch): Use futures-async-stream to implement HashJoin executor #2119

refactor(batch): Use futures-async-stream to implement HashJoin executor #2119

D2Lark commented Apr 25, 2022

D2Lark commented Apr 25, 2022

liurenjie1024 Apr 25, 2022

D2Lark Apr 25, 2022

liurenjie1024 Apr 25, 2022

liurenjie1024 Apr 25, 2022

D2Lark Apr 25, 2022

liurenjie1024 Apr 25, 2022

D2Lark Apr 25, 2022 •

edited

liurenjie1024 commented Apr 25, 2022

D2Lark commented Apr 25, 2022 •

edited

skyzh commented Apr 25, 2022

D2Lark commented Apr 25, 2022

TennyZhuang commented Apr 28, 2022

TennyZhuang commented Apr 28, 2022 •

edited

liurenjie1024 left a comment

TennyZhuang commented Apr 28, 2022

codecov bot commented Apr 28, 2022 •

edited

yuhao-su commented Apr 28, 2022

TennyZhuang commented Apr 28, 2022

liurenjie1024 commented Apr 28, 2022

D2Lark Apr 29, 2022

refactor(batch): Use futures-async-stream to implement HashJoin executor #2119

refactor(batch): Use futures-async-stream to implement HashJoin executor #2119

Conversation

D2Lark commented Apr 25, 2022

What's changed and what's your intention?

Checklist

Refer to a related PR or issue link (optional)

D2Lark commented Apr 25, 2022

liurenjie1024 Apr 25, 2022

Choose a reason for hiding this comment

D2Lark Apr 25, 2022

Choose a reason for hiding this comment

liurenjie1024 Apr 25, 2022

Choose a reason for hiding this comment

liurenjie1024 Apr 25, 2022

Choose a reason for hiding this comment

D2Lark Apr 25, 2022

Choose a reason for hiding this comment

liurenjie1024 Apr 25, 2022

Choose a reason for hiding this comment

D2Lark Apr 25, 2022 • edited

Choose a reason for hiding this comment

liurenjie1024 commented Apr 25, 2022

D2Lark commented Apr 25, 2022 • edited

skyzh commented Apr 25, 2022

D2Lark commented Apr 25, 2022

TennyZhuang commented Apr 28, 2022

TennyZhuang commented Apr 28, 2022 • edited

liurenjie1024 left a comment

Choose a reason for hiding this comment

TennyZhuang commented Apr 28, 2022

codecov bot commented Apr 28, 2022 • edited

Codecov Report

yuhao-su commented Apr 28, 2022

TennyZhuang commented Apr 28, 2022

liurenjie1024 commented Apr 28, 2022

D2Lark Apr 29, 2022

Choose a reason for hiding this comment

D2Lark Apr 25, 2022 •

edited

D2Lark commented Apr 25, 2022 •

edited

TennyZhuang commented Apr 28, 2022 •

edited

codecov bot commented Apr 28, 2022 •

edited