feat(storage): support replicated LocalHummockStorage

[kwannoel]

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

What's changed and what's your intention?

Part of #10225.

The temporary storage is needed to replicate upstream shared buffer changes, if the two are not on the same CN.

We add is_replicated flag to indicate if a local state store is replicated.

In the case that it is, we will only read its read version when reading via state table.

This is because we have to consider an edge case: when the executor using replicated storage is scheduled to the same CN as the upstream mview executor. If replication happens, we will have duplicate read versions in the shared buffer.

Because local state store read (which happens for state table reads) will ONLY read its own read version, it is fine for state table.

But storage table will read from all read versions (via global state store read), which is an issue since we will have 2 copies (original + replicated).

Hence in this PR we also amend global state store read to de duplicate by ignoring ALL read version replicas.

• Ensure that when we create ReadVersion, it should follow the local options.
• Ensure global state store filters out ReadVersion which are replicates.

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

• My PR DOES NOT contain user-facing changes.

Click here for Documentation

Types of user-facing changes

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

• 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

[codecov]

Codecov Report

Merging #10226 (69781d0) into main (8eb0e43) will increase coverage by 0.02%.
The diff coverage is 93.77%.

@@            Coverage Diff             @@
##             main   #10226      +/-   ##
==========================================
+ Coverage   70.52%   70.55%   +0.02%     
==========================================
  Files        1244     1244              
  Lines      213316   213517     +201     
==========================================
+ Hits       150435   150638     +203     
+ Misses      62881    62879       -2     

Flag	Coverage Δ
rust	70.55% <93.77%> (+0.02%)	⬆️

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

Impacted Files	Coverage Δ
src/storage/hummock_trace/src/opts.rs	23.43% <ø> (ø)
src/storage/src/hummock/event_handler/mod.rs	19.64% <0.00%> (-1.52%)	⬇️
src/storage/src/store.rs	67.85% <95.45%> (+6.74%)	⬆️
src/storage/hummock_test/src/state_store_tests.rs	84.05% <95.73%> (+1.43%)	⬆️
src/storage/hummock_trace/src/collector.rs	51.19% <100.00%> (+0.14%)	⬆️
...src/hummock/event_handler/hummock_event_handler.rs	73.02% <100.00%> (+0.22%)	⬆️
src/storage/src/hummock/mod.rs	85.60% <100.00%> (+0.05%)	⬆️
src/storage/src/hummock/state_store.rs	70.50% <100.00%> (+0.69%)	⬆️
src/stream/src/common/table/state_table.rs	88.84% <100.00%> (ø)

... and 2 files with indirect coverage changes

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

[wenym1]

Currently, if we read from state table (with underlying local state store), we only read the ReadVersion associated to it, and if we read from storage table (with underlying global state store), it will read all ReadVersions.

Before merging this PR, we should decide whether the read from the storage table can read the data from the replicated ReadVersions. If allowed, later we should carefully ensure that only the data of vnode from other CNs are replicated to the current CN. If not allowed, in this PR we should distinguish whether a ReadVersion is replicated or not, and only return the not replicated ReadVersions to handle global state store read.

[hzxa21]

Can we add test cases to ensure that 1) replicated ReadVersion can be seen by the corresponding LocalStateStore (for streaming read) and 2) cannot be seen by HummockStorage (for batch read)?

[kwannoel]

This filters out replicated read versions, when reading from global state store.

[kwannoel]

Can we add test cases to ensure that 1) replicated ReadVersion can be seen by the corresponding LocalStateStore (for streaming read) and 2) cannot be seen by HummockStorage (for batch read)?

Added them in 95f9337.

[Li0k]

LGTM

[Li0k]

@chenzl25 PTAL

[chenzl25]

LGTM!