1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
use crate::{RocksEngine, UserProperties};
use engine_traits::{
DecodeProperties, MvccProperties, MvccPropertiesExt, Result, TableProperties,
TablePropertiesCollection, TablePropertiesExt,
};
use txn_types::TimeStamp;
pub(crate) const PROP_NUM_ERRORS: &str = "tikv.num_errors";
pub(crate) const PROP_MIN_TS: &str = "tikv.min_ts";
pub(crate) const PROP_MAX_TS: &str = "tikv.max_ts";
pub(crate) const PROP_NUM_ROWS: &str = "tikv.num_rows";
pub(crate) const PROP_NUM_PUTS: &str = "tikv.num_puts";
pub(crate) const PROP_NUM_DELETES: &str = "tikv.num_deletes";
pub(crate) const PROP_NUM_VERSIONS: &str = "tikv.num_versions";
pub(crate) const PROP_MAX_ROW_VERSIONS: &str = "tikv.max_row_versions";
pub(crate) const PROP_ROWS_INDEX: &str = "tikv.rows_index";
pub(crate) const PROP_ROWS_INDEX_DISTANCE: u64 = 10000;
pub struct RocksMvccProperties;
impl RocksMvccProperties {
pub fn encode(mvcc_props: &MvccProperties) -> UserProperties {
let mut props = UserProperties::new();
props.encode_u64(PROP_MIN_TS, mvcc_props.min_ts.into_inner());
props.encode_u64(PROP_MAX_TS, mvcc_props.max_ts.into_inner());
props.encode_u64(PROP_NUM_ROWS, mvcc_props.num_rows);
props.encode_u64(PROP_NUM_PUTS, mvcc_props.num_puts);
props.encode_u64(PROP_NUM_DELETES, mvcc_props.num_deletes);
props.encode_u64(PROP_NUM_VERSIONS, mvcc_props.num_versions);
props.encode_u64(PROP_MAX_ROW_VERSIONS, mvcc_props.max_row_versions);
props
}
pub fn decode<T: DecodeProperties>(props: &T) -> Result<MvccProperties> {
let mut res = MvccProperties::new();
res.min_ts = props.decode_u64(PROP_MIN_TS)?.into();
res.max_ts = props.decode_u64(PROP_MAX_TS)?.into();
res.num_rows = props.decode_u64(PROP_NUM_ROWS)?;
res.num_puts = props.decode_u64(PROP_NUM_PUTS)?;
res.num_versions = props.decode_u64(PROP_NUM_VERSIONS)?;
res.num_deletes = props
.decode_u64(PROP_NUM_DELETES)
.unwrap_or_else(|_| res.num_versions - res.num_puts);
res.max_row_versions = props.decode_u64(PROP_MAX_ROW_VERSIONS)?;
Ok(res)
}
}
impl MvccPropertiesExt for RocksEngine {
fn get_mvcc_properties_cf(
&self,
cf: &str,
safe_point: TimeStamp,
start_key: &[u8],
end_key: &[u8],
) -> Option<MvccProperties> {
let collection = match self.get_range_properties_cf(cf, &start_key, &end_key) {
Ok(c) if !c.is_empty() => c,
_ => return None,
};
let mut props = MvccProperties::new();
for (_, v) in collection.iter() {
let mvcc = match RocksMvccProperties::decode(&v.user_collected_properties()) {
Ok(m) => m,
Err(_) => return None,
};
if mvcc.min_ts > safe_point {
continue;
}
props.add(&mvcc);
}
Some(props)
}
}