refactor(state-store): Different actors should use the same table id

[BowenXiao1999]

What's changed and what's your intention?

Should resolve #2028 and #2032 . After this PR, different actors of same fragment should fix back to use same table id. And it do not introduce new round of StreamNode traverse.

• Add a HashMap table_ids_map in fragmenter. (operator_id -> local table ids). It records the local table ids of each stateful operator. For cases that generate new operator

  • Exchange -> Dispatcher/Merge. They do not need table ids so it's safe.
  • Delta Join rewrite. Might generate Arrange Executors, so if Delta Join needs table states it should update this map.

• When serialize StreamNode, get local table id from table_ids_map via operator id. Add the start_table_id and we get global table id.

There will be a two-phase rewrite. First rewrite only write local table id into StreamNode (Build Fragment). Second rewrite will write local table id + offset (global table id) into StreamNode (Build actors).

Note that I do not use a enum to wrap TableId (Like LocalFragmentId, LocalActorId) cuz i think it's a bit over-design.

Welcome for suggestions.

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)

[codecov]

Codecov Report

Merging #2057 (4a7502a) into main (400dca9) will decrease coverage by 0.02%.
The diff coverage is 48.57%.

@@            Coverage Diff             @@
##             main    #2057      +/-   ##
==========================================
- Coverage   70.66%   70.64%   -0.03%     
==========================================
  Files         636      636              
  Lines       80718    80705      -13     
==========================================
- Hits        57037    57011      -26     
- Misses      23681    23694      +13     

Flag	Coverage Δ
rust	70.64% <48.57%> (-0.03%)	⬇️

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

Impacted Files	Coverage Δ
src/meta/src/stream/fragmenter/mod.rs	80.57% <44.00%> (-3.39%)	⬇️
...c/meta/src/stream/fragmenter/graph/stream_graph.rs	62.56% <55.55%> (-1.49%)	⬇️
src/meta/src/stream/stream_manager.rs	63.63% <100.00%> (ø)

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

[skyzh]

LGTM. Better to have a table_id also in lookup executor and arrange executor.