use linkrs_core::types::{EdgeId, Timestamp};
use linkrs_core::{StorageError, StorageResult, Value};
use std::collections::HashMap;
use std::sync::Arc;
use crate::column_stats::{compute_stats_streaming, deserialize_stat_value, serialize_stat_value};
use crate::encoding::EncodingType;
#[derive(Debug, Clone)]
pub struct TombstoneStats {
pub count: usize,
pub memory_bytes: usize,
pub oldest_delete_ts: Option<Timestamp>,
pub newest_delete_ts: Option<Timestamp>,
}
impl TombstoneStats {
pub fn estimate_memory(count: usize) -> usize {
const HASH_NODE_OVERHEAD: usize = 48;
const AUTHORITY_RECORD: usize =
std::mem::size_of::<EdgeId>() + 2 * std::mem::size_of::<Timestamp>();
count * (AUTHORITY_RECORD + HASH_NODE_OVERHEAD)
}
}
#[derive(Debug, Clone, Default)]
pub struct DeletionStats {
pub total_deleted_edges: u64,
pub total_live_edges: u64,
}
impl DeletionStats {
pub fn deletion_ratio(&self) -> f64 {
let total = self
.total_live_edges
.saturating_add(self.total_deleted_edges);
if total == 0 {
0.0
} else {
self.total_deleted_edges as f64 / total as f64
}
}
pub fn deletion_percentage(&self) -> f64 {
self.deletion_ratio() * 100.0
}
pub fn is_significant(&self) -> bool {
self.deletion_ratio() > 0.1
}
}
#[derive(Debug, Clone)]
pub struct GroupSegmentStats {
pub group: u32,
pub row_count: u64,
pub live_count: u64,
pub null_count: u64,
pub sort_min: Option<Value>,
pub sort_max: Option<Value>,
pub columns: HashMap<Arc<str>, crate::column_stats::ColumnStats>,
}
impl GroupSegmentStats {
pub fn collect(
group: u32,
row_count: u64,
live_count: u64,
endpoints: &[u32],
column_values: &HashMap<Arc<str>, Vec<Option<Value>>>,
column_encodings: &HashMap<Arc<str>, EncodingType>,
) -> Self {
let endpoint_values = endpoints
.iter()
.map(|endpoint| Some(Value::BigInt(*endpoint as i64)));
let endpoint_stats = compute_stats_streaming(endpoint_values, EncodingType::None, 0, 0);
let (sort_min, sort_max) = (
endpoint_stats.min_value.clone(),
endpoint_stats.max_value.clone(),
);
let mut null_count = 0u64;
let mut columns = HashMap::with_capacity(column_values.len());
for (name, values) in column_values {
let encoding = column_encodings
.get(name)
.copied()
.unwrap_or(EncodingType::None);
let stats = compute_stats_streaming(values.iter().cloned(), encoding, 0, 0);
null_count = null_count.saturating_add(stats.null_count);
columns.insert(name.clone(), stats);
}
Self {
group,
row_count,
live_count,
null_count,
sort_min,
sort_max,
columns,
}
}
pub fn widen_with(&mut self, fresh: &GroupSegmentStats) {
self.row_count = fresh.row_count;
self.live_count = fresh.live_count;
self.null_count = fresh.null_count;
match (&self.sort_min, &fresh.sort_min) {
(_, None) => {}
(None, Some(_)) => self.sort_min = fresh.sort_min.clone(),
(Some(current), Some(next)) => {
if crate::cursor::compare_stat_values(next, current) == std::cmp::Ordering::Less {
self.sort_min = fresh.sort_min.clone();
}
}
}
match (&self.sort_max, &fresh.sort_max) {
(_, None) => {}
(None, Some(_)) => self.sort_max = fresh.sort_max.clone(),
(Some(current), Some(next)) => {
if crate::cursor::compare_stat_values(next, current) == std::cmp::Ordering::Greater
{
self.sort_max = fresh.sort_max.clone();
}
}
}
for (name, fresh_stats) in &fresh.columns {
match self.columns.get_mut(name) {
None => {
self.columns.insert(name.clone(), fresh_stats.clone());
}
Some(current) => {
match (¤t.min_value, &fresh_stats.min_value) {
(_, None) => {}
(None, Some(_)) => current.min_value = fresh_stats.min_value.clone(),
(Some(a), Some(b)) => {
if crate::cursor::compare_stat_values(b, a) == std::cmp::Ordering::Less
{
current.min_value = fresh_stats.min_value.clone();
}
}
}
match (¤t.max_value, &fresh_stats.max_value) {
(_, None) => {}
(None, Some(_)) => current.max_value = fresh_stats.max_value.clone(),
(Some(a), Some(b)) => {
if crate::cursor::compare_stat_values(b, a)
== std::cmp::Ordering::Greater
{
current.max_value = fresh_stats.max_value.clone();
}
}
}
current.null_count = fresh_stats.null_count;
match (¤t.hll, &fresh_stats.hll) {
(Some(_), Some(fresh_hll)) => {
if let Some(merged) = current.hll.clone() {
let mut combined = merged;
combined.merge(fresh_hll);
current.distinct_count = Some(combined.estimate());
current.hll = Some(combined);
} else {
current.hll = fresh_stats.hll.clone();
current.distinct_count = fresh_stats.distinct_count;
}
}
(None, Some(_)) => {
current.hll = fresh_stats.hll.clone();
current.distinct_count = fresh_stats.distinct_count;
}
_ => {
current.distinct_count = fresh_stats.distinct_count;
}
}
current.encoding_type = fresh_stats.encoding_type;
}
}
}
}
pub fn may_contain(&self, predicates: &[crate::cursor::ScanPredicate]) -> bool {
if predicates.is_empty() {
return true;
}
for range in crate::cursor::ScanPredicate::merged_ranges(predicates) {
let Some(stats) = self.columns.get(&*range.column) else {
continue;
};
let (Some(min), Some(max)) = (stats.min_value.as_ref(), stats.max_value.as_ref())
else {
continue;
};
if !range.overlaps(min, max) {
return false;
}
}
true
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::new();
out.extend_from_slice(&self.group.to_le_bytes());
out.extend_from_slice(&self.row_count.to_le_bytes());
out.extend_from_slice(&self.live_count.to_le_bytes());
out.extend_from_slice(&self.null_count.to_le_bytes());
out.push(self.sort_min.is_some() as u8);
if let Some(ref value) = self.sort_min {
let mut buf = Vec::new();
if serialize_stat_value(&mut buf, value).is_ok() {
out.extend_from_slice(&(buf.len() as u32).to_le_bytes());
out.extend_from_slice(&buf);
} else {
out.extend_from_slice(&0u32.to_le_bytes());
}
}
out.push(self.sort_max.is_some() as u8);
if let Some(ref value) = self.sort_max {
let mut buf = Vec::new();
if serialize_stat_value(&mut buf, value).is_ok() {
out.extend_from_slice(&(buf.len() as u32).to_le_bytes());
out.extend_from_slice(&buf);
} else {
out.extend_from_slice(&0u32.to_le_bytes());
}
}
let mut names: Vec<&Arc<str>> = self.columns.keys().collect();
names.sort();
out.extend_from_slice(&(names.len() as u32).to_le_bytes());
for name in names {
let bytes = name.as_bytes();
out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
out.extend_from_slice(bytes);
let mut stats_buf = Vec::new();
if self.columns[name].serialize_meta(&mut stats_buf).is_ok() {
out.extend_from_slice(&(stats_buf.len() as u32).to_le_bytes());
out.extend_from_slice(&stats_buf);
} else {
out.extend_from_slice(&0u32.to_le_bytes());
}
}
out
}
pub fn decode(data: &[u8], cursor: &mut usize) -> StorageResult<Self> {
let take = |data: &[u8], cursor: &mut usize, len: usize| -> StorageResult<Vec<u8>> {
if data.len().saturating_sub(*cursor) < len {
return Err(StorageError::deserialize_error(
"segment stats record too short",
));
}
let slice = data[*cursor..*cursor + len].to_vec();
*cursor += len;
Ok(slice)
};
let group = u32::from_le_bytes(
take(data, cursor, 4)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment group too short"))?,
);
let row_count = u64::from_le_bytes(
take(data, cursor, 8)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment row count too short"))?,
);
let live_count = u64::from_le_bytes(
take(data, cursor, 8)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment live count too short"))?,
);
let null_count = u64::from_le_bytes(
take(data, cursor, 8)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment null count too short"))?,
);
let has_min = take(data, cursor, 1)?[0] != 0;
let sort_min = if has_min {
let len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
StorageError::deserialize_error("segment sort min length too short")
})?) as usize;
let bytes = take(data, cursor, len)?;
let mut slice = &bytes[..];
Some(deserialize_stat_value(&mut slice)?)
} else {
None
};
let has_max = take(data, cursor, 1)?[0] != 0;
let sort_max = if has_max {
let len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
StorageError::deserialize_error("segment sort max length too short")
})?) as usize;
let bytes = take(data, cursor, len)?;
let mut slice = &bytes[..];
Some(deserialize_stat_value(&mut slice)?)
} else {
None
};
let column_count = u32::from_le_bytes(
take(data, cursor, 4)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment column count too short"))?,
) as usize;
let mut columns = HashMap::with_capacity(column_count);
for _ in 0..column_count {
let name_len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
StorageError::deserialize_error("segment column name length too short")
})?) as usize;
let name_bytes = take(data, cursor, name_len)?;
let name = String::from_utf8(name_bytes)
.map_err(|e| StorageError::deserialize_error(e.to_string()))?;
let stats_len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
StorageError::deserialize_error("segment column stats length too short")
})?) as usize;
let stats_bytes = take(data, cursor, stats_len)?;
let mut slice = &stats_bytes[..];
let stats = crate::column_stats::ColumnStats::deserialize_meta(&mut slice)?;
if !slice.is_empty() {
return Err(StorageError::deserialize_error(
"unexpected trailing data in segment column stats",
));
}
columns.insert(Arc::from(name.as_str()), stats);
}
Ok(Self {
group,
row_count,
live_count,
null_count,
sort_min,
sort_max,
columns,
})
}
}
pub fn encode_segment_snapshot(stats: &HashMap<u32, GroupSegmentStats>) -> Vec<u8> {
let mut groups: Vec<u32> = stats.keys().copied().collect();
groups.sort_unstable();
let mut out = Vec::new();
out.extend_from_slice(&(groups.len() as u32).to_le_bytes());
for group in groups {
out.extend_from_slice(&stats[&group].encode());
}
out
}
pub fn decode_segment_snapshot(data: &[u8]) -> StorageResult<HashMap<u32, GroupSegmentStats>> {
let mut cursor = 0usize;
let take = |data: &[u8], cursor: &mut usize, len: usize| -> StorageResult<Vec<u8>> {
if data.len().saturating_sub(*cursor) < len {
return Err(StorageError::deserialize_error(
"segment snapshot too short",
));
}
let slice = data[*cursor..*cursor + len].to_vec();
*cursor += len;
Ok(slice)
};
let count = u32::from_le_bytes(
take(data, &mut cursor, 4)?
.try_into()
.map_err(|_| StorageError::deserialize_error("segment snapshot count too short"))?,
) as usize;
let mut out = HashMap::with_capacity(count);
for _ in 0..count {
let record = GroupSegmentStats::decode(data, &mut cursor)?;
out.insert(record.group, record);
}
if cursor != data.len() {
return Err(StorageError::deserialize_error(
"unexpected trailing data in segment snapshot",
));
}
Ok(out)
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScanPruneReport {
pub segments_total: usize,
pub segments_pruned: usize,
pub rows_scanned: usize,
pub rows_filtered: usize,
}
impl ScanPruneReport {
pub fn prune_rate(&self) -> f64 {
if self.segments_total == 0 {
0.0
} else {
self.segments_pruned as f64 / self.segments_total as f64
}
}
pub fn filter_rate(&self) -> f64 {
if self.rows_scanned == 0 {
0.0
} else {
self.rows_filtered as f64 / self.rows_scanned as f64
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct FormProfileSnapshot {
pub width_bytes: u64,
pub width_samples: u64,
pub reads: u64,
pub writes: u64,
}
impl FormProfileSnapshot {
pub fn is_unknown(&self) -> bool {
self.width_samples == 0
}
pub fn avg_width_bytes(&self) -> Option<f64> {
if self.width_samples == 0 {
None
} else {
Some(self.width_bytes as f64 / self.width_samples as f64)
}
}
pub fn write_share(&self) -> Option<f64> {
let total = self.reads.saturating_add(self.writes);
if total == 0 {
None
} else {
Some(self.writes as f64 / total as f64)
}
}
pub fn read_to_write_ratio(&self) -> Option<f64> {
if self.writes == 0 {
None
} else {
Some(self.reads as f64 / self.writes as f64)
}
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(32);
out.extend_from_slice(&self.width_bytes.to_le_bytes());
out.extend_from_slice(&self.width_samples.to_le_bytes());
out.extend_from_slice(&self.reads.to_le_bytes());
out.extend_from_slice(&self.writes.to_le_bytes());
out
}
pub fn decode(data: &[u8]) -> StorageResult<Self> {
if data.len() != 32 {
return Err(StorageError::deserialize_error(
"form profile snapshot must hold 32 bytes",
));
}
let mut cursor = 0usize;
let take_u64 = |data: &[u8], cursor: &mut usize| {
let mut bytes = [0u8; 8];
bytes.copy_from_slice(&data[*cursor..*cursor + 8]);
*cursor += 8;
u64::from_le_bytes(bytes)
};
Ok(Self {
width_bytes: take_u64(data, &mut cursor),
width_samples: take_u64(data, &mut cursor),
reads: take_u64(data, &mut cursor),
writes: take_u64(data, &mut cursor),
})
}
}
pub fn estimate_value_bytes(value: &Value) -> u64 {
match value {
Value::Empty => 0,
Value::Null(_) => 0,
Value::Bool(_) => 1,
Value::SmallInt(_) => 2,
Value::Int(_) => 4,
Value::BigInt(_) => 8,
Value::Float(_) => 4,
Value::Double(_) => 8,
Value::Decimal128(_) => 16,
Value::String(s) => s.len() as u64,
Value::FixedString(s) => s.len() as u64,
Value::Blob(b) => b.len() as u64,
Value::Date(_) => 8,
Value::Time(_) => 8,
Value::DateTime(_) => 8,
Value::Uuid(_) => 16,
Value::Interval(_) => 16,
Value::Json(_) | Value::JsonB(_) => 32,
Value::List(_) | Value::Map(_) | Value::Set(_) => 32,
Value::Geography(_) => 32,
Value::Vector(v) => (v.dimension() as u64).saturating_mul(4),
Value::DataSet(_) => 32,
Value::Struct(_) | Value::Array(_) => 32,
Value::Vertex(_) | Value::Edge(_) | Value::EdgeHeader(_) | Value::Path(_) => 32,
Value::VertexId(_) | Value::EdgeId(_) => 8,
}
}
pub fn estimate_props_bytes(props: &[(std::sync::Arc<str>, Value)]) -> u64 {
props.iter().map(|(_, v)| estimate_value_bytes(v)).sum()
}
#[cfg(test)]
mod tests {
use super::super::super::{EdgeSchema, EdgeStrategy, RecordForm};
use super::*;
use crate::edge::edge_table::config::EdgeTableConfig;
use crate::edge::edge_table::core::EdgeStore;
use crate::types::StoragePropertyDef;
use linkrs_core::types::DataType;
use linkrs_core::Value;
fn create_edge_table() -> EdgeStore {
let schema = EdgeSchema {
label_id: 0,
label_name: "knows".into(),
src_label: 0,
dst_label: 0,
properties: vec![],
oe_strategy: EdgeStrategy::Multiple,
ie_strategy: EdgeStrategy::Multiple,
schema_version: 1,
record_form: RecordForm::default(),
};
EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
}
fn create_edge_table_with_props() -> EdgeStore {
let schema = EdgeSchema {
label_id: 0,
label_name: "knows".into(),
src_label: 0,
dst_label: 0,
properties: vec![StoragePropertyDef {
name: "weight".into(),
data_type: DataType::Double,
nullable: false,
default_value: Some(Value::Double(0.0)),
}],
oe_strategy: EdgeStrategy::Multiple,
ie_strategy: EdgeStrategy::Multiple,
schema_version: 1,
record_form: RecordForm::default(),
};
EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
}
#[test]
fn test_deletion_ratio() {
let mut stats = DeletionStats::default();
assert_eq!(stats.deletion_ratio(), 0.0);
assert_eq!(stats.deletion_percentage(), 0.0);
assert!(!stats.is_significant());
stats.total_live_edges = 100;
stats.total_deleted_edges = 50;
assert!((stats.deletion_ratio() - 50.0 / 150.0).abs() < 1e-9);
assert!((stats.deletion_percentage() - 100.0 * 50.0 / 150.0).abs() < 1e-9);
assert!(stats.is_significant());
stats.total_deleted_edges = 5;
assert!((stats.deletion_ratio() - 5.0 / 105.0).abs() < 1e-9);
assert!(!stats.is_significant());
stats.total_live_edges = 0;
stats.total_deleted_edges = 3;
assert_eq!(stats.deletion_ratio(), 1.0);
}
#[test]
fn test_tombstone_stats_accuracy() {
let mut table = create_edge_table_with_props();
table
.insert_edge(0, 1, 0, &[("weight".into(), Value::Double(1.0))], 50)
.unwrap();
table
.insert_edge(0, 2, 0, &[("weight".into(), Value::Double(2.0))], 100)
.unwrap();
table
.insert_edge(0, 3, 0, &[("weight".into(), Value::Double(3.0))], 150)
.unwrap();
table.delete_edge(0, 1, 0, 200).unwrap();
table.delete_edge(0, 2, 0, 250).unwrap();
table.delete_edge(0, 3, 0, 300).unwrap();
let stats = table.mvcc.tombstone_stats();
assert_eq!(stats.count, 3);
assert!(stats.memory_bytes > 0);
assert_eq!(stats.oldest_delete_ts, Some(200));
assert_eq!(stats.newest_delete_ts, Some(300));
}
#[test]
fn test_deletion_stats_tracking() {
let mut table = create_edge_table_with_props();
for i in 0..5u64 {
table
.insert_edge(
0,
1,
i as i64,
&[("weight".into(), Value::Double(i as f64))],
100 + i,
)
.unwrap();
}
let stats = table.deletion_stats();
assert_eq!(stats.total_deleted_edges, 0);
assert_eq!(stats.deletion_percentage(), 0.0);
table.delete_edge(0, 1, 0, 110).unwrap();
table.delete_edge(0, 1, 1, 111).unwrap();
let stats = table.deletion_stats();
assert_eq!(stats.total_deleted_edges, 2);
assert!(stats.deletion_percentage() >= 0.0);
}
#[test]
fn test_deletion_stats_complete_table_deletion() {
let mut table = create_edge_table();
for i in 0..3 {
table.insert_edge(0, 1, i as i64, &[], 100).unwrap();
}
for i in 0..3 {
table.delete_edge(0, 1, i as i64, 110).unwrap();
}
let stats = table.deletion_stats();
assert_eq!(stats.total_deleted_edges, 3);
assert_eq!(stats.total_live_edges, 0);
}
#[test]
fn test_segment_stats_collect_prune_and_roundtrip() {
use crate::cursor::ScanPredicate;
use std::collections::HashMap;
let mut column_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
column_values.insert(
"weight".into(),
vec![
Some(Value::Double(1.0)),
Some(Value::Double(2.0)),
None,
Some(Value::Double(9.0)),
],
);
let stats =
GroupSegmentStats::collect(0, 4096, 3, &[10, 20, 30], &column_values, &HashMap::new());
assert_eq!(stats.group, 0);
assert_eq!(stats.row_count, 4096);
assert_eq!(stats.live_count, 3);
assert_eq!(stats.null_count, 1);
assert_eq!(stats.sort_min, Some(Value::BigInt(10)));
assert_eq!(stats.sort_max, Some(Value::BigInt(30)));
let inside = vec![ScanPredicate::ColumnEqual {
column: "weight".into(),
value: Value::Double(2.0),
}];
assert!(stats.may_contain(&inside));
let outside = vec![ScanPredicate::ColumnRange {
column: "weight".into(),
lower: Some(Value::Double(100.0)),
upper: None,
include_lower: true,
include_upper: true,
}];
assert!(!stats.may_contain(&outside));
let unknown = vec![ScanPredicate::ColumnEqual {
column: "missing".into(),
value: Value::Double(1.0),
}];
assert!(stats.may_contain(&unknown));
let mut cursor = 0usize;
let encoded = stats.encode();
let decoded = GroupSegmentStats::decode(&encoded, &mut cursor).unwrap();
assert_eq!(cursor, encoded.len());
assert_eq!(decoded.group, stats.group);
assert_eq!(decoded.row_count, stats.row_count);
assert_eq!(decoded.live_count, stats.live_count);
assert_eq!(decoded.null_count, stats.null_count);
assert_eq!(decoded.sort_min, stats.sort_min);
assert_eq!(decoded.sort_max, stats.sort_max);
let mut snapshot = HashMap::new();
snapshot.insert(0u32, stats);
let bytes = encode_segment_snapshot(&snapshot);
let restored = decode_segment_snapshot(&bytes).unwrap();
assert_eq!(restored.len(), 1);
assert!(restored.contains_key(&0));
let mut bad = bytes.clone();
bad.push(0);
assert!(decode_segment_snapshot(&bad).is_err());
}
#[test]
fn test_segment_stats_widen_only_bounds() {
use std::collections::HashMap;
let mut first_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
first_values.insert(
"weight".into(),
vec![Some(Value::Double(5.0)), Some(Value::Double(6.0))],
);
let mut stats =
GroupSegmentStats::collect(0, 4096, 2, &[5], &first_values, &HashMap::new());
let mut second_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
second_values.insert("weight".into(), vec![Some(Value::Double(1.0))]);
let fresh = GroupSegmentStats::collect(1, 4096, 1, &[50], &second_values, &HashMap::new());
stats.widen_with(&fresh);
assert_eq!(stats.row_count, 4096);
assert_eq!(stats.live_count, 1);
assert_eq!(stats.sort_min, Some(Value::BigInt(5)));
assert_eq!(stats.sort_max, Some(Value::BigInt(50)));
let column = stats.columns.get("weight").unwrap();
assert_eq!(column.min_value, Some(Value::Double(1.0)));
assert_eq!(column.max_value, Some(Value::Double(6.0)));
}
#[test]
fn test_prune_report_rates_are_observable() {
let report = ScanPruneReport {
segments_total: 4,
segments_pruned: 1,
rows_scanned: 100,
rows_filtered: 25,
};
assert!((report.prune_rate() - 0.25).abs() < 1e-9);
assert!((report.filter_rate() - 0.25).abs() < 1e-9);
assert_eq!(ScanPruneReport::default().prune_rate(), 0.0);
assert_eq!(ScanPruneReport::default().filter_rate(), 0.0);
}
}