feat(flow): shared in-memory state for dataflow operator

[discord9]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

#3187

What's changed and what's your intention?

A shared state of key-value pair for various state
in dataflow execution, called Arrangement

i.e: Mfp operator with temporal filter need to store it's future output so that it can add now, and delete later. To get all needed updates in a time span, use get_updates_in_range

And reduce operator need full state of it's output, so that it can query(and modify by calling apply_updates)
existing state, also need a way to expire keys. To get a key's current value, use get with time being now
so it's like:
mfp operator -> arrange(store futures only, no expire) -> reduce operator <-> arrange(full, with key expiring time) -> output

Note the two way arrow between reduce operator and arrange, it's because reduce operator need to query existing state
and also need to update existing state.

This state will be used in building&execution dataflow graph

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR does not require documentation updates.

[codecov]

Codecov Report

Attention: Patch coverage is 95.29703% with 19 lines in your changes are missing coverage. Please review.

Project coverage is 85.00%. Comparing base (b85d7bb) to head (256a510).
Report is 13 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3508      +/-   ##
==========================================
- Coverage   85.30%   85.00%   -0.31%     
==========================================
  Files         904      907       +3     
  Lines      150170   151406    +1236     
==========================================
+ Hits       128106   128704     +598     
- Misses      22064    22702     +638     

[zhongzc]

LGTM

[waynexia]

LGTM