feat(optimizer): introduce watermark group

[stdrc]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Background

Previously we don't record the derivation relation among different watermark columns. For instance, window_start and window_end generated by tumble/hop time window functions are always related to each other and the original event time column, however when we plan for an EOWC query, we lost this information and hence couldn't succeed if more than 1 of these exist (not pruned) at the same time.

Changes

This PR introduces a new way to represent watermark columns in plan nodes -- watermark groups. In this PR we organize watermark columns into several groups, in each of which, all watermark columns are related. With this, when planning EOWC queries, we can handle multiple watermark columns as long as they are in the same one group. For example in the following query, we won't warn our user that there're more than one watermark column any more.

create materialized view mv as
select
  foo, ts, window_start, window_end
from tumble(t, ts, interval '5 mins')
emit on window close;

Among the changes, ~400 LoC are real, others are planner tests. Basically the core changes all happen in StreamXxx::new, with watermark derivation logic unchanged. The main contribution is to determine which columns among all output watermark columns are related for every stream node.

Future work

1. We still use FixedBitSet to record a flattened watermark_columns field in TableCatalog and TableDesc, and hence in Table protobuf. As a result, When we create MV on MV, all the watermark columns in the base MV will be considered belonging to different watermark groups no matter whether they are actually related. I will change the table catalog field later to fix this.
2. We still can't do create mv as select window_start, window_end, count(*) as cnt from tumble(t, ts, ...) group by window_start, window_end emit on window close, because I don't want to have too many changes in this single PR. Will modify StreamHashAgg later to support it. (Resolved in feat(eowc): allow multiple watermark columns in eowc hash agg #19998)
3. Since we don't support duplicated plan node reusing in streaming plan, when we have two side of a join both scanning the same table with join condition left.window = right.window, we cannot know that the two window columns are derived from the same source watermark column. This is because the watermark group ID is allocated and assigned during optimization phase and not persisted.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

[stdrc]

• feat(eowc): allow multiple watermark columns in eowc hash agg #19998
• feat(optimizer): introduce watermark group #19894 👈 (View in Graphite)
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[chenzl25]

Since we don't support shared node in streaming plan (IIRC),

Do you want this self-join plan with StreamShare?

risingwave/src/frontend/planner_test/tests/testdata/output/share.yaml

Lines 19 to 55 in c7a4a39

	- id: self_join
	before:
	- create_sources
	sql: |
	select count(*) cnt from auction A join auction B on A.id = B.id where A.initial_bid = 1 and B.initial_bid = 2;
	batch_plan: |-
	BatchSimpleAgg { aggs: [sum0(count)] }
	└─BatchExchange { order: [], dist: Single }
	└─BatchSimpleAgg { aggs: [count] }
	└─BatchHashJoin { type: Inner, predicate: id = id, output: [] }
	├─BatchExchange { order: [], dist: HashShard(id) }
	│ └─BatchFilter { predicate: (initial_bid = 1:Int32) }
	│ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
	└─BatchExchange { order: [], dist: HashShard(id) }
	└─BatchFilter { predicate: (initial_bid = 2:Int32) }
	└─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
	stream_plan: |-
	StreamMaterialize { columns: [cnt], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
	└─StreamProject { exprs: [sum0(count)] }
	└─StreamSimpleAgg [append_only] { aggs: [sum0(count), count] }
	└─StreamExchange { dist: Single }
	└─StreamStatelessSimpleAgg { aggs: [count] }
	└─StreamHashJoin [append_only] { type: Inner, predicate: id = id, output: [_row_id, id, _row_id] }
	├─StreamExchange { dist: HashShard(id) }
	│ └─StreamFilter { predicate: (initial_bid = 1:Int32) }
	│ └─StreamShare { id: 4 }
	│ └─StreamProject { exprs: [id, initial_bid, _row_id] }
	│ └─StreamFilter { predicate: ((initial_bid = 1:Int32) OR (initial_bid = 2:Int32)) }
	│ └─StreamRowIdGen { row_id_index: 10 }
	│ └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
	└─StreamExchange { dist: HashShard(id) }
	└─StreamFilter { predicate: (initial_bid = 2:Int32) }
	└─StreamShare { id: 4 }
	└─StreamProject { exprs: [id, initial_bid, _row_id] }
	└─StreamFilter { predicate: ((initial_bid = 1:Int32) OR (initial_bid = 2:Int32)) }
	└─StreamRowIdGen { row_id_index: 10 }
	└─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }

[github-actions]

license-eye has checked 5536 files.

Valid	Invalid	Ignored	Fixed
2332	2	3202	0

Click to see the invalid file list

• src/common/src/util/functional.rs
• src/frontend/src/optimizer/property/watermark_columns.rs

Use this command to fix any missing license headers

```bash

docker run -it --rm -v $(pwd):/github/workspace apache/skywalking-eyes header fix

</details>

[stdrc]

@chenzl25 In this plan, is it possible to use a shared StreamTableScan on the two sides?

[chenzl25]

You can use a cte to construct a share operator.

create materialized view v as  
with cte as (select foo, window_start as win from tumble(t, ts, interval '5 mins'))
select l.foo + r.foo as foo, l.win as win1, r.win as win2
from cte as l join cte as r on l.win = r.win;

[stdrc]

Thanks!