feat(storage): refactor and support local state store flushed snapshot reader

[wenym1]

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

What's changed and what's your intention?

Previously in the sink log store, for log reader, we use methods in StateStore to access the data flushed by log writer. Methods in StateStore will read all shards (aka HummockReadVersion) of the table_id on the same CN. However, since the log reader is always in the same SinkExecutor as log writer, it actually only needs to read the shard that belongs to the writer, and using StateStore will unnecessarily read more unrelated data.

In this PR, we will add a new method new_flushed_snapshot_reader to LocalStateStore, whose return type is a new associated type FlushedSnapshotReader that should implement the StateStoreRead trait.

pub trait LocalStateStore: StaticSendSync {
    type FlushedSnapshotReader: StateStoreRead + Clone;
   ...

    fn new_flushed_snapshot_reader(&self) -> Self::FlushedSnapshotReader;
}

The semantic of this new FlushedSnapshotReader is, it can read the uncommitted data written by the parent LocalStateStore as long as the data has been flusheed by the parent LocalStateStore. Such FlushedSnapshotReader will only read the same shard as the local state store.

The KvLogStoreReader will change to use this FlushedSnapshotReader to read to avoid reading unnecessary data in other shards in the CN.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

[kwannoel]

LGTM

[wenym1]

• refactor(storage): unify state store trait get method #20175
• refactor(storage): create new read snapshot to read for StateStore trait #20172
• feat(storage): refactor and support local state store flushed snapshot reader #20153 👈 (View in Graphite)
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.