feat(storage): support replicate batch (aka remote write)

[skyzh]

Signed-off-by: Alex Chi [email protected]

What's changed and what's your intention?

See https://singularity-data.quip.com/JXxMAhxzBKmD/Shared-State-Shared-Index-and-Efficient-Joins for more details. Now we will support "replicate batch" in Hummock, so that executors can write a batch without uploading them to S3. This interface is replicate_batch in StateStore.

We also refactor a lot for Hummock shared buffer so that we don't need to clone keys. Previously,

    async fn ingest_batch(&self, kv_pairs: Vec<(Bytes, Option<Bytes>)>, epoch: u64) -> Result<()> {
         self.storage
             .write_batch(
                 kv_pairs
                     .into_iter()
                     .map(|(k, v)| (k.to_vec(), v.map(|x| x.to_vec()).into())),
                 epoch,
             )
             .await?;

         Ok(())
     }

Every write batch requires a copy. Now we will use Bytes across our system.

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 #718 (525bdb3) into main (33d427e) will decrease coverage by 0.04%.
The diff coverage is 46.51%.

@@             Coverage Diff              @@
##               main     #718      +/-   ##
============================================
- Coverage     72.33%   72.28%   -0.05%     
  Complexity     2758     2758              
============================================
  Files           907      907              
  Lines         52466    52529      +63     
  Branches       1783     1783              
============================================
+ Hits          37949    37971      +22     
- Misses        13625    13666      +41     
  Partials        892      892              

Flag	Coverage Δ
java	61.16% <ø> (ø)
rust	77.04% <46.51%> (-0.08%)	⬇️

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

Impacted Files	Coverage Δ
rust/storage/src/hummock/state_store.rs	0.00% <0.00%> (ø)
rust/storage/src/hummock/state_store_tests.rs	81.81% <ø> (ø)
rust/storage/src/monitor/monitored_store.rs	1.47% <0.00%> (-0.03%)	⬇️
rust/storage/src/panic_store.rs	0.00% <0.00%> (ø)
rust/storage/src/rocksdb_local.rs	0.00% <ø> (ø)
rust/storage/src/rocksdb_local_mock.rs	0.00% <0.00%> (ø)
rust/storage/src/store.rs	28.57% <ø> (-0.85%)	⬇️
rust/storage/src/tikv.rs	0.00% <ø> (ø)
rust/storage/src/tikv_mock.rs	0.00% <0.00%> (ø)
rust/storage/src/write_batch.rs	81.13% <0.00%> (-6.63%)	⬇️
... and 6 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 33d427e...525bdb3. Read the comment docs.

[BugenZhao]

LSTM!

[skyzh]

hold until discussion tomorrow. btw, this might conflict with #709