feat(streaming): Support hash based parallelized chain node #1846

zbzbw · 2022-04-14T10:30:21Z

What's changed and what's your intention?

This PR implements parallelized chain node in a very straightforward and naive way:

Changed chain node's distribution, and assign each chain with one upstream materialize actor.
For batch query node, we split data into multiple parts by hashing. So each executor will still scan the whole table

We should change the batch query node to scan the table by range after we figured out how to split table into partitions in a good way.

p.s. Dashboard has some small issue now when resolving mv on mv.

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)

One step forward of #619

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

zbzbw · 2022-04-14T10:56:44Z

Some illustrations.

create table t1 (v1 int not null, v2 int not null, v3 int not null);
create materialized view mv1 as select a.v1 as av1, b.v1 as bv1 from t1 a, t1 b where a.v2<>b.v2 and a.v3=b.v3;

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

codecov · 2022-04-14T10:59:47Z

Codecov Report

Merging #1846 (e488bf5) into main (c14ebe4) will decrease coverage by 0.02%.
The diff coverage is 40.65%.

@@            Coverage Diff             @@
##             main    #1846      +/-   ##
==========================================
- Coverage   70.86%   70.84%   -0.03%     
==========================================
  Files         611      611              
  Lines       79591    79667      +76     
==========================================
+ Hits        56403    56440      +37     
- Misses      23188    23227      +39

Flag	Coverage Δ
rust	`70.84% <40.65%> (-0.03%)`	⬇️

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

Impacted Files	Coverage Δ
src/meta/src/barrier/command.rs	`58.87% <ø> (+0.54%)`	⬆️
src/meta/src/rpc/service/ddl_service.rs	`0.00% <0.00%> (ø)`
src/meta/src/rpc/service/stream_service.rs	`0.00% <0.00%> (ø)`
src/meta/src/stream/stream_manager.rs	`72.50% <ø> (ø)`
src/stream/src/executor/batch_query.rs	`0.00% <0.00%> (ø)`
src/stream/src/executor_v2/v1_compat.rs	`37.44% <0.00%> (-1.12%)`	⬇️
src/meta/src/stream/graph/stream_graph.rs	`62.36% <21.42%> (-4.31%)`	⬇️
src/meta/src/stream/fragmenter.rs	`83.11% <80.00%> (+0.16%)`	⬆️
src/stream/src/executor_v2/batch_query.rs	`85.71% <95.65%> (+3.74%)`	⬆️
...ntend/src/optimizer/plan_node/stream_table_scan.rs	`95.00% <100.00%> (+0.08%)`	⬆️
... and 6 more

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

zbzbw · 2022-04-14T11:15:37Z

Seems we have too much log now 😢

skyzh

Not sure if this implementation is correct. Would you please elaborate:

What distribution is chain following?
What's the distribution of BatchPlanNode and MergeNode separately? (Is BatchPlanNode really using Distribution::HashShard(logical.base.pk_indices.clone()) as its distribution?)
What's distribution is chain's dispatcher? How is it determined?

skyzh · 2022-04-14T11:14:42Z

src/stream/src/executor_v2/batch_query.rs

+            .get_hash_values(self.info.pk_indices.as_ref(), CRC32FastBuilder)
+            .unwrap();
+        let n = data_chunk.cardinality();
+        let (columns, _visibility) = data_chunk.into_parts();


can we ensure that data_chunk's visibility is None?

Concern about this as well. By the way, there're also some other executors ignoring the visibility. :(

AFAIK, collect_data_chunk in CellBasedTableRowIter will always return a chunk with None visibility.

By the way, there're also some other executors ignoring the visibility. :(

Added compact in execute_inner

pls add an assert that visibility is None.

src/frontend/test_runner/tests/testdata/basic_query_2.yaml

src/stream/src/executor_v2/batch_query.rs

BugenZhao · 2022-04-15T02:56:50Z

src/stream/src/executor_v2/batch_query.rs

+            .get_hash_values(self.info.pk_indices.as_ref(), CRC32FastBuilder)
+            .unwrap();
+        let n = data_chunk.cardinality();
+        let (columns, _visibility) = data_chunk.into_parts();


Concern about this as well. By the way, there're also some other executors ignoring the visibility. :(

BugenZhao · 2022-04-15T03:10:07Z

src/frontend/src/optimizer/plan_node/stream_table_scan.rs

@@ -42,7 +42,7 @@ impl StreamTableScan {
            ctx,
            logical.schema().clone(),
            logical.base.pk_indices.clone(),
-            Distribution::Single,
+            Distribution::HashShard(logical.base.pk_indices.clone()),


Could you please also change the distribution in the Java frontend? I'm afraid the current e2e result cannot cover some cases.

I've added a workaround for Java frontend in fragmenter (because I'm not familiar with Java part 😅). Will remove this after we deprecate Java frontend.

zbzbw · 2022-04-15T03:18:58Z

I probably didn't catch up the multi-dispatcher part 😢, will reopen it later.

BugenZhao · 2022-04-15T03:22:10Z

According to the dashboard graph, there's still a hash dispatcher after each chain. So it seems the distribution of Chain is not used. The final solution should be inserting exchanges after the BatchQuery or using the partition scan to ensure the same distribution between the chain and the scan. You may check this doc for more details.

Anyway, as long as there's exchange after chain, the result will be correct.

skyzh · 2022-04-15T03:32:54Z

I probably didn't catch up the multi-dispatcher part 😢, will reopen it later.

Multi-dispatcher can definitely help the implementation of this PR. But it's not well-tested yet -- at least compute-node doesn't support multi-dispatcher.

A possible approach is to always follow the distribution of upstream materialize executor, and therefore, the dispatcher for them can be "broadcast", and downstream need shuffle after table scan.

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

yezizp2012 · 2022-04-15T05:51:23Z

You can force a chain singleton to be specified in the StreamManagerService on the meta (these requests come from the Java frontend) to avoid modifying the Java frontend. By this way the workload is minimal and pass the Java e2e test.

fuyufjh · 2022-04-15T07:32:04Z

We should change the batch query node to scan the table by range after we figured out how to split table into partitions in a good way.

Please just use consistent hashing to partition the batch query. See more on Proposal: Use Consistent Hash Across the System.

Also, in this way, no exchange would be needed for Chain.

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

skyzh · 2022-04-15T08:33:14Z

Also, in this way, no exchange would be needed for Chain.

No, we still need it.

We cannot guarantee that upstream has the same actor number as chain actors.
Even if we have the same actor, currently we didn't guarantee that materialize node uses hash distribution by hash of materialize key. Currently, it simply follows input's distribution.

https://github.com/singularity-data/risingwave/blob/7f9911a8603b06014bdb46efee55766f49ea5064/src/frontend/src/optimizer/plan_node/stream_materialize.rs#L55

skyzh · 2022-04-15T08:35:03Z

Also, materialize stream node is created after enforcing distribution. Need refactor the create MV optimize process to make everything work.

https://github.com/singularity-data/risingwave/blob/7f9911a8603b06014bdb46efee55766f49ea5064/src/frontend/src/optimizer/mod.rs#L200-L206

yezizp2012 · 2022-04-15T08:35:49Z

After offline discussion, we will merge this PR first to make it runnable. After this, @zbzbw will try to refine the batch query scan logic to adjust consistent hashing distribution in depended mv.

fuyufjh · 2022-04-15T08:36:02Z

We cannot guarantee that upstream has the same actor number as chain actors.

Agree.

Even if we have the same actor, currently we didn't guarantee that materialize node uses hash distribution by hash of materialize key. Currently, it simply follows input's distribution.

Hmmmm... Should be guaranteed, I think.

fuyufjh · 2022-04-15T08:40:20Z

After offline discussion, we will merge this PR first to make it runnable. After this, @zbzbw will try to refine the batch query scan logic to adjust consistent hashing distribution in depended mv.

For others not in the discussion:

The stream merged from batch query and upstream looks weird to me because it's under a weird distribution, neither distributed by streaming nor batch.

We may refine this later by letting the batch query scan data with the same distribution of the upstream Materialize. In this way, the distribution can be expressed as Hash(xxx) in optimizer, and optimizer can further determine whether it needs another Exchange.

yezizp2012 · 2022-04-15T08:43:14Z

Even if we have the same actor, currently we didn't guarantee that materialize node uses hash distribution by hash of materialize key. Currently, it simply follows input's distribution.

https://github.com/singularity-data/risingwave/blob/7f9911a8603b06014bdb46efee55766f49ea5064/src/frontend/src/optimizer/plan_node/stream_materialize.rs#L55

That's what I concerned before too. 🤔 So we choice to add an exchange right after chain in current implementation.

skyzh · 2022-04-15T08:47:06Z

merge?

Bowen Zhou added 2 commits April 14, 2022 14:42

cherry-pick parallel query, discarding fragmenter changes

cc6dfb3

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

impl & refactor

b54b77e

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

zbzbw requested review from skyzh, yezizp2012, MrCroxx and BugenZhao April 14, 2022 10:30

zbzbw and others added 2 commits April 14, 2022 18:37

Update stream_table_scan.rs

5ce90e7

Merge remote-tracking branch 'origin/main' into zbw/parallel-chain-v2

8f893a1

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

github-actions bot added the type/feature label Apr 14, 2022

zbzbw requested a review from TennyZhuang April 14, 2022 10:57

zbzbw marked this pull request as ready for review April 14, 2022 10:57

fix proto fmt

bbb0648

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

skyzh reviewed Apr 14, 2022

View reviewed changes

BugenZhao reviewed Apr 15, 2022

View reviewed changes

zbzbw marked this pull request as draft April 15, 2022 03:12

Bowen Zhou added 2 commits April 15, 2022 12:04

change distribution to AnyShard

61e3cfe

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

some fixes

62cf37c

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

Bowen Zhou and others added 3 commits April 15, 2022 15:36

support java frontend with workaround

d428983

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

fmt

76a65fc

Signed-off-by: Bowen Zhou <bowenzhou@singularity-data.com>

Merge branch 'main' into zbw/parallel-chain-v2

e488bf5

zbzbw marked this pull request as ready for review April 15, 2022 08:25

yezizp2012 approved these changes Apr 15, 2022

View reviewed changes

zbzbw mentioned this pull request Apr 15, 2022

Streaming: use consistent hashing in batch query node #1875

Closed

skyzh approved these changes Apr 15, 2022

View reviewed changes

zbzbw merged commit ae1c231 into main Apr 15, 2022

zbzbw deleted the zbw/parallel-chain-v2 branch April 15, 2022 08:47

MrCroxx mentioned this pull request Jul 22, 2022

Tracking: Parallelized query on MV #619

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(streaming): Support hash based parallelized chain node #1846

feat(streaming): Support hash based parallelized chain node #1846

zbzbw commented Apr 14, 2022 •

edited

zbzbw commented Apr 14, 2022

codecov bot commented Apr 14, 2022 •

edited

zbzbw commented Apr 14, 2022

skyzh left a comment •

edited

skyzh Apr 14, 2022

BugenZhao Apr 15, 2022 •

edited

zbzbw Apr 15, 2022

skyzh Apr 15, 2022

BugenZhao Apr 15, 2022 •

edited

BugenZhao Apr 15, 2022 •

edited

zbzbw Apr 15, 2022

zbzbw commented Apr 15, 2022

BugenZhao commented Apr 15, 2022

skyzh commented Apr 15, 2022

yezizp2012 commented Apr 15, 2022 •

edited

fuyufjh commented Apr 15, 2022 •

edited

skyzh commented Apr 15, 2022 •

edited

skyzh commented Apr 15, 2022

yezizp2012 commented Apr 15, 2022

fuyufjh commented Apr 15, 2022

fuyufjh commented Apr 15, 2022 •

edited

yezizp2012 commented Apr 15, 2022 •

edited

skyzh commented Apr 15, 2022

feat(streaming): Support hash based parallelized chain node #1846

feat(streaming): Support hash based parallelized chain node #1846

Conversation

zbzbw commented Apr 14, 2022 • edited

What's changed and what's your intention?

Checklist

Refer to a related PR or issue link (optional)

zbzbw commented Apr 14, 2022

codecov bot commented Apr 14, 2022 • edited

Codecov Report

zbzbw commented Apr 14, 2022

skyzh left a comment • edited

Choose a reason for hiding this comment

skyzh Apr 14, 2022

Choose a reason for hiding this comment

BugenZhao Apr 15, 2022 • edited

Choose a reason for hiding this comment

zbzbw Apr 15, 2022

Choose a reason for hiding this comment

skyzh Apr 15, 2022

Choose a reason for hiding this comment

BugenZhao Apr 15, 2022 • edited

Choose a reason for hiding this comment

BugenZhao Apr 15, 2022 • edited

Choose a reason for hiding this comment

zbzbw Apr 15, 2022

Choose a reason for hiding this comment

zbzbw commented Apr 15, 2022

BugenZhao commented Apr 15, 2022

skyzh commented Apr 15, 2022

yezizp2012 commented Apr 15, 2022 • edited

fuyufjh commented Apr 15, 2022 • edited

skyzh commented Apr 15, 2022 • edited

skyzh commented Apr 15, 2022

yezizp2012 commented Apr 15, 2022

fuyufjh commented Apr 15, 2022

fuyufjh commented Apr 15, 2022 • edited

yezizp2012 commented Apr 15, 2022 • edited

skyzh commented Apr 15, 2022

zbzbw commented Apr 14, 2022 •

edited

codecov bot commented Apr 14, 2022 •

edited

skyzh left a comment •

edited

BugenZhao Apr 15, 2022 •

edited

BugenZhao Apr 15, 2022 •

edited

BugenZhao Apr 15, 2022 •

edited

yezizp2012 commented Apr 15, 2022 •

edited

fuyufjh commented Apr 15, 2022 •

edited

skyzh commented Apr 15, 2022 •

edited

fuyufjh commented Apr 15, 2022 •

edited

yezizp2012 commented Apr 15, 2022 •

edited