sst_importer: add delete type key for sst importer writer
@@ -12,6 +12,10 @@ use std::time::{Duration, Instant};
use futures_util::io::{AsyncRead, AsyncReadExt};
use kvproto::backup::StorageBackend;
+#[cfg(feature = "prost-codec")]
+use kvproto::import_sstpb::pair::Op as PairOp;
+#[cfg(not(feature = "prost-codec"))]
+use kvproto::import_sstpb::PairOp;
use kvproto::import_sstpb::*;
use tokio::time::timeout;
use uuid::{Builder as UuidBuilder, Uuid};
@@ -472,25 +476,25 @@ impl<E: KvEngine> SSTWriter<E> {
let commit_ts = TimeStamp::new(batch.get_commit_ts());
for m in batch.get_pairs().iter() {
let k = Key::from_raw(m.get_key()).append_ts(commit_ts);
- self.put(k.as_encoded(), m.get_value())?;
+ self.put(k.as_encoded(), m.get_value(), m.get_op())?;
}
Ok(())
}
- fn put(&mut self, key: &[u8], value: &[u8]) -> Result<()> {
+ fn put(&mut self, key: &[u8], value: &[u8], op: PairOp) -> Result<()> {
let k = keys::data_key(key);
let (_, commit_ts) = Key::split_on_ts_for(key)?;
- if is_short_value(value) {
- let w = KvWrite::new(WriteType::Put, commit_ts, Some(value.to_vec()));
- self.write.put(&k, &w.as_ref().to_bytes())?;
- self.write_entries += 1;
- } else {
- let w = KvWrite::new(WriteType::Put, commit_ts, None);
- self.write.put(&k, &w.as_ref().to_bytes())?;
- self.write_entries += 1;
+ let w = match (op, is_short_value(value)) {
+ (PairOp::Delete, _) => KvWrite::new(WriteType::Delete, commit_ts, None),
+ (PairOp::Put, true) => KvWrite::new(WriteType::Put, commit_ts, Some(value.to_vec())),
+ (PairOp::Put, false) => {
self.default.put(&k, value)?;
self.default_entries += 1;
+ KvWrite::new(WriteType::Put, commit_ts, None)
}
+ };
+ self.write.put(&k, &w.as_ref().to_bytes())?;
+ self.write_entries += 1;
Ok(())
}
@@ -1639,23 +1643,32 @@ mod tests {
let mut batch = WriteBatch::default();
let mut pairs = vec![];
- // wirte cf
+ // put short value kv in wirte cf
let mut pair = Pair::default();
pair.set_key(b"k1".to_vec());
pair.set_value(b"short_value".to_vec());
pairs.push(pair);
- // default cf
+ // put big value kv in default cf
let big_value = vec![42; 256];
let mut pair = Pair::default();
pair.set_key(b"k2".to_vec());
pair.set_value(big_value);
pairs.push(pair);
+ // put delete type key in write cf
+ let mut pair = Pair::default();
+ pair.set_key(b"k3".to_vec());
+ pair.set_op(PairOp::Delete);
+ pairs.push(pair);
+
// generate two cf metas
batch.set_commit_ts(10);
batch.set_pairs(pairs.into());
w.write(batch).unwrap();
+ assert_eq!(w.write_entries, 3);
+ assert_eq!(w.default_entries, 1);
+
let metas = w.finish().unwrap();
assert_eq!(metas.len(), 2);
}
