use crate::version::Version;
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
#[non_exhaustive]
pub enum StorageStatus {
Healthy,
FullCompactionAvailable,
TightCompactionAvailable,
ReadOnlyOutOfSpace,
CompactionInProgress,
}
#[must_use]
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct FilterMemory {
pub serialised_bytes: u64,
pub resident_bytes: u64,
pub budget_bytes: Option<u64>,
pub over_budget: bool,
}
#[must_use]
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct StorageStats {
pub used_bytes: u64,
pub capacity_bytes: Option<u64>,
pub available_bytes: Option<u64>,
pub compaction_possible: bool,
pub full_compaction_bytes: u64,
pub tight_compaction_bytes: u64,
pub item_count: u64,
pub table_count: u64,
pub avg_entry_on_disk_bytes: u64,
pub avg_key_bytes: Option<u64>,
pub avg_value_bytes: Option<u64>,
pub reclaimable_bytes_estimate: u64,
pub status: StorageStatus,
pub blob_references: BlobReferenceStats,
}
#[must_use]
#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
pub struct BlobReferenceStats {
pub count: u64,
pub depth: u64,
}
pub(crate) fn blob_reference_stats<'a>(
tables: impl IntoIterator<Item = &'a crate::table::Table>,
) -> crate::Result<BlobReferenceStats> {
let mut comparator = None;
let mut spans = Vec::new();
collect_spans(tables, &mut comparator, &mut spans)?;
Ok(comparator.map_or_else(BlobReferenceStats::default, |cmp| {
span_stats(&mut spans, cmp.as_ref())
}))
}
pub(crate) fn blob_file_depths<'a>(
tables: impl IntoIterator<Item = &'a crate::table::Table>,
) -> crate::Result<Vec<(crate::vlog::BlobFileId, u64)>> {
let mut comparator = None;
let mut spans = Vec::new();
collect_spans(tables, &mut comparator, &mut spans)?;
Ok(comparator.map_or_else(Vec::new, |cmp| span_depths(&mut spans, cmp.as_ref())))
}
fn span_depths(
spans: &mut [Span<'_>],
cmp: &dyn crate::comparator::UserComparator,
) -> Vec<(crate::vlog::BlobFileId, u64)> {
let (merged, _) = merge_spans(spans, cmp);
let events = sweep_events(&merged, cmp);
let mut open_after = Vec::with_capacity(events.len());
let mut bounds = alloc::vec![(0usize, 0usize); merged.len()];
let mut open = 0u64;
for (position, &(_, is_end, index)) in events.iter().enumerate() {
if let Some(bound) = bounds.get_mut(index) {
if is_end {
open -= 1;
bound.1 = position;
} else {
open += 1;
bound.0 = position;
}
}
open_after.push(open);
}
let range_max = RangeMax::new(open_after);
let mut per_file: Vec<(crate::vlog::BlobFileId, u64)> = Vec::new();
for ((id, _, _), (start, end)) in merged.iter().zip(bounds) {
debug_assert!(end > start, "a span ends after it starts");
let depth = range_max.max(start, end - 1);
match per_file.last_mut() {
Some((last, value)) if last == id => *value = (*value).max(depth),
_ => per_file.push((*id, depth)),
}
}
per_file
}
struct RangeMax {
levels: Vec<Vec<u64>>,
}
impl RangeMax {
fn new(values: Vec<u64>) -> Self {
let mut levels = alloc::vec![values];
let mut width = 1;
while let Some(previous) = levels.last()
&& previous.len() > width
{
let next: Vec<u64> = previous
.iter()
.zip(previous.iter().skip(width))
.map(|(a, b)| (*a).max(*b))
.collect();
levels.push(next);
width *= 2;
}
Self { levels }
}
fn max(&self, from: usize, to: usize) -> u64 {
debug_assert!(from <= to, "an empty range has no maximum");
let len = to - from + 1;
let level = (usize::BITS - 1 - len.leading_zeros()) as usize;
let Some(values) = self.levels.get(level) else {
return 0;
};
let right = to + 1 - (1 << level);
let left = values.get(from).copied().unwrap_or(0);
left.max(values.get(right).copied().unwrap_or(0))
}
}
pub(crate) fn pick_overlapping_blob_files<'a>(
tables: impl IntoIterator<Item = &'a crate::table::Table>,
fixed: &[(crate::vlog::BlobFileId, usize)],
ranked: &[(crate::vlog::BlobFileId, usize, u64)],
budget: u64,
) -> crate::Result<(Vec<bool>, u64)> {
let mut comparator = None;
let mut spans = Vec::new();
collect_spans(tables, &mut comparator, &mut spans)?;
Ok(match comparator {
Some(cmp) => pick_overlapping(&mut spans, fixed, ranked, budget, cmp.as_ref()),
None => (alloc::vec![false; ranked.len()], 0),
})
}
#[derive(Clone, Copy)]
enum Role {
Fixed(usize),
Ranked(usize),
}
fn pick_overlapping(
spans: &mut Vec<Span<'_>>,
fixed: &[(crate::vlog::BlobFileId, usize)],
ranked: &[(crate::vlog::BlobFileId, usize, u64)],
budget: u64,
cmp: &dyn crate::comparator::UserComparator,
) -> (Vec<bool>, u64) {
let mut roles: Vec<(crate::vlog::BlobFileId, Role)> = fixed
.iter()
.map(|&(id, group)| (id, Role::Fixed(group)))
.chain(
ranked
.iter()
.enumerate()
.map(|(rank, &(id, _, _))| (id, Role::Ranked(rank))),
)
.collect();
roles.sort_unstable_by_key(|&(id, _)| id);
let role_of = |id: crate::vlog::BlobFileId| {
roles
.binary_search_by_key(&id, |&(file, _)| file)
.ok()
.and_then(|at| roles.get(at))
.map(|&(_, role)| role)
};
spans.retain(|&(id, _, _)| role_of(id).is_some());
let (merged, _) = merge_spans(spans, cmp);
let events = sweep_events(&merged, cmp);
let positions = events.len();
let mut bounds = alloc::vec![(0usize, 0usize); merged.len()];
for (position, &(_, is_end, index)) in events.iter().enumerate() {
if let Some(bound) = bounds.get_mut(index) {
if is_end {
bound.1 = position;
} else {
bound.0 = position;
}
}
}
let group_and_rank = |index: usize| {
let role = merged.get(index).and_then(|&(id, _, _)| role_of(id))?;
Some(match role {
Role::Fixed(group) => (group, None),
Role::Ranked(rank) => (ranked.get(rank)?.1, Some(rank)),
})
};
let groups = fixed
.iter()
.map(|&(_, group)| group)
.chain(ranked.iter().map(|&(_, group, _)| group))
.max()
.map_or(0, |group| group + 1);
let mut spans_of: Vec<Vec<(usize, usize)>> = alloc::vec![Vec::new(); ranked.len()];
for (index, &bound) in bounds.iter().enumerate() {
if let Some((_, Some(rank))) = group_and_rank(index)
&& let Some(own) = spans_of.get_mut(rank)
{
own.push(bound);
}
}
let mut partner: Vec<Option<(u64, usize)>> = alloc::vec![None; ranked.len()];
let mut started: Vec<PrefixCheapest> = (0..groups)
.map(|_| PrefixCheapest::new(positions))
.collect();
for &(_, is_end, index) in &events {
let Some((group, Some(rank))) = group_and_rank(index) else {
continue;
};
let (Some(&(start, end)), Some(&(_, _, cost)), Some(index_of_group)) =
(bounds.get(index), ranked.get(rank), started.get_mut(group))
else {
continue;
};
if is_end {
let found = index_of_group
.before(positions - 1 - start)
.best_except(rank);
if let (Some(found), Some(best)) = (found, partner.get_mut(rank))
&& best.is_none_or(|best| found < best)
{
*best = Some(found);
}
} else {
index_of_group.add(positions - 1 - end, (cost, rank));
}
}
let mut relocated: Vec<PrefixMax> = (0..groups).map(|_| PrefixMax::new(positions)).collect();
for (index, &(start, end)) in bounds.iter().enumerate() {
if let Some((group, None)) = group_and_rank(index)
&& let Some(tree) = relocated.get_mut(group)
{
tree.raise(start, end);
}
}
let mut chosen = alloc::vec![false; ranked.len()];
let mut spent = 0u64;
for (rank, &(_, group, cost)) in ranked.iter().enumerate() {
if chosen.get(rank) != Some(&false) || spent + cost > budget {
continue;
}
let own = spans_of.get(rank).map_or(&[][..], Vec::as_slice);
let joins = relocated
.get(group)
.is_some_and(|tree| own.iter().any(|&(start, end)| tree.max_before(end) > start));
let with = if joins {
None
} else {
match partner.get(rank).copied().flatten() {
Some((partner_cost, partner_rank)) if spent + cost + partner_cost <= budget => {
Some(partner_rank)
}
_ => continue,
}
};
for pick in core::iter::once(rank).chain(with) {
let (Some(flag), Some(&(_, _, pick_cost)), Some(tree)) = (
chosen.get_mut(pick),
ranked.get(pick),
relocated.get_mut(group),
) else {
continue;
};
*flag = true;
spent += pick_cost;
for &(start, end) in spans_of.get(pick).map_or(&[][..], Vec::as_slice) {
tree.raise(start, end);
}
}
}
(chosen, spent)
}
struct PrefixMax(Vec<usize>);
impl PrefixMax {
fn new(positions: usize) -> Self {
Self(alloc::vec![0; positions])
}
fn raise(&mut self, at: usize, value: usize) {
let mut node = at + 1;
while let Some(slot) = self.0.get_mut(node - 1) {
*slot = (*slot).max(value);
node += node.isolate_lowest_one();
}
}
fn max_before(&self, before: usize) -> usize {
let mut node = before;
let mut max = 0;
while node > 0 {
if let Some(&value) = self.0.get(node - 1) {
max = max.max(value);
}
node &= node - 1;
}
max
}
}
#[derive(Clone, Copy, Default)]
struct TwoCheapest([Option<(u64, usize)>; 2]);
impl TwoCheapest {
fn add(&mut self, offer: (u64, usize)) {
let [first, second] = self.0;
let mut all = [first, second, Some(offer)];
all.sort_unstable_by_key(|offer| offer.unwrap_or((u64::MAX, usize::MAX)));
let mut kept = [None; 2];
let mut slots = kept.iter_mut();
let mut first_rank = None;
for offer in all.into_iter().flatten() {
if first_rank == Some(offer.1) {
continue;
}
let Some(slot) = slots.next() else {
break;
};
first_rank.get_or_insert(offer.1);
*slot = Some(offer);
}
self.0 = kept;
}
fn best_except(&self, rank: usize) -> Option<(u64, usize)> {
self.0.into_iter().flatten().find(|offer| offer.1 != rank)
}
}
struct PrefixCheapest(Vec<TwoCheapest>);
impl PrefixCheapest {
fn new(positions: usize) -> Self {
Self(alloc::vec![TwoCheapest::default(); positions])
}
fn add(&mut self, at: usize, offer: (u64, usize)) {
let mut node = at + 1;
while let Some(slot) = self.0.get_mut(node - 1) {
slot.add(offer);
node += node.isolate_lowest_one();
}
}
fn before(&self, before: usize) -> TwoCheapest {
let mut node = before;
let mut out = TwoCheapest::default();
while node > 0 {
if let Some(slot) = self.0.get(node - 1) {
for offer in slot.0.into_iter().flatten() {
out.add(offer);
}
}
node &= node - 1;
}
out
}
}
type Span<'k> = (
crate::vlog::BlobFileId,
&'k crate::UserKey,
&'k crate::UserKey,
);
fn collect_spans<'a>(
tables: impl IntoIterator<Item = &'a crate::table::Table>,
comparator: &mut Option<crate::comparator::SharedComparator>,
spans: &mut Vec<Span<'a>>,
) -> crate::Result<()> {
for table in tables {
comparator.get_or_insert_with(|| table.comparator.clone());
spans.extend(
table
.blob_links()?
.iter()
.map(|link| (link.blob_file_id, &link.first_key, &link.last_key)),
);
}
Ok(())
}
fn span_stats(
spans: &mut [Span<'_>],
cmp: &dyn crate::comparator::UserComparator,
) -> BlobReferenceStats {
let (merged, count) = merge_spans(spans, cmp);
let (mut open, mut depth) = (0u64, 0u64);
for (_, is_end, _) in sweep_events(&merged, cmp) {
if is_end {
open -= 1;
} else {
open += 1;
depth = depth.max(open);
}
}
BlobReferenceStats { count, depth }
}
fn merge_spans<'k>(
spans: &mut [Span<'k>],
cmp: &dyn crate::comparator::UserComparator,
) -> (Vec<Span<'k>>, u64) {
use core::cmp::Ordering;
for span in spans.iter_mut() {
if cmp.compare(span.1, span.2) == Ordering::Greater {
core::mem::swap(&mut span.1, &mut span.2);
}
}
spans.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| cmp.compare(a.1, b.1)));
let mut merged: Vec<Span<'k>> = Vec::with_capacity(spans.len());
let mut count = 0u64;
for &span in spans.iter() {
match merged.last_mut() {
Some((id, _, last))
if *id == span.0 && cmp.compare(span.1, last) != Ordering::Greater =>
{
if cmp.compare(span.2, last) == Ordering::Greater {
*last = span.2;
}
}
last => {
if last.is_none_or(|(id, _, _)| *id != span.0) {
count += 1;
}
merged.push(span);
}
}
}
(merged, count)
}
fn sweep_events<'k>(
merged: &[Span<'k>],
cmp: &dyn crate::comparator::UserComparator,
) -> Vec<(&'k crate::UserKey, bool, usize)> {
let mut events: Vec<(&'k crate::UserKey, bool, usize)> = merged
.iter()
.enumerate()
.flat_map(|(index, &(_, first, last))| [(first, false, index), (last, true, index)])
.collect();
events.sort_by(|a, b| cmp.compare(a.0, b.0).then(a.1.cmp(&b.1)));
events
}
#[must_use]
#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
pub struct ApproximateRangeStats {
pub bytes: u64,
pub key_count: u64,
}
#[must_use]
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct SegmentStats {
pub table_id: crate::TableId,
pub level: usize,
pub used_bytes: u64,
pub item_count: u64,
pub reads: u64,
pub last_access_secs: u64,
pub blob_references: BlobReferenceStats,
}
#[must_use]
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LevelStats {
pub level: usize,
pub segment_count: usize,
pub used_bytes: u64,
pub item_count: u64,
pub reads: u64,
pub last_access_secs: u64,
pub blob_references: BlobReferenceStats,
pub segments: Vec<SegmentStats>,
}
#[must_use]
#[derive(Copy, Clone, Debug, Default, PartialEq)]
pub struct RangeCardinality {
pub rows: u64,
pub selectivity: f64,
}
pub trait StorageStatistics {
fn storage_stats(&self) -> crate::Result<StorageStats>;
fn level_segment_stats(&self) -> crate::Result<Vec<LevelStats>>;
fn compaction_debt(&self, strategy: &dyn crate::compaction::CompactionStrategy) -> u64;
#[cfg(feature = "metrics")]
fn cache_stats(&self) -> crate::CacheStats;
}
impl<T: crate::AbstractTree + ?Sized> StorageStatistics for T {
fn storage_stats(&self) -> crate::Result<StorageStats> {
crate::AbstractTree::storage_stats(self)
}
fn level_segment_stats(&self) -> crate::Result<Vec<LevelStats>> {
crate::AbstractTree::level_segment_stats(self)
}
fn compaction_debt(&self, strategy: &dyn crate::compaction::CompactionStrategy) -> u64 {
crate::AbstractTree::compaction_debt(self, strategy)
}
#[cfg(feature = "metrics")]
fn cache_stats(&self) -> crate::CacheStats {
crate::AbstractTree::cache_stats(self)
}
}
impl StorageStats {
#[must_use]
pub fn estimated_remaining_entries(&self, budget_bytes: u64) -> u64 {
budget_bytes
.checked_div(self.avg_entry_on_disk_bytes)
.unwrap_or(0)
}
}
pub(crate) fn compute_used_bytes(version: &Version) -> crate::Result<u64> {
let mut used_bytes = 0u64;
for table in version.iter_tables() {
used_bytes += table_on_disk_bytes(table)?;
}
for blob in version.blob_files.iter() {
used_bytes += blob_on_disk_bytes(blob)?;
}
Ok(used_bytes)
}
#[cfg(feature = "std")]
pub(crate) fn compute_used_bytes_but(
version: &Version,
tables: &crate::HashSet<crate::TableId>,
blobs: &crate::HashSet<crate::vlog::BlobFileId>,
) -> crate::Result<u64> {
let mut used_bytes = 0u64;
for table in version.iter_tables() {
if !tables.contains(&table.id()) {
used_bytes += table_on_disk_bytes(table)?;
}
}
for blob in version.blob_files.iter() {
if !blobs.contains(&blob.id()) {
used_bytes += blob_on_disk_bytes(blob)?;
}
}
Ok(used_bytes)
}
pub(crate) fn blob_on_disk_bytes(blob: &crate::vlog::BlobFile) -> crate::Result<u64> {
Ok(crate::file::on_disk_bytes(&*blob.0.fs, &blob.0.path)?)
}
pub(crate) fn table_on_disk_bytes(table: &crate::table::Table) -> crate::Result<u64> {
#[cfg_attr(not(feature = "std"), expect(unused_mut, reason = "no sidecar arm"))]
let mut bytes = crate::file::on_disk_bytes(&*table.fs, &table.path)?;
#[cfg(feature = "std")]
if table.restrict_lower_bound().is_some() {
match table
.fs
.metadata(&crate::restrict_bound::sidecar_path(&table.path))
{
Ok(m) => bytes += m.len,
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => return Err(e.into()),
}
}
Ok(bytes)
}
pub(crate) fn full_compaction_demand_bytes(version: &Version) -> crate::Result<u64> {
let mut largest = 0u64;
for level in version.iter_levels() {
let mut size = 0u64;
for run in level.iter() {
for table in run.iter() {
size += table.live_file_size()?;
}
}
largest = largest.max(size);
}
Ok(largest)
}
pub(crate) fn compute_storage_stats(
version: &Version,
is_compacting: bool,
value_bytes_are_user_values: bool,
) -> crate::Result<StorageStats> {
let mut used_bytes = 0u64;
let mut item_count = 0u64;
let mut table_count = 0u64;
let mut reclaimable_entries = 0u64;
let mut sum_key = 0u64;
let mut sum_value = 0u64;
let mut all_have_shape = true;
for table in version.iter_tables() {
let m = &table.metadata;
let on_disk = table_on_disk_bytes(table)?;
used_bytes += on_disk;
let live_items = table.live_item_count(crate::table::util::ReadCharge::Untraced)?;
let share = |total: u64| -> u64 {
if live_items == m.item_count || m.item_count == 0 {
return total;
}
u64::try_from(u128::from(total) * u128::from(live_items) / u128::from(m.item_count))
.unwrap_or(total)
};
item_count += live_items;
table_count += 1;
reclaimable_entries += share(m.weak_tombstone_reclaimable);
match (
m.sum_user_key_bytes.map(share),
m.sum_value_bytes.map(share),
) {
(Some(k), Some(v)) => {
sum_key += k;
sum_value += v;
}
_ => all_have_shape = false,
}
}
for blob in version.blob_files.iter() {
used_bytes += blob_on_disk_bytes(blob)?;
}
let avg_entry_on_disk_bytes = used_bytes.checked_div(item_count).unwrap_or(0);
let have_shape = all_have_shape && item_count > 0;
let avg_key_bytes = have_shape.then(|| sum_key / item_count);
let avg_value_bytes =
(have_shape && value_bytes_are_user_values).then(|| sum_value / item_count);
let reclaimable_bytes_estimate = reclaimable_entries * avg_entry_on_disk_bytes;
let full_compaction_bytes = full_compaction_demand_bytes(version)?;
let tight_compaction_bytes = crate::tree::MIN_RESERVED_HEADROOM;
let status = if is_compacting {
StorageStatus::CompactionInProgress
} else {
StorageStatus::Healthy
};
Ok(StorageStats {
used_bytes,
capacity_bytes: None,
available_bytes: None,
compaction_possible: true,
full_compaction_bytes,
tight_compaction_bytes,
item_count,
table_count,
avg_entry_on_disk_bytes,
avg_key_bytes,
avg_value_bytes,
reclaimable_bytes_estimate,
status,
blob_references: blob_reference_stats(version.iter_tables())?,
})
}
pub(crate) fn compute_level_segment_stats(version: &Version) -> crate::Result<Vec<LevelStats>> {
use core::sync::atomic::Ordering::Relaxed;
let mut levels = Vec::with_capacity(version.level_count());
for (level, run_group) in version.iter_levels().enumerate() {
let mut segments = Vec::new();
let mut used_bytes = 0u64;
let mut item_count = 0u64;
let mut reads = 0u64;
let mut last_access_secs = 0u64;
let mut comparator = None;
let mut level_spans = Vec::new();
let mut table_spans = Vec::new();
for run in run_group.iter() {
for table in run.iter() {
collect_spans([table], &mut comparator, &mut table_spans)?;
let blob_references = comparator
.as_ref()
.map_or_else(BlobReferenceStats::default, |cmp| {
span_stats(&mut table_spans, cmp.as_ref())
});
level_spans.append(&mut table_spans);
let on_disk = table_on_disk_bytes(table)?;
let items = table.live_item_count(crate::table::util::ReadCharge::Untraced)?;
let seg_reads = table.read_count.load(Relaxed);
let seg_access = table.last_access_secs.load(Relaxed);
used_bytes += on_disk;
item_count += items;
reads = reads.saturating_add(seg_reads);
last_access_secs = last_access_secs.max(seg_access);
segments.push(SegmentStats {
table_id: table.metadata.id,
level,
used_bytes: on_disk,
item_count: items,
reads: seg_reads,
last_access_secs: seg_access,
blob_references,
});
}
}
levels.push(LevelStats {
level,
segment_count: segments.len(),
used_bytes,
item_count,
reads,
last_access_secs,
blob_references: comparator.map_or_else(BlobReferenceStats::default, |cmp| {
span_stats(&mut level_spans, cmp.as_ref())
}),
segments,
});
}
Ok(levels)
}
#[cfg(test)]
mod tests;