feat(stream): implement streaming part of now executor #6408

soundOfDestiny · 2022-11-16T14:43:20Z

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

Summarize your change (mandatory)
implement streaming part of now executor, which will return current time now()
How does this PR work? Need a brief introduction for the changed logic (optional)
emit on most barrier
Describe any limitations of the current code (optional)
frontend part not done

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
All checks passed in ./risedev check (or alias, ./risedev c)

Documentation

If your pull request contains user-facing changes, please specify the types of the changes, and create a release note. Otherwise, please feel free to remove this section.

Types of user-facing changes

Please keep the types that apply to your changes, and remove those that do not apply.

Installation and deployment
Connector (sources & sinks)
SQL commands, functions, and operators
RisingWave cluster configuration changes
Other (please specify in the release note below)

Release note

Please create a release note for your changes. In the release note, focus on the impact on users, and mention the environment or conditions where the impact may occur.

Refer to a related PR or issue link (optional)

#6407

codecov · 2022-11-16T15:25:53Z

Codecov Report

Merging #6408 (fe1aeb9) into main (5a0fad1) will increase coverage by 0.00%.
The diff coverage is 79.37%.

@@           Coverage Diff            @@
##             main    #6408    +/-   ##
========================================
  Coverage   73.29%   73.29%            
========================================
  Files        1012     1014     +2     
  Lines      162172   162331   +159     
========================================
+ Hits       118862   118987   +125     
- Misses      43310    43344    +34

Flag	Coverage Δ
rust	`73.29% <79.37%> (+<0.01%)`	⬆️

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

Impacted Files	Coverage Δ
src/stream/src/from_proto/mod.rs	`0.00% <ø> (ø)`
src/stream/src/from_proto/now.rs	`0.00% <0.00%> (ø)`
src/stream/src/executor/mod.rs	`49.07% <78.57%> (+1.13%)`	⬆️
src/stream/src/executor/now.rs	`91.05% <91.05%> (ø)`
src/common/src/util/epoch.rs	`90.32% <100.00%> (+0.32%)`	⬆️
src/source/src/row_id.rs	`91.01% <0.00%> (-1.13%)`	⬇️
src/common/src/field_generator/mod.rs	`80.10% <0.00%> (-0.54%)`	⬇️
src/stream/src/executor/aggregation/minput.rs	`96.49% <0.00%> (+0.10%)`	⬆️

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

soundOfDestiny · 2022-11-16T15:25:55Z

we should use barrier_receiver

soundOfDestiny · 2022-11-16T15:41:03Z

we should use barrier_receiver

done

src/common/src/util/epoch.rs

dashboard/package.json

dashboard/package-lock.json

src/stream/src/executor/mod.rs

TennyZhuang · 2022-11-22T07:32:20Z

related #6472

BugenZhao

Rest LGTM.

BugenZhao · 2022-11-23T04:24:46Z

src/stream/src/executor/now.rs

+        let mut last_timestamp = state_row.and_then(|row| row[0].clone());
+
+        while let Some(barrier) = barrier_receiver.recv().await {
+            if !barrier.is_update() && !barrier.is_resume() {


Suggest maintaining a paused state instead of relying on the assumption of "pause -> update -> resume`, in case we have other barriers inside.

Suggest maintaining a paused state instead of relying on the assumption of "pause -> update -> resume`, in case we have other barriers inside.

fixed

BugenZhao · 2022-11-23T04:25:36Z

src/stream/src/executor/now.rs

+        state_table.init_epoch(barrier.epoch);
+
+        // The first barrier message should be propagated.
+        yield Message::Barrier(barrier);


So we don't emit the timestamp on the first barrier? Is this expected?

So we don't emit the timestamp on the first barrier? Is this expected?

We can't because no message is permitted before the first barrier.

How about emitting the timestamp right after the barrier?

How about emitting the timestamp right after the barrier?

The timestamp will be updated on the second barrier so I think that it is unnecessary.

BugenZhao · 2022-11-23T04:28:01Z

src/stream/src/executor/now.rs

+                if last_timestamp.is_some() {
+                    let chunk_popped = data_chunk_builder
+                        .append_one_row_from_datums([&last_timestamp].into_iter());
+                    debug_assert!(chunk_popped.is_none());
+                }
+                let data_chunk = data_chunk_builder
+                    .append_one_row_from_datums([&timestamp].into_iter())
+                    .unwrap();


Directly use DataChunk::from_rows?

Directly use DataChunk::from_rows?

fixed

TennyZhuang · 2022-11-30T06:26:56Z

Generally LGTM.

I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

soundOfDestiny · 2022-11-30T07:20:13Z

Generally LGTM.

I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

There will be too much things in this PR:

backend part
frontend part
barrier part because we need to modify meta to let it send barriers to actors which have Now Executor

The latter two cannot be placed in a separate PR before this because they need to use Now Executor.

If it is recommended, I will put these into this PR.

TennyZhuang · 2022-11-30T10:05:47Z

Generally LGTM.
I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

There will be too much things in this PR:

backend part

frontend part

barrier part because we need to modify meta to let it send barriers to actors which have Now Executor

The latter two cannot be placed in a separate PR before this because they need to use Now Executor.

If it is recommended, I will put these into this PR.

In multiple PRs, but merge at the same time?

soundOfDestiny · 2022-11-30T10:28:15Z

Generally LGTM.
I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

There will be too much things in this PR:

backend part

frontend part

barrier part because we need to modify meta to let it send barriers to actors which have Now Executor

The latter two cannot be placed in a separate PR before this because they need to use Now Executor.
If it is recommended, I will put these into this PR.

In multiple PRs, but merge at the same time?

Do you mean that create multiple PRs and other PRs also include codes of NowExecutor in this PR?

soundOfDestiny · 2022-11-30T14:44:21Z

Generally LGTM.
I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

There will be too much things in this PR:

backend part

frontend part

barrier part because we need to modify meta to let it send barriers to actors which have Now Executor

The latter two cannot be placed in a separate PR before this because they need to use Now Executor.
If it is recommended, I will put these into this PR.

In multiple PRs, but merge at the same time?

Therefore let's work on this base PR: #6677 of following multiple PRs first. It defines interfaces of Now.

jon-chuang · 2022-12-01T03:55:11Z

I'd prefer to hold the PR for a while, and wait for the frontend part, so that we can test e2e with dynamic filter.

Actually, if we expect this PR is > 90% correct, then I don't see the harm of merging first, and then once we have e2e with dynamic filter, we can fix any bugs we encounter. And incremental change is easier to review.

IMO Now executor is also independent of dynamic filter.

wdyt @TennyZhuang ?

jon-chuang

LGTM

* feat(stream): implement now executor (close #6407) * use if let * use barrier receiver * timestamp sanity check * add now executor builder * fix gen proto * rebase main Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

soundOfDestiny requested review from wangrunji0408, BugenZhao and yezizp2012 November 16, 2022 14:43

github-actions bot added the type/feature label Nov 16, 2022

soundOfDestiny requested a review from TennyZhuang November 16, 2022 14:44

soundOfDestiny force-pushed the zl_now branch from 24845d9 to 9100591 Compare November 16, 2022 14:46

soundOfDestiny marked this pull request as draft November 16, 2022 15:18

soundOfDestiny marked this pull request as ready for review November 16, 2022 15:33

soundOfDestiny force-pushed the zl_now branch from 386cb81 to ec1ddf4 Compare November 16, 2022 15:34

TennyZhuang requested a review from st1page November 16, 2022 15:48

soundOfDestiny force-pushed the zl_now branch from ce2b319 to 42a69db Compare November 17, 2022 03:24

jon-chuang reviewed Nov 17, 2022

View reviewed changes

src/common/src/util/epoch.rs Show resolved Hide resolved

soundOfDestiny force-pushed the zl_now branch from 42a69db to e0fc47c Compare November 17, 2022 04:14

TennyZhuang reviewed Nov 18, 2022

View reviewed changes

dashboard/package.json Outdated Show resolved Hide resolved

dashboard/package-lock.json Outdated Show resolved Hide resolved

src/stream/src/executor/mod.rs Outdated Show resolved Hide resolved

soundOfDestiny force-pushed the zl_now branch 3 times, most recently from 6cb4643 to de1d1bf Compare November 18, 2022 07:30

soundOfDestiny mentioned this pull request Nov 21, 2022

streaming: Now Executor for TemporalFilter #6407

Closed

6 tasks

soundOfDestiny requested a review from TennyZhuang November 22, 2022 07:01

soundOfDestiny requested a review from jon-chuang November 22, 2022 07:39

soundOfDestiny force-pushed the zl_now branch 3 times, most recently from 29a6845 to 794cec8 Compare November 22, 2022 11:28

BugenZhao reviewed Nov 23, 2022

View reviewed changes

soundOfDestiny requested a review from BugenZhao November 23, 2022 08:16

soundOfDestiny force-pushed the zl_now branch from 210f748 to 2fc2346 Compare November 23, 2022 08:22

soundOfDestiny force-pushed the zl_now branch 3 times, most recently from 73cbf0f to 2dfbafd Compare November 24, 2022 08:38

Liang Zhao added 6 commits November 24, 2022 16:44

feat(stream): implement now executor (close #6407)

ab85913

use if let

3becad8

use barrier receiver

68de931

timestamp sanity check

af41b83

add now executor builder

e4e4a39

fix gen proto

cde35d6

soundOfDestiny force-pushed the zl_now branch from 2dfbafd to cde35d6 Compare November 24, 2022 08:46

rebase main

26d572f

soundOfDestiny mentioned this pull request Nov 30, 2022

feat(stream): implement framework of now executor #6677

Closed

2 tasks

jon-chuang approved these changes Dec 1, 2022

View reviewed changes

TennyZhuang approved these changes Dec 1, 2022

View reviewed changes

soundOfDestiny added the mergify/can-merge label Dec 2, 2022

Merge branch 'main' into zl_now

fe1aeb9

mergify bot merged commit 240aaa9 into main Dec 2, 2022

mergify bot deleted the zl_now branch December 2, 2022 01:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(stream): implement streaming part of now executor #6408

feat(stream): implement streaming part of now executor #6408

soundOfDestiny commented Nov 16, 2022 •

edited

Loading

codecov bot commented Nov 16, 2022 •

edited

Loading

soundOfDestiny commented Nov 16, 2022

soundOfDestiny commented Nov 16, 2022

TennyZhuang commented Nov 22, 2022

BugenZhao left a comment

BugenZhao Nov 23, 2022

soundOfDestiny Nov 23, 2022

BugenZhao Nov 23, 2022

soundOfDestiny Nov 23, 2022

BugenZhao Nov 23, 2022

soundOfDestiny Nov 23, 2022

BugenZhao Nov 23, 2022

soundOfDestiny Nov 23, 2022

TennyZhuang commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022 •

edited

Loading

TennyZhuang commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022

jon-chuang commented Dec 1, 2022

jon-chuang left a comment

feat(stream): implement streaming part of now executor #6408

feat(stream): implement streaming part of now executor #6408

Conversation

soundOfDestiny commented Nov 16, 2022 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Types of user-facing changes

Release note

Refer to a related PR or issue link (optional)

codecov bot commented Nov 16, 2022 • edited Loading

Codecov Report

soundOfDestiny commented Nov 16, 2022

soundOfDestiny commented Nov 16, 2022

TennyZhuang commented Nov 22, 2022

BugenZhao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TennyZhuang commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022 • edited Loading

TennyZhuang commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022

soundOfDestiny commented Nov 30, 2022

jon-chuang commented Dec 1, 2022

jon-chuang left a comment

Choose a reason for hiding this comment

soundOfDestiny commented Nov 16, 2022 •

edited

Loading

codecov bot commented Nov 16, 2022 •

edited

Loading

soundOfDestiny commented Nov 30, 2022 •

edited

Loading