feat(log-store): support kv-based log store

[wenym1]

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

What's changed and what's your intention?

In this PR, we implement a kv log store based on the serde implemented from #9451 and #10090.

First, we implement a KvLogStoreBuffer that acts as a channel between the log writer and reader. The writer writes data to the buffer, and the reader reads from the buffer and await on the buffer when it is empty. The buffer has a capacity. When the capacity has not been reached, the writer will write the stream chunk into the buffer. When the capacity has been reached, the stream chunk will be written to state store, and only the start and end seq id is added to the buffer. When the reader gets initialized with a latest epoch, the reader will read all data flushed before such epoch, and then read the data from the buffer.

Implement risingwavelabs/rfcs#55

🤖 Generated by Copilot at 539e472

This pull request refactors the log store logic in the stream module and implements a new kv log store based on a key-value state store. It also changes the visibility of some serde types and functions to be crate-private. It moves the in-memory log store logic to a separate module in_mem.rs and adds new modules for the kv log store components: buffer.rs, reader.rs, writer.rs, and mod.rs. It updates the imports of the BoundedInMemLogStoreFactory in the sink.rs and from_proto/sink.rs modules to reflect the new location.

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

[hzxa21]

Should this be first_write_epoch (although the key_range can guarantee correctness)?

[wenym1]

We have a safe_epoch in hummock version. Reading below the safe epoch may return an error even though it is correct. Since we don't pin the epoch in log reader, the safe epoch will bump up to above the first write epoch, and then read from the first write epoch may return an error.

[hzxa21]

By safe epoch, do you mean the compaction watermark? I don't think this is an issue because all keys written to log store are unique.

[wenym1]

Same reason as above.

[hzxa21]

Does it mean that barrier will keep appending to the in-mem buffer even when sink is lagging behind?

[wenym1]

Yes.

In future PR, we can try merging the data of multiple epochs if the sink is lagging too behind.

[hzxa21]

Thanks for the PR! LGTM

[tabVersion]

also please attach the design doc in pr description