linkrs_storage/edge/csr_with_properties/

maintenance.rs

use super::{CsrWithProperties, RowVisibility};
use crate::edge::csr_shared::EdgeIdBitSet;
use crate::edge::property_schema::PropertySchema;
use linkrs_core::types::{EdgeId, Timestamp};

impl CsrWithProperties {
    /// Pre-allocate row and column capacity for an incoming batch.
    ///
    /// Sizes visibility, reverse indexes and every property column once so a
    /// batch of `additional` edges lands in reserved gaps instead of growing
    /// row by row. Inline stubs keep no rows and skip reservation.
    pub fn reserve_for_batch(&mut self, additional: usize) {
        if additional == 0 || self.is_inline_stub() {
            return;
        }
        self.visibility.reserve(additional);
        self.row_to_edge.reserve(additional);
        self.free_list.reserve(additional);
        for col in &mut self.property_columns {
            col.reserve(additional);
        }
    }

    pub fn compaction_stats(&self) -> crate::edge::property_schema::PropertyCompactionStats {
        let tombstone_count = self
            .visibility
            .iter()
            .filter(|v| v.delete_ts.is_some())
            .count();
        let live_records = self.visibility.iter().filter(|v| v.is_live()).count();
        let mut reclaimable_bytes = 0usize;
        for v in &self.visibility {
            if v.delete_ts.is_some() {
                reclaimable_bytes += 32 * self.property_schema.len();
            }
        }
        crate::edge::property_schema::PropertyCompactionStats {
            tombstone_count,
            total_records: self.visibility.len(),
            live_records,
            reclaimable_bytes,
        }
    }

    /// Garbage-collect property version-chain entries that no active snapshot
    /// can observe. Returns the total number of before-images removed.
    pub fn gc_property_versions(&mut self, min_active_snapshot_ts: Timestamp) -> usize {
        self.property_columns
            .iter_mut()
            .map(|col| col.gc_versions(min_active_snapshot_ts))
            .sum()
    }

    pub fn used_memory_size(&self) -> usize {
        let mut total = std::mem::size_of::<Self>();
        total += self.visibility.capacity() * std::mem::size_of::<RowVisibility>();
        total += self.edge_map_memory_bytes();
        total += self.free_list.capacity() * std::mem::size_of::<u32>();
        total += self.row_to_edge.capacity() * std::mem::size_of::<Option<EdgeId>>();
        total += self.column_index.len()
            * (std::mem::size_of::<String>() + std::mem::size_of::<usize>());
        for col in &self.property_columns {
            total += col.memory_size();
        }
        total += self.property_schema.len() * std::mem::size_of::<PropertySchema>();
        total
    }

    /// Split property memory into mapping bytes (edge map plus reverse index
    /// plus free list) and row bytes (visibility plus columns). Caliber for
    /// per-edge accounting without walking rows.
    pub fn mapping_row_bytes(&self) -> (usize, usize) {
        let mapping = self.edge_map_memory_bytes()
            + self.row_to_edge.capacity() * std::mem::size_of::<Option<EdgeId>>()
            + self.free_list.capacity() * std::mem::size_of::<u32>();
        let mut rows = self.visibility.capacity() * std::mem::size_of::<RowVisibility>();
        for col in &self.property_columns {
            rows += col.memory_size();
        }
        (mapping, rows)
    }

    pub(crate) fn reclaim_slots(
        &mut self,
        valid_edge_ids: &EdgeIdBitSet,
        retention_bound: Timestamp,
    ) -> usize {
        if retention_bound == Timestamp::MAX {
            return 0;
        }
        let mut to_reclaim = Vec::new();
        for (idx, vis) in self.visibility.iter().enumerate() {
            if vis.is_virgin() {
                continue;
            }
            // O(1) ownership check via the reverse index instead of scanning
            // the full edge map for every row.
            let has_live_edge = self
                .row_to_edge
                .get(idx)
                .and_then(|slot| *slot)
                .is_some_and(|eid| valid_edge_ids.contains(&eid));
            if has_live_edge {
                continue;
            }
            if let Some(del_ts) = vis.delete_ts {
                // Exclusive waterfront: deletable exactly when invisible to
                // every snapshot at or past the cutoff.
                if crate::mvcc_visibility::Visibility::is_gc_eligible(del_ts, retention_bound) {
                    to_reclaim.push(idx);
                }
            }
        }
        if to_reclaim.is_empty() {
            return 0;
        }
        for &idx in &to_reclaim {
            self.visibility[idx].create_ts = 0;
            self.visibility[idx].delete_ts = None;
            if let Some(taken) = self.row_to_edge.get_mut(idx).and_then(|slot| slot.take()) {
                self.map_remove(taken);
            }
            // Reclaimed rows are virgin after the stamp clear, so the free
            // list admits each of them exactly once with no membership set.
            self.free_list.push(idx as u32);
            self.row_count = self.row_count.saturating_sub(1);
        }
        to_reclaim.len()
    }

    /// Recompute persisted per-column statistics from flush buffers.
    ///
    /// Called before the property file is serialized so statistics follow the
    /// checkpoint instead of drifting. Only columns marked dirty are
    /// recomputed; clean columns keep their persisted statistics. Columns
    /// that fail to compute keep their previous statistics.
    pub fn refresh_column_stats(&mut self) {
        if self.dirty_columns.is_empty() {
            for col in &mut self.property_columns {
                if let Ok(stats) = col.compute_stats() {
                    col.set_stats(stats);
                }
            }
            return;
        }
        // Dirt marks stay until the checkpoint clears them: stats refresh
        // must not consume the marks that drive dirty-column persistence.
        let dirty: Vec<usize> = self.dirty_columns.iter().copied().collect();
        for idx in dirty {
            if let Some(col) = self.property_columns.get_mut(idx) {
                if let Ok(stats) = col.compute_stats() {
                    col.set_stats(stats);
                }
            }
        }
    }

    /// Refresh statistics for one column only. Used by column-level
    /// checkpoint follow-up when only a subset changed.
    pub fn refresh_column_stats_for(&mut self, column: &str) {
        if let Some(&idx) = self.column_index.get(column) {
            if let Some(col) = self.property_columns.get_mut(idx) {
                if let Ok(stats) = col.compute_stats() {
                    col.set_stats(stats);
                }
            }
        }
    }
}