feat(meta): split stateful operators into different fragments

[skyzh]

Signed-off-by: Alex Chi iskyzh@gmail.com

What's changed and what's your intention?

This PR adds a rewriter on meta service to split stateful operators into different fragments.

before:

now:

changes:

• all stream plan nodes now have a schema. otherwise meta node cannot insert exchange node with the correct exchange fields.
• remove a test suite. e2e is already good enough to test things out, no need to manually construct all plan nodes.
• remove fields (schema) from exchange node.
• Rust frontend will now automatically add a schema for all plan nodes. Java frontend won't, but as it will be deprecated soon, we won't refactor it.
• a new exchange type called no_shuffle, that is used between such splits. No-shuffle exchange will add 1-to-1 channel between upstream actors and downstream actors within a fragment.

Checklist

• I have written necessary docs and comments
• I have added necessary unit tests and integration tests

Refer to a related PR or issue link (optional)

close #1745

[skyzh]

One might ask why I don't do this on frontend. That's because the fragmenter is on meta, so such rewrites can only be efficiently done on meta. Also, for constructing delta join plans, such rewrites are also useful.

[codecov]

Codecov Report

Merging #1817 (da40f37) into main (50f51fd) will decrease coverage by 0.72%.
The diff coverage is 65.03%.

@@            Coverage Diff             @@
##             main    #1817      +/-   ##
==========================================
- Coverage   71.40%   70.67%   -0.73%     
==========================================
  Files         608      607       -1     
  Lines       79403    79263     -140     
==========================================
- Hits        56695    56017     -678     
- Misses      22708    23246     +538     

Flag	Coverage Δ
rust	70.67% <65.03%> (-0.73%)	⬇️

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

Impacted Files	Coverage Δ
...rontend/src/optimizer/plan_node/stream_exchange.rs	77.14% <ø> (-0.64%)	⬇️
src/stream/src/task/mod.rs	44.87% <ø> (-11.54%)	⬇️
src/stream/src/task/stream_manager.rs	0.00% <ø> (-52.46%)	⬇️
src/meta/src/stream/fragmenter.rs	90.28% <56.09%> (-5.92%)	⬇️
src/meta/src/stream/graph/stream_graph.rs	67.18% <63.63%> (-3.13%)	⬇️
src/frontend/src/optimizer/plan_node/mod.rs	98.00% <100.00%> (+0.02%)	⬆️
...ntend/src/optimizer/plan_node/stream_table_scan.rs	94.91% <100.00%> (+0.08%)	⬆️
src/meta/src/stream/test_fragmenter.rs	99.61% <100.00%> (+<0.01%)	⬆️
src/stream/src/executor/merge.rs	0.00% <0.00%> (-100.00%)	⬇️
src/stream/src/executor/project.rs	0.00% <0.00%> (-100.00%)	⬇️
... and 15 more

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[MrCroxx]

Can we simply take its ownership in generate_graph? Anyway, it need to be cloned. 🤔

[skyzh]

Yes. It won't need. After this step, the plan is now fragments in fragment graph.

[yezizp2012]

Can rewrite_stream_node_inner be done in frontend? It might be more reasonable. 🤔

[skyzh]

Then we will have to move the whole fragmenter to frontend, or have a proto to pass fragment among nodes 🤣 Currently, fragments are only intermediate representation that are generated on meta node.