featherdb_mvcc/

version.rs

//! Version chain storage for MVCC

use crate::Snapshot;
use bytes::{Buf, BufMut};
use featherdb_core::{Error, Result, TransactionId, Value};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};

/// Pointer to a version in the version store
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct VersionPtr(pub u64);

impl VersionPtr {
    pub const NULL: VersionPtr = VersionPtr(0);

    pub fn is_null(&self) -> bool {
        self.0 == 0
    }
}

/// A versioned value with MVCC metadata
#[derive(Debug, Clone)]
pub struct VersionedValue {
    /// Current (latest) version data
    pub current: Vec<Value>,
    /// Transaction that created the current version
    pub created_by: TransactionId,
    /// Transaction that deleted this row (if any)
    pub deleted_by: Option<TransactionId>,
    /// Pointer to older versions
    pub version_chain: Option<VersionPtr>,
}

impl VersionedValue {
    /// Create a new versioned value
    pub fn new(data: Vec<Value>, created_by: TransactionId) -> Self {
        VersionedValue {
            current: data,
            created_by,
            deleted_by: None,
            version_chain: None,
        }
    }

    /// Mark as deleted
    pub fn mark_deleted(&mut self, deleted_by: TransactionId) {
        self.deleted_by = Some(deleted_by);
    }

    /// Check if visible to a snapshot
    pub fn is_visible(&self, snapshot: &Snapshot) -> bool {
        snapshot.is_visible(self.created_by, self.deleted_by)
    }

    /// Serialize to bytes
    pub fn serialize(&self) -> Vec<u8> {
        let mut buf = Vec::new();

        // created_by
        buf.put_u64_le(self.created_by.0);

        // deleted_by
        match self.deleted_by {
            Some(txn) => {
                buf.put_u8(1);
                buf.put_u64_le(txn.0);
            }
            None => {
                buf.put_u8(0);
            }
        }

        // version_chain
        match self.version_chain {
            Some(ptr) => {
                buf.put_u8(1);
                buf.put_u64_le(ptr.0);
            }
            None => {
                buf.put_u8(0);
            }
        }

        // Number of values
        buf.put_u32_le(self.current.len() as u32);

        // Values
        for value in &self.current {
            value.serialize(&mut buf);
        }

        buf
    }

    /// Deserialize from bytes
    pub fn deserialize(data: &[u8]) -> Result<Self> {
        let mut cursor = data;

        if cursor.remaining() < 9 {
            return Err(Error::Serialization("VersionedValue too small".into()));
        }

        let created_by = TransactionId(cursor.get_u64_le());

        let deleted_by = if cursor.get_u8() == 1 {
            Some(TransactionId(cursor.get_u64_le()))
        } else {
            None
        };

        let version_chain = if cursor.get_u8() == 1 {
            Some(VersionPtr(cursor.get_u64_le()))
        } else {
            None
        };

        let value_count = cursor.get_u32_le() as usize;
        let mut current = Vec::with_capacity(value_count);

        for _ in 0..value_count {
            current.push(Value::deserialize(&mut cursor)?);
        }

        Ok(VersionedValue {
            current,
            created_by,
            deleted_by,
            version_chain,
        })
    }
}

/// An old version in the version chain
#[derive(Debug, Clone)]
pub struct OldVersion {
    /// Row data
    pub data: Vec<Value>,
    /// Transaction that created this version
    pub created_by: TransactionId,
    /// Transaction that deleted this version (replaced it)
    pub deleted_by: Option<TransactionId>,
    /// Pointer to even older version
    pub prev_version: Option<VersionPtr>,
}

impl OldVersion {
    /// Serialize to bytes
    pub fn serialize(&self) -> Vec<u8> {
        let mut buf = Vec::new();

        buf.put_u64_le(self.created_by.0);

        match self.deleted_by {
            Some(txn) => {
                buf.put_u8(1);
                buf.put_u64_le(txn.0);
            }
            None => {
                buf.put_u8(0);
            }
        }

        match self.prev_version {
            Some(ptr) => {
                buf.put_u8(1);
                buf.put_u64_le(ptr.0);
            }
            None => {
                buf.put_u8(0);
            }
        }

        buf.put_u32_le(self.data.len() as u32);
        for value in &self.data {
            value.serialize(&mut buf);
        }

        buf
    }

    /// Deserialize from bytes
    pub fn deserialize(data: &[u8]) -> Result<Self> {
        let mut cursor = data;

        let created_by = TransactionId(cursor.get_u64_le());

        let deleted_by = if cursor.get_u8() == 1 {
            Some(TransactionId(cursor.get_u64_le()))
        } else {
            None
        };

        let prev_version = if cursor.get_u8() == 1 {
            Some(VersionPtr(cursor.get_u64_le()))
        } else {
            None
        };

        let value_count = cursor.get_u32_le() as usize;
        let mut data_values = Vec::with_capacity(value_count);

        for _ in 0..value_count {
            data_values.push(Value::deserialize(&mut cursor)?);
        }

        Ok(OldVersion {
            data: data_values,
            created_by,
            deleted_by,
            prev_version,
        })
    }
}

/// In-memory version store for old versions
pub struct VersionStore {
    /// Old versions, keyed by VersionPtr
    versions: HashMap<VersionPtr, OldVersion>,
    /// Next version pointer to allocate
    next_ptr: AtomicU64,
}

impl VersionStore {
    /// Create a new version store
    pub fn new() -> Self {
        VersionStore {
            versions: HashMap::new(),
            next_ptr: AtomicU64::new(1), // 0 is NULL
        }
    }

    /// Insert an old version
    pub fn insert(&mut self, version: OldVersion) -> VersionPtr {
        let ptr = VersionPtr(self.next_ptr.fetch_add(1, Ordering::SeqCst));
        self.versions.insert(ptr, version);
        ptr
    }

    /// Get an old version
    pub fn get(&self, ptr: VersionPtr) -> Option<&OldVersion> {
        self.versions.get(&ptr)
    }

    /// Remove an old version (during GC)
    pub fn remove(&mut self, ptr: VersionPtr) -> Option<OldVersion> {
        self.versions.remove(&ptr)
    }

    /// Get the number of versions stored
    pub fn len(&self) -> usize {
        self.versions.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.versions.is_empty()
    }
}

impl Default for VersionStore {
    fn default() -> Self {
        Self::new()
    }
}

/// Statistics from garbage collection
#[derive(Debug, Clone, Default)]
pub struct GcStats {
    /// Number of versions removed
    pub versions_removed: usize,
    /// Approximate bytes freed
    pub bytes_freed: usize,
}

impl VersionStore {
    /// Garbage collect old versions that no active transaction can see.
    ///
    /// A version is garbage if:
    /// 1. It was deleted by a transaction older than the oldest active transaction
    ///    (meaning all active transactions can see the deletion)
    /// 2. There's no active transaction that could possibly need this version
    ///
    /// # Arguments
    /// * `oldest_active_txn` - The ID of the oldest active transaction, or None if no active transactions
    ///
    /// # Returns
    /// Statistics about what was collected
    pub fn gc(&mut self, oldest_active_txn: Option<featherdb_core::TransactionId>) -> GcStats {
        let mut stats = GcStats::default();

        // If no active transactions, everything can be garbage collected
        // except versions that have never been deleted (they're still current)
        let oldest = match oldest_active_txn {
            Some(txn) => txn.0,
            None => u64::MAX, // All versions with deleted_by can be collected
        };

        // Collect pointers to remove (can't modify while iterating)
        let to_remove: Vec<VersionPtr> = self
            .versions
            .iter()
            .filter_map(|(ptr, version)| {
                // Version is garbage if it was deleted by a transaction
                // that's older than all active transactions
                // (meaning everyone can see the deletion)
                if let Some(deleted_by) = version.deleted_by {
                    if deleted_by.0 < oldest {
                        // All active transactions can see this deletion,
                        // so this old version is no longer needed
                        return Some(*ptr);
                    }
                }
                None
            })
            .collect();

        // Remove garbage versions
        for ptr in to_remove {
            if let Some(version) = self.versions.remove(&ptr) {
                stats.versions_removed += 1;
                // Estimate bytes: rough size of the value data
                stats.bytes_freed += version
                    .data
                    .iter()
                    .map(|v| v.size_estimate())
                    .sum::<usize>();
            }
        }

        stats
    }

    /// Get all version pointers (for iteration/inspection)
    pub fn pointers(&self) -> impl Iterator<Item = &VersionPtr> {
        self.versions.keys()
    }
}

/// Version chain helper for navigating versions
pub struct VersionChain<'a> {
    version_store: &'a VersionStore,
    current: Option<VersionPtr>,
}

impl<'a> VersionChain<'a> {
    /// Create a new version chain starting from a pointer
    pub fn new(version_store: &'a VersionStore, start: Option<VersionPtr>) -> Self {
        VersionChain {
            version_store,
            current: start,
        }
    }

    /// Find a version visible to the given snapshot
    pub fn find_visible(&mut self, snapshot: &Snapshot) -> Option<&'a OldVersion> {
        while let Some(ptr) = self.current {
            if let Some(version) = self.version_store.get(ptr) {
                if snapshot.is_visible(version.created_by, version.deleted_by) {
                    return Some(version);
                }
                self.current = version.prev_version;
            } else {
                break;
            }
        }
        None
    }
}

impl<'a> Iterator for VersionChain<'a> {
    type Item = &'a OldVersion;

    fn next(&mut self) -> Option<Self::Item> {
        let ptr = self.current?;
        let version = self.version_store.get(ptr)?;
        self.current = version.prev_version;
        Some(version)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_versioned_value_serialization() {
        let value = VersionedValue {
            current: vec![Value::Integer(42), Value::Text("hello".into())],
            created_by: TransactionId(10),
            deleted_by: Some(TransactionId(20)),
            version_chain: Some(VersionPtr(5)),
        };

        let serialized = value.serialize();
        let deserialized = VersionedValue::deserialize(&serialized).unwrap();

        assert_eq!(value.created_by, deserialized.created_by);
        assert_eq!(value.deleted_by, deserialized.deleted_by);
        assert_eq!(value.version_chain, deserialized.version_chain);
        assert_eq!(value.current.len(), deserialized.current.len());
    }

    #[test]
    fn test_version_store() {
        let mut store = VersionStore::new();

        let v1 = OldVersion {
            data: vec![Value::Integer(1)],
            created_by: TransactionId(1),
            deleted_by: None,
            prev_version: None,
        };

        let ptr1 = store.insert(v1);
        assert!(!ptr1.is_null());
        assert!(store.get(ptr1).is_some());
    }

    #[test]
    fn test_version_chain() {
        let mut store = VersionStore::new();

        // Create a chain of versions
        let v1 = OldVersion {
            data: vec![Value::Integer(1)],
            created_by: TransactionId(1),
            deleted_by: Some(TransactionId(2)),
            prev_version: None,
        };
        let ptr1 = store.insert(v1);

        let v2 = OldVersion {
            data: vec![Value::Integer(2)],
            created_by: TransactionId(2),
            deleted_by: Some(TransactionId(3)),
            prev_version: Some(ptr1),
        };
        let ptr2 = store.insert(v2);

        // Iterate through chain
        let chain: Vec<_> = VersionChain::new(&store, Some(ptr2)).collect();
        assert_eq!(chain.len(), 2);
        assert_eq!(chain[0].data[0], Value::Integer(2));
        assert_eq!(chain[1].data[0], Value::Integer(1));
    }

    #[test]
    fn test_gc_removes_old_deleted_versions() {
        let mut store = VersionStore::new();

        // Insert version deleted by transaction 5
        let v1 = OldVersion {
            data: vec![Value::Integer(100)],
            created_by: TransactionId(1),
            deleted_by: Some(TransactionId(5)),
            prev_version: None,
        };
        store.insert(v1);

        // Insert version deleted by transaction 10
        let v2 = OldVersion {
            data: vec![Value::Integer(200)],
            created_by: TransactionId(2),
            deleted_by: Some(TransactionId(10)),
            prev_version: None,
        };
        store.insert(v2);

        // Insert version NOT deleted (still current)
        let v3 = OldVersion {
            data: vec![Value::Integer(300)],
            created_by: TransactionId(3),
            deleted_by: None,
            prev_version: None,
        };
        store.insert(v3);

        assert_eq!(store.len(), 3);

        // GC with oldest active = 7
        // Should remove v1 (deleted by 5 < 7)
        // Should keep v2 (deleted by 10 >= 7)
        // Should keep v3 (not deleted)
        let stats = store.gc(Some(TransactionId(7)));

        assert_eq!(stats.versions_removed, 1);
        assert_eq!(store.len(), 2);
    }

    #[test]
    fn test_gc_with_no_active_transactions() {
        let mut store = VersionStore::new();

        // Insert deleted versions
        let v1 = OldVersion {
            data: vec![Value::Integer(100)],
            created_by: TransactionId(1),
            deleted_by: Some(TransactionId(5)),
            prev_version: None,
        };
        store.insert(v1);

        let v2 = OldVersion {
            data: vec![Value::Integer(200)],
            created_by: TransactionId(2),
            deleted_by: Some(TransactionId(10)),
            prev_version: None,
        };
        store.insert(v2);

        // Version not deleted
        let v3 = OldVersion {
            data: vec![Value::Integer(300)],
            created_by: TransactionId(3),
            deleted_by: None,
            prev_version: None,
        };
        store.insert(v3);

        assert_eq!(store.len(), 3);

        // GC with no active transactions (oldest = None)
        // Should remove all deleted versions
        let stats = store.gc(None);

        assert_eq!(stats.versions_removed, 2);
        assert_eq!(store.len(), 1); // Only v3 remains
    }

    #[test]
    fn test_gc_bytes_freed() {
        let mut store = VersionStore::new();

        // Insert version with substantial data
        let v1 = OldVersion {
            data: vec![Value::Text("hello world".to_string())],
            created_by: TransactionId(1),
            deleted_by: Some(TransactionId(5)),
            prev_version: None,
        };
        store.insert(v1);

        let stats = store.gc(Some(TransactionId(10)));

        assert_eq!(stats.versions_removed, 1);
        assert!(stats.bytes_freed > 0);
    }
}