feat(emit-on-window-close): enable queries with EMIT ON WINDOW CLOSE keyword

[stdrc]

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

What's changed and what's your intention?

This PR

• Allows LogicalXxx::to_stream to be aware of EMIT ON WINDOW CLOSE mode by adding a field to ToStreamContext. (Later in feat(over window): generate EOWC stream plan from LogicalOverAgg #9597 we will have our first operator that generates EOWC version of stream node for EOWC emit mode.)
• Adds emit_on_window_close field to PlanBase so that every plan node has to decide whether it's EOWC.
• Enforces EOWC property by adding a StreamSort above the generated stream plan root when user required EOWC mode.
• Supports EOWC mode in planner test.

Checklist For Contributors

• 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 Sqlsmith: Sql feature generation #7934).
• 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 to the issue)
• All checks passed in ./risedev check (or alias, ./risedev c)

Checklist For Reviewers

• I have requested macro/micro-benchmarks as this PR can affect performance substantially, and the results are shown.

Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• SQL commands, functions, and operators

Release note

Support EMIT ON WINDOW CLOSE mode for queries on sources with watermark column.

Example:

CREATE SOURCE t (a TIMESTAMP, b INT, WATERMARK FOR a AS a - INTERVAL '5 minutes') WITH ...;
CREATE MATERIALIZED VIEW mv EMIT ON WINDOW CLOSE AS
SELECT
    window_start, max(b)
FROM tumble(t, a, INTERVAL '1 hour')
GROUP BY window_start;

[soundOfDestiny]

LGTM

[codecov]

Codecov Report

Merging #9622 (866c6be) into main (2110af0) will increase coverage by 0.01%.
The diff coverage is 63.09%.

@@            Coverage Diff             @@
##             main    #9622      +/-   ##
==========================================
+ Coverage   70.80%   70.82%   +0.01%     
==========================================
  Files        1238     1238              
  Lines      207141   207367     +226     
==========================================
+ Hits       146676   146873     +197     
- Misses      60465    60494      +29     

Flag	Coverage Δ
rust	70.82% <63.09%> (+0.01%)	⬆️

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

Impacted Files	Coverage Δ
src/common/src/util/stream_graph_visitor.rs	54.07% <0.00%> (-1.99%)	⬇️
src/frontend/src/handler/explain.rs	69.23% <0.00%> (-3.45%)	⬇️
src/frontend/src/optimizer/plan_node/stream.rs	14.00% <0.00%> (-0.23%)	⬇️
.../frontend/src/optimizer/plan_node/stream_derive.rs	0.00% <0.00%> (ø)
...src/optimizer/plan_node/stream_eowc_over_window.rs	0.00% <0.00%> (ø)
.../frontend/src/optimizer/plan_node/stream_expand.rs	0.00% <0.00%> (ø)
src/frontend/src/optimizer/plan_node/plan_base.rs	94.76% <53.84%> (-3.35%)	⬇️
src/frontend/src/optimizer/plan_node/convert.rs	80.61% <75.67%> (-3.26%)	⬇️
src/frontend/planner_test/src/lib.rs	76.57% <91.52%> (+1.33%)	⬆️
src/frontend/src/handler/create_mv.rs	95.54% <100.00%> (+0.14%)	⬆️
... and 28 more

... and 8 files with indirect coverage changes

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

[st1page]

LGTM