feat(streaming): handle watermark for mv on mv

[chenzl25]

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

What's changed and what's your intention?

• Ignore watermark during snapshot consumption and forward watermark messages to downstream after that.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features).
• I have demonstrated that backward compatibility is not broken by breaking changes and created issues to track deprecated features to be removed in the future. (Please refer the issue)
• 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)

[codecov]

Codecov Report

Merging #7790 (400de73) into main (8fb2d3f) will decrease coverage by 0.01%.
The diff coverage is 2.63%.

@@            Coverage Diff             @@
##             main    #7790      +/-   ##
==========================================
- Coverage   71.68%   71.67%   -0.01%     
==========================================
  Files        1101     1101              
  Lines      175776   175800      +24     
==========================================
  Hits       126007   126007              
- Misses      49769    49793      +24     

Flag	Coverage Δ
rust	71.67% <2.63%> (-0.01%)	⬇️

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

Impacted Files	Coverage Δ
src/stream/src/executor/backfill.rs	0.00% <0.00%> (ø)
src/stream/src/executor/rearranged_chain.rs	0.00% <0.00%> (ø)
src/stream/src/executor/chain.rs	85.71% <14.28%> (-5.20%)	⬇️
src/batch/src/task/task_execution.rs	53.38% <0.00%> (-0.51%)	⬇️
src/common/src/field_generator/mod.rs	74.50% <0.00%> (-0.50%)	⬇️
src/stream/src/executor/aggregation/minput.rs	96.34% <0.00%> (+0.10%)	⬆️
src/common/src/cache.rs	97.26% <0.00%> (+0.10%)	⬆️
src/object_store/src/object/mod.rs	51.73% <0.00%> (+0.21%)	⬆️

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

[st1page]

A random thought, if a watermark column exists as the first column in the pk, can we emit the watermark when backfill? Even maybe we can do some optimization for backfill with that watermark?

[chenzl25]

Good catch. I think if the first column of the pk is ascending, we can emit watermark once it passes the first column of current_pos of the backfill progress, otherwise we can emit the first column of current_pos itself.