feat(streaming): introduce watermark filter #6224

yuhao-su · 2022-11-07T06:36:55Z

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

What's changed and what's your intention?

The executor will generate a Watermark after each chunk.
This will also guarantee all later rows with event time less than the watermark will be filtered.

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)

Refer to a related PR or issue link (optional)

close #6046

…troduce_watermark_filter

yuhao-su · 2022-11-07T06:57:58Z

src/common/src/types/mod.rs

+    pub fn min(&self) -> Datum {
+        match self {
+            DataType::Int16 => Some(ScalarImpl::Int16(i16::MIN)),
+            DataType::Int32 => Some(ScalarImpl::Int32(i32::MIN)),
+            DataType::Int64 => Some(ScalarImpl::Int64(i64::MIN)),
+            DataType::Float32 => Some(ScalarImpl::Float32(OrderedF32::neg_infinity())),
+            DataType::Float64 => Some(ScalarImpl::Float64(OrderedF64::neg_infinity())),
+            DataType::Boolean => Some(ScalarImpl::Bool(false)),
+            DataType::Varchar => Some(ScalarImpl::Utf8("".to_string())),
+            DataType::Date => Some(ScalarImpl::NaiveDate(NaiveDateWrapper(NaiveDate::MIN))),
+            DataType::Time => Some(ScalarImpl::NaiveTime(NaiveTimeWrapper(
+                NaiveTime::from_hms(0, 0, 0),
+            ))),
+            DataType::Timestamp => Some(ScalarImpl::NaiveDateTime(NaiveDateTimeWrapper(
+                NaiveDateTime::MIN,
+            ))),
+            // FIXME(yuhao): Add a timestampz scalar.
+            DataType::Timestampz => Some(ScalarImpl::Int64(i64::MIN)),
+            DataType::Decimal => Some(ScalarImpl::Decimal(Decimal::NegativeInf)),
+            DataType::Interval => Some(ScalarImpl::Interval(IntervalUnit::min())),
+            DataType::Struct { .. } => Some(ScalarImpl::Struct(StructValue::new(vec![]))),
+            DataType::List { .. } => Some(ScalarImpl::List(ListValue::new(vec![]))),
+        }
+    }


PTAL @xiangjinwu

We may need to recursively call this function to match the schema of Struct.

src/common/src/buffer/bitmap.rs

codecov · 2022-11-07T07:13:06Z

Codecov Report

Merging #6224 (85a833a) into main (7b57fbf) will increase coverage by 0.03%.
The diff coverage is 78.94%.

@@            Coverage Diff             @@
##             main    #6224      +/-   ##
==========================================
+ Coverage   74.56%   74.60%   +0.03%     
==========================================
  Files         945      947       +2     
  Lines      152769   153125     +356     
==========================================
+ Hits       113914   114238     +324     
- Misses      38855    38887      +32

Flag	Coverage Δ
rust	`74.60% <78.94%> (+0.03%)`	⬆️

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

Impacted Files	Coverage Δ
src/common/src/array/data_chunk_iter.rs	`83.89% <0.00%> (-0.86%)`	⬇️
src/frontend/src/handler/explain.rs	`52.42% <0.00%> (-2.68%)`	⬇️
src/meta/src/barrier/recovery.rs	`70.33% <ø> (-0.13%)`	⬇️
src/meta/src/hummock/compaction_group/mod.rs	`100.00% <ø> (ø)`
src/meta/src/hummock/compaction_scheduler.rs	`82.08% <ø> (ø)`
src/meta/src/hummock/manager/compaction.rs	`91.27% <0.00%> (+0.60%)`	⬆️
src/meta/src/rpc/service/hummock_service.rs	`3.20% <0.00%> (-0.44%)`	⬇️
src/meta/src/rpc/service/notification_service.rs	`15.15% <ø> (+0.15%)`	⬆️
src/stream/src/executor/simple.rs	`30.76% <ø> (ø)`
src/stream/src/from_proto/mod.rs	`0.00% <ø> (ø)`
... and 31 more

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

src/stream/src/executor/watermark_filter.rs

BugenZhao · 2022-11-07T07:42:31Z

src/stream/src/executor/watermark_filter.rs

+    async fn get_global_max_watermark(
+        table: &StateTable<S>,
+        watermark_type: DataType,
+    ) -> StreamExecutorResult<Datum> {


Should we use ScalarImpl here, or what's the semantic for watermark being NULL? cc @st1page.

Agree, in the currently watermark filter design it should not be Null

But maybe do it in later PRs.

src/stream/src/executor/watermark_filter.rs

BugenZhao · 2022-11-07T07:56:10Z

src/stream/src/executor/watermark_filter.rs

+                        for vnode in vnodes.iter_pos() {
+                            let pk = Some(ScalarImpl::Int16(vnode as _));
+                            let row = Row::new(vec![pk, current_watermark.clone()]);
+                            table.insert(row);


We may need upsert here since we're unsure about the prev value after taking the maximum on fail-over & scaling.

For now, we may simply disable_sanity_check to make it run. Could you please also make the unit test case cover this?

BugenZhao · 2022-11-07T07:59:11Z

src/stream/src/executor/watermark_filter.rs

+            .enumerate()
+            .map(|(id, data_type)| ColumnDesc::unnamed(ColumnId::new(id as i32), data_type.clone()))
+            .collect_vec();
+        StateTable::new_without_distribution_partial(


It would be better if we use the from_catalog to create a state table here, as...

the distribution should be Hash in most cases,

with the vnode_col_idx_in_pk set.

I can't find any other executor (except SourceExecutor) using from_catalog in tests. NTFS!

BugenZhao · 2022-11-07T08:00:39Z

src/stream/src/from_proto/watermark_filter.rs

+            params
+                .vnode_bitmap
+                .expect("vnodes not set for watermark filter"),


Will it be possible that the watermark filter is a singleton? cc @st1page @TennyZhuang

src/stream/src/from_proto/mod.rs

src/stream/src/executor/watermark_filter.rs

src/common/src/types/interval.rs

src/stream/src/executor/watermark_filter.rs

BugenZhao

Rest LGTM. Great work!!

BugenZhao · 2022-11-08T11:39:23Z

src/common/src/buffer/bitmap.rs

@@ -276,6 +276,12 @@ impl Bitmap {
        }
    }

+    pub fn iter_pos(&self) -> impl Iterator<Item = usize> + '_ {


We may reuse this in other places as well.

yuhao-su added 7 commits October 27, 2022 15:53

init

83c7b2d

Merge branch 'main' of github.com:singularity-data/risingwave into in…

4d57d7a

…troduce_watermark_filter

modify filter

054fd1f

add watermark table

78d4037

Merge branch 'main' of github.com:singularity-data/risingwave into in…

05489fb

…troduce_watermark_filter

pass test

6641eb4

resolve conflict

47fa07a

yuhao-su requested review from TennyZhuang, fuyufjh, BugenZhao and st1page November 7, 2022 06:36

github-actions bot added the type/feature Type: New feature. label Nov 7, 2022

yuhao-su added 2 commits November 7, 2022 14:38

fix typo

9b3f0f5

proto

c3785b8

yuhao-su commented Nov 7, 2022

View reviewed changes

yuhao-su requested review from xx01cyx and soundOfDestiny November 7, 2022 06:58

soundOfDestiny reviewed Nov 7, 2022

View reviewed changes

src/common/src/buffer/bitmap.rs Outdated Show resolved Hide resolved

soundOfDestiny reviewed Nov 7, 2022

View reviewed changes

src/stream/src/executor/watermark_filter.rs Show resolved Hide resolved

BugenZhao reviewed Nov 7, 2022

View reviewed changes

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

BugenZhao reviewed Nov 7, 2022

View reviewed changes

src/stream/src/executor/watermark_filter.rs Show resolved Hide resolved

BugenZhao reviewed Nov 7, 2022

View reviewed changes

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

BugenZhao reviewed Nov 7, 2022

View reviewed changes

src/stream/src/executor/watermark_filter.rs Show resolved Hide resolved

BugenZhao reviewed Nov 7, 2022

View reviewed changes

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

BugenZhao reviewed Nov 7, 2022

View reviewed changes

xx01cyx reviewed Nov 7, 2022

View reviewed changes

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

src/common/src/types/interval.rs Show resolved Hide resolved

resolve comments

9871961

BugenZhao reviewed Nov 8, 2022

View reviewed changes

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

BugenZhao reviewed Nov 8, 2022

View reviewed changes

more refinement

2308b00

BugenZhao approved these changes Nov 8, 2022

View reviewed changes

yuhao-su added the mergify/can-merge label Nov 8, 2022

Merge branch 'main' into introduce_watermark_filter

85a833a

TennyZhuang approved these changes Nov 8, 2022

View reviewed changes

BugenZhao reviewed Nov 8, 2022

View reviewed changes

mergify bot merged commit b0dfdb8 into main Nov 8, 2022

mergify bot deleted the introduce_watermark_filter branch November 8, 2022 11:56

feat(streaming): introduce watermark filter #6224

feat(streaming): introduce watermark filter #6224

Uh oh!

Conversation

yuhao-su commented Nov 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What's changed and what's your intention?

Checklist

Refer to a related PR or issue link (optional)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

codecov bot commented Nov 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

BugenZhao Nov 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

BugenZhao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

yuhao-su commented Nov 7, 2022 •

edited

Loading

codecov bot commented Nov 7, 2022 •

edited

Loading

BugenZhao Nov 8, 2022 •

edited

Loading