storage/

delta.rs

//! Delta Encoding for Vector Updates
//!
//! This module provides efficient storage of vector updates by encoding only
//! the differences (deltas) between consecutive vector versions. This is useful
//! for frequently updated vectors where changes are typically small.
//!
//! Key features:
//! - Sparse delta encoding (only changed components stored)
//! - Run-length encoding for contiguous changes
//! - Configurable compression thresholds
//! - Version tracking and reconstruction
//! - Automatic compaction of delta chains

use common::{NamespaceId, Vector, VectorId};
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};

// ============================================================================
// Delta Encoding Configuration
// ============================================================================

/// Configuration for delta encoding
#[derive(Debug, Clone)]
pub struct DeltaConfig {
    /// Maximum number of deltas to chain before forcing a full snapshot
    pub max_delta_chain: usize,
    /// Threshold for sparse encoding (ratio of changed components)
    /// Below this threshold, use sparse encoding; above, store full vector
    pub sparse_threshold: f32,
    /// Minimum change magnitude to consider a component changed
    pub epsilon: f32,
    /// Enable run-length encoding for contiguous changes
    pub enable_rle: bool,
    /// Maximum memory for delta storage per namespace (bytes)
    pub max_memory_per_namespace: usize,
    /// Enable automatic compaction
    pub auto_compact: bool,
    /// Compact when delta chain reaches this length
    pub compact_threshold: usize,
}

impl Default for DeltaConfig {
    fn default() -> Self {
        Self {
            max_delta_chain: 10,
            sparse_threshold: 0.5, // If >50% components changed, store full vector
            epsilon: 1e-7,
            enable_rle: true,
            max_memory_per_namespace: 100 * 1024 * 1024, // 100MB
            auto_compact: true,
            compact_threshold: 5,
        }
    }
}

impl DeltaConfig {
    /// Create a new config
    pub fn new() -> Self {
        Self::default()
    }

    /// Set max delta chain length
    pub fn with_max_chain(mut self, max: usize) -> Self {
        self.max_delta_chain = max;
        self
    }

    /// Set sparse threshold
    pub fn with_sparse_threshold(mut self, threshold: f32) -> Self {
        self.sparse_threshold = threshold.clamp(0.0, 1.0);
        self
    }

    /// Set epsilon for change detection
    pub fn with_epsilon(mut self, epsilon: f32) -> Self {
        self.epsilon = epsilon.abs();
        self
    }

    /// Disable RLE
    pub fn without_rle(mut self) -> Self {
        self.enable_rle = false;
        self
    }

    /// Disable auto compaction
    pub fn without_auto_compact(mut self) -> Self {
        self.auto_compact = false;
        self
    }
}

// ============================================================================
// Delta Types
// ============================================================================

/// A single component change
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComponentChange {
    /// Index of the changed component
    pub index: u32,
    /// New value
    pub value: f32,
}

/// Run-length encoded change (contiguous range)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RleChange {
    /// Starting index
    pub start: u32,
    /// Values for contiguous indices
    pub values: Vec<f32>,
}

/// Encoding type for a delta
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DeltaEncoding {
    /// Sparse encoding: list of individual component changes
    Sparse(Vec<ComponentChange>),
    /// Run-length encoded: contiguous ranges
    Rle(Vec<RleChange>),
    /// Full vector (when too many changes)
    Full(Vec<f32>),
}

/// A delta representing changes between vector versions
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VectorDelta {
    /// Version number this delta leads to
    pub version: u64,
    /// Previous version (base for this delta)
    pub base_version: u64,
    /// Timestamp when delta was created
    pub timestamp: u64,
    /// The encoded changes
    pub encoding: DeltaEncoding,
    /// Size in bytes (estimated)
    pub size_bytes: usize,
}

impl VectorDelta {
    /// Calculate the number of changed components
    pub fn change_count(&self) -> usize {
        match &self.encoding {
            DeltaEncoding::Sparse(changes) => changes.len(),
            DeltaEncoding::Rle(runs) => runs.iter().map(|r| r.values.len()).sum(),
            DeltaEncoding::Full(values) => values.len(),
        }
    }

    /// Check if this is a full snapshot (not a delta)
    pub fn is_full(&self) -> bool {
        matches!(self.encoding, DeltaEncoding::Full(_))
    }
}

/// A versioned vector with its delta history
#[derive(Debug, Clone)]
pub struct VersionedVector {
    /// Vector ID
    pub id: VectorId,
    /// Current version number
    pub current_version: u64,
    /// Base snapshot (full vector at some version)
    pub base_snapshot: Vec<f32>,
    /// Version of the base snapshot
    pub base_version: u64,
    /// Chain of deltas from base to current
    pub deltas: VecDeque<VectorDelta>,
    /// Metadata (preserved across versions)
    pub metadata: Option<serde_json::Value>,
    /// TTL fields
    pub ttl_seconds: Option<u64>,
    pub expires_at: Option<u64>,
}

impl VersionedVector {
    /// Create a new versioned vector
    pub fn new(vector: &Vector) -> Self {
        Self {
            id: vector.id.clone(),
            current_version: 1,
            base_snapshot: vector.values.clone(),
            base_version: 1,
            deltas: VecDeque::new(),
            metadata: vector.metadata.clone(),
            ttl_seconds: vector.ttl_seconds,
            expires_at: vector.expires_at,
        }
    }

    /// Reconstruct the current vector values
    pub fn reconstruct(&self) -> Vec<f32> {
        let mut values = self.base_snapshot.clone();

        for delta in &self.deltas {
            apply_delta(&mut values, delta);
        }

        values
    }

    /// Reconstruct vector at a specific version
    pub fn reconstruct_at_version(&self, version: u64) -> Option<Vec<f32>> {
        if version < self.base_version {
            return None; // Version not available
        }

        if version == self.base_version {
            return Some(self.base_snapshot.clone());
        }

        let mut values = self.base_snapshot.clone();

        for delta in &self.deltas {
            if delta.version > version {
                break;
            }
            apply_delta(&mut values, delta);
        }

        Some(values)
    }

    /// Get the total size in bytes (estimated)
    pub fn size_bytes(&self) -> usize {
        let base_size = self.base_snapshot.len() * 4;
        let delta_size: usize = self.deltas.iter().map(|d| d.size_bytes).sum();
        let metadata_size = self
            .metadata
            .as_ref()
            .map(|m| serde_json::to_string(m).map(|s| s.len()).unwrap_or(0))
            .unwrap_or(0);

        base_size + delta_size + metadata_size + 64 // overhead
    }

    /// Convert to a Vector
    pub fn to_vector(&self) -> Vector {
        Vector {
            id: self.id.clone(),
            values: self.reconstruct(),
            metadata: self.metadata.clone(),
            ttl_seconds: self.ttl_seconds,
            expires_at: self.expires_at,
        }
    }

    /// Get delta chain length
    pub fn delta_chain_length(&self) -> usize {
        self.deltas.len()
    }
}

// ============================================================================
// Delta Operations
// ============================================================================

/// Apply a delta to a vector, returning the new values
/// This handles dimension changes by resizing when needed
fn apply_delta(values: &mut Vec<f32>, delta: &VectorDelta) {
    match &delta.encoding {
        DeltaEncoding::Sparse(changes) => {
            for change in changes {
                let idx = change.index as usize;
                // Extend if index is beyond current length
                if idx >= values.len() {
                    values.resize(idx + 1, 0.0);
                }
                values[idx] = change.value;
            }
        }
        DeltaEncoding::Rle(runs) => {
            for run in runs {
                let start = run.start as usize;
                for (i, &value) in run.values.iter().enumerate() {
                    let idx = start + i;
                    // Extend if index is beyond current length
                    if idx >= values.len() {
                        values.resize(idx + 1, 0.0);
                    }
                    values[idx] = value;
                }
            }
        }
        DeltaEncoding::Full(new_values) => {
            // Full replacement - just clone the new values
            values.clear();
            values.extend_from_slice(new_values);
        }
    }
}

/// Compute delta between old and new vectors
pub fn compute_delta(
    old_values: &[f32],
    new_values: &[f32],
    config: &DeltaConfig,
    base_version: u64,
    new_version: u64,
) -> VectorDelta {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();

    // If dimensions changed, always use full encoding to avoid data corruption
    if old_values.len() != new_values.len() {
        let size_bytes = estimate_encoding_size(&DeltaEncoding::Full(new_values.to_vec()));
        return VectorDelta {
            version: new_version,
            base_version,
            timestamp: now,
            encoding: DeltaEncoding::Full(new_values.to_vec()),
            size_bytes,
        };
    }

    // Find changed components
    let mut changes: Vec<(usize, f32)> = Vec::new();
    let len = old_values.len();

    for i in 0..len {
        if (old_values[i] - new_values[i]).abs() > config.epsilon {
            changes.push((i, new_values[i]));
        }
    }

    let change_ratio = changes.len() as f32 / new_values.len().max(1) as f32;

    // Decide encoding
    let encoding = if change_ratio > config.sparse_threshold {
        // Too many changes, store full vector
        DeltaEncoding::Full(new_values.to_vec())
    } else if config.enable_rle && changes.len() > 2 {
        // Try RLE encoding
        encode_rle(&changes)
    } else {
        // Sparse encoding
        DeltaEncoding::Sparse(
            changes
                .into_iter()
                .map(|(idx, val)| ComponentChange {
                    index: idx as u32,
                    value: val,
                })
                .collect(),
        )
    };

    let size_bytes = estimate_encoding_size(&encoding);

    VectorDelta {
        version: new_version,
        base_version,
        timestamp: now,
        encoding,
        size_bytes,
    }
}

/// Encode changes using run-length encoding
fn encode_rle(changes: &[(usize, f32)]) -> DeltaEncoding {
    if changes.is_empty() {
        return DeltaEncoding::Sparse(vec![]);
    }

    let mut runs: Vec<RleChange> = Vec::new();
    let mut current_run: Option<RleChange> = None;

    for &(idx, value) in changes {
        match &mut current_run {
            Some(run) => {
                let expected_idx = run.start as usize + run.values.len();
                if idx == expected_idx {
                    // Extend current run
                    run.values.push(value);
                } else {
                    // End current run, start new one
                    runs.push(current_run.take().unwrap());
                    current_run = Some(RleChange {
                        start: idx as u32,
                        values: vec![value],
                    });
                }
            }
            None => {
                current_run = Some(RleChange {
                    start: idx as u32,
                    values: vec![value],
                });
            }
        }
    }

    if let Some(run) = current_run {
        runs.push(run);
    }

    // Decide if RLE is worth it
    let rle_size: usize = runs.iter().map(|r| 4 + r.values.len() * 4).sum();
    let sparse_size = changes.len() * 8; // index + value

    if rle_size < sparse_size {
        DeltaEncoding::Rle(runs)
    } else {
        DeltaEncoding::Sparse(
            changes
                .iter()
                .map(|&(idx, val)| ComponentChange {
                    index: idx as u32,
                    value: val,
                })
                .collect(),
        )
    }
}

/// Estimate size of encoding in bytes
fn estimate_encoding_size(encoding: &DeltaEncoding) -> usize {
    match encoding {
        DeltaEncoding::Sparse(changes) => changes.len() * 8 + 16,
        DeltaEncoding::Rle(runs) => runs.iter().map(|r| 8 + r.values.len() * 4).sum::<usize>() + 16,
        DeltaEncoding::Full(values) => values.len() * 4 + 16,
    }
}

// ============================================================================
// Delta Store
// ============================================================================

/// Statistics for delta storage
#[derive(Debug, Clone, Default)]
pub struct DeltaStats {
    /// Total versioned vectors
    pub total_vectors: u64,
    /// Total deltas stored
    pub total_deltas: u64,
    /// Total base snapshots
    pub total_snapshots: u64,
    /// Estimated memory usage in bytes
    pub memory_bytes: u64,
    /// Average delta chain length
    pub avg_chain_length: f64,
    /// Compression ratio (original / compressed)
    pub compression_ratio: f64,
    /// Number of compactions performed
    pub compactions: u64,
}

/// Delta store for a namespace
pub struct NamespaceDeltaStore {
    /// Versioned vectors
    vectors: RwLock<HashMap<VectorId, VersionedVector>>,
    /// Configuration
    config: DeltaConfig,
    /// Statistics
    stats: AtomicDeltaStats,
}

/// Atomic statistics
struct AtomicDeltaStats {
    total_vectors: AtomicU64,
    total_deltas: AtomicU64,
    compactions: AtomicU64,
}

impl AtomicDeltaStats {
    fn new() -> Self {
        Self {
            total_vectors: AtomicU64::new(0),
            total_deltas: AtomicU64::new(0),
            compactions: AtomicU64::new(0),
        }
    }
}

impl NamespaceDeltaStore {
    /// Create a new delta store
    pub fn new(config: DeltaConfig) -> Self {
        Self {
            vectors: RwLock::new(HashMap::new()),
            config,
            stats: AtomicDeltaStats::new(),
        }
    }

    /// Insert or update a vector
    pub fn upsert(&self, vector: &Vector) -> UpsertResult {
        let mut vectors = self.vectors.write();

        if let Some(existing) = vectors.get_mut(&vector.id) {
            // Compute delta
            let current_values = existing.reconstruct();
            let new_version = existing.current_version + 1;

            let delta = compute_delta(
                &current_values,
                &vector.values,
                &self.config,
                existing.current_version,
                new_version,
            );

            let delta_size = delta.size_bytes;
            let is_full = delta.is_full();

            // Add delta
            existing.deltas.push_back(delta);
            existing.current_version = new_version;
            existing.metadata = vector.metadata.clone();
            existing.ttl_seconds = vector.ttl_seconds;
            existing.expires_at = vector.expires_at;

            self.stats.total_deltas.fetch_add(1, Ordering::SeqCst);

            // Check if compaction needed
            let should_compact = self.config.auto_compact
                && existing.delta_chain_length() >= self.config.compact_threshold;

            if should_compact {
                self.compact_vector(existing);
            }

            UpsertResult {
                is_new: false,
                version: new_version,
                delta_size,
                used_full_encoding: is_full,
                compacted: should_compact,
            }
        } else {
            // New vector
            let versioned = VersionedVector::new(vector);
            vectors.insert(vector.id.clone(), versioned);

            self.stats.total_vectors.fetch_add(1, Ordering::SeqCst);

            UpsertResult {
                is_new: true,
                version: 1,
                delta_size: 0,
                used_full_encoding: false,
                compacted: false,
            }
        }
    }

    /// Get a vector by ID
    pub fn get(&self, id: &VectorId) -> Option<Vector> {
        self.vectors.read().get(id).map(|v| v.to_vector())
    }

    /// Get a vector at a specific version
    pub fn get_at_version(&self, id: &VectorId, version: u64) -> Option<Vector> {
        let vectors = self.vectors.read();
        let versioned = vectors.get(id)?;

        let values = versioned.reconstruct_at_version(version)?;

        Some(Vector {
            id: versioned.id.clone(),
            values,
            metadata: versioned.metadata.clone(),
            ttl_seconds: versioned.ttl_seconds,
            expires_at: versioned.expires_at,
        })
    }

    /// Get version history for a vector
    pub fn get_version_info(&self, id: &VectorId) -> Option<VersionInfo> {
        let vectors = self.vectors.read();
        let versioned = vectors.get(id)?;

        Some(VersionInfo {
            id: versioned.id.clone(),
            current_version: versioned.current_version,
            base_version: versioned.base_version,
            delta_count: versioned.deltas.len(),
            size_bytes: versioned.size_bytes(),
        })
    }

    /// Delete a vector
    pub fn delete(&self, id: &VectorId) -> bool {
        let removed = self.vectors.write().remove(id).is_some();
        if removed {
            self.stats.total_vectors.fetch_sub(1, Ordering::SeqCst);
        }
        removed
    }

    /// Get all vectors
    pub fn get_all(&self) -> Vec<Vector> {
        self.vectors
            .read()
            .values()
            .map(|v| v.to_vector())
            .collect()
    }

    /// Compact a vector's delta chain (collapse to new base snapshot)
    fn compact_vector(&self, versioned: &mut VersionedVector) {
        // Reconstruct current state
        let current_values = versioned.reconstruct();

        // Replace with new base
        versioned.base_snapshot = current_values;
        versioned.base_version = versioned.current_version;
        versioned.deltas.clear();

        self.stats.compactions.fetch_add(1, Ordering::SeqCst);
    }

    /// Manually compact a specific vector
    pub fn compact(&self, id: &VectorId) -> bool {
        let mut vectors = self.vectors.write();
        if let Some(versioned) = vectors.get_mut(id) {
            self.compact_vector(versioned);
            true
        } else {
            false
        }
    }

    /// Compact all vectors in the store
    pub fn compact_all(&self) -> usize {
        let mut vectors = self.vectors.write();
        let mut count = 0;

        for versioned in vectors.values_mut() {
            if !versioned.deltas.is_empty() {
                self.compact_vector(versioned);
                count += 1;
            }
        }

        count
    }

    /// Get statistics
    pub fn stats(&self) -> DeltaStats {
        let vectors = self.vectors.read();

        let total_vectors = vectors.len() as u64;
        let total_deltas: usize = vectors.values().map(|v| v.deltas.len()).sum();
        let memory_bytes: usize = vectors.values().map(|v| v.size_bytes()).sum();

        let avg_chain = if total_vectors > 0 {
            total_deltas as f64 / total_vectors as f64
        } else {
            0.0
        };

        // Calculate compression ratio
        let original_size: usize = vectors
            .values()
            .map(|v| v.reconstruct().len() * 4 * (v.deltas.len() + 1))
            .sum();

        let compression_ratio = if memory_bytes > 0 {
            original_size as f64 / memory_bytes as f64
        } else {
            1.0
        };

        DeltaStats {
            total_vectors,
            total_deltas: total_deltas as u64,
            total_snapshots: total_vectors,
            memory_bytes: memory_bytes as u64,
            avg_chain_length: avg_chain,
            compression_ratio,
            compactions: self.stats.compactions.load(Ordering::SeqCst),
        }
    }

    /// Get vector count
    pub fn count(&self) -> usize {
        self.vectors.read().len()
    }

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

    /// Clear all vectors
    pub fn clear(&self) {
        self.vectors.write().clear();
        self.stats.total_vectors.store(0, Ordering::SeqCst);
        self.stats.total_deltas.store(0, Ordering::SeqCst);
    }
}

/// Result of an upsert operation
#[derive(Debug, Clone)]
pub struct UpsertResult {
    /// Whether this was a new vector
    pub is_new: bool,
    /// New version number
    pub version: u64,
    /// Size of delta (0 for new vectors)
    pub delta_size: usize,
    /// Whether full encoding was used (not sparse/RLE)
    pub used_full_encoding: bool,
    /// Whether compaction was triggered
    pub compacted: bool,
}

/// Version information for a vector
#[derive(Debug, Clone)]
pub struct VersionInfo {
    /// Vector ID
    pub id: VectorId,
    /// Current version number
    pub current_version: u64,
    /// Base snapshot version
    pub base_version: u64,
    /// Number of deltas in chain
    pub delta_count: usize,
    /// Total size in bytes
    pub size_bytes: usize,
}

// ============================================================================
// Delta Store Manager (Multi-Namespace)
// ============================================================================

/// Manager for delta stores across namespaces
pub struct DeltaStoreManager {
    stores: RwLock<HashMap<NamespaceId, NamespaceDeltaStore>>,
    config: DeltaConfig,
}

impl DeltaStoreManager {
    /// Create a new manager
    pub fn new(config: DeltaConfig) -> Self {
        Self {
            stores: RwLock::new(HashMap::new()),
            config,
        }
    }

    /// Create with default config
    pub fn with_defaults() -> Self {
        Self::new(DeltaConfig::default())
    }

    /// Get or create a namespace store
    pub fn get_or_create(&self, namespace: &NamespaceId) -> bool {
        let mut stores = self.stores.write();
        if !stores.contains_key(namespace) {
            stores.insert(
                namespace.clone(),
                NamespaceDeltaStore::new(self.config.clone()),
            );
            true
        } else {
            false
        }
    }

    /// Upsert vectors into a namespace
    pub fn upsert(&self, namespace: &NamespaceId, vectors: &[Vector]) -> Vec<UpsertResult> {
        self.get_or_create(namespace);

        let stores = self.stores.read();
        if let Some(store) = stores.get(namespace) {
            vectors.iter().map(|v| store.upsert(v)).collect()
        } else {
            vec![]
        }
    }

    /// Get a vector
    pub fn get(&self, namespace: &NamespaceId, id: &VectorId) -> Option<Vector> {
        self.stores.read().get(namespace)?.get(id)
    }

    /// Get all vectors in namespace
    pub fn get_all(&self, namespace: &NamespaceId) -> Vec<Vector> {
        self.stores
            .read()
            .get(namespace)
            .map(|s| s.get_all())
            .unwrap_or_default()
    }

    /// Delete a vector
    pub fn delete(&self, namespace: &NamespaceId, id: &VectorId) -> bool {
        self.stores
            .read()
            .get(namespace)
            .map(|s| s.delete(id))
            .unwrap_or(false)
    }

    /// Delete a namespace
    pub fn delete_namespace(&self, namespace: &NamespaceId) -> bool {
        self.stores.write().remove(namespace).is_some()
    }

    /// Compact all vectors in a namespace
    pub fn compact_namespace(&self, namespace: &NamespaceId) -> usize {
        self.stores
            .read()
            .get(namespace)
            .map(|s| s.compact_all())
            .unwrap_or(0)
    }

    /// Get stats for a namespace
    pub fn namespace_stats(&self, namespace: &NamespaceId) -> Option<DeltaStats> {
        self.stores.read().get(namespace).map(|s| s.stats())
    }

    /// Get combined stats
    pub fn stats(&self) -> DeltaStats {
        let stores = self.stores.read();

        let mut combined = DeltaStats::default();

        for store in stores.values() {
            let s = store.stats();
            combined.total_vectors += s.total_vectors;
            combined.total_deltas += s.total_deltas;
            combined.total_snapshots += s.total_snapshots;
            combined.memory_bytes += s.memory_bytes;
            combined.compactions += s.compactions;
        }

        if combined.total_vectors > 0 {
            combined.avg_chain_length =
                combined.total_deltas as f64 / combined.total_vectors as f64;
        }

        combined
    }

    /// List namespaces
    pub fn list_namespaces(&self) -> Vec<NamespaceId> {
        self.stores.read().keys().cloned().collect()
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    fn make_vector(id: &str, values: Vec<f32>) -> Vector {
        Vector {
            id: id.to_string(),
            values,
            metadata: None,
            ttl_seconds: None,
            expires_at: None,
        }
    }

    #[test]
    fn test_delta_config_builder() {
        let config = DeltaConfig::new()
            .with_max_chain(20)
            .with_sparse_threshold(0.3)
            .with_epsilon(1e-5)
            .without_rle()
            .without_auto_compact();

        assert_eq!(config.max_delta_chain, 20);
        assert!((config.sparse_threshold - 0.3).abs() < 0.001);
        assert!((config.epsilon - 1e-5).abs() < 1e-10);
        assert!(!config.enable_rle);
        assert!(!config.auto_compact);
    }

    #[test]
    fn test_compute_delta_sparse() {
        let old = vec![1.0, 2.0, 3.0, 4.0, 5.0];
        let new = vec![1.0, 2.5, 3.0, 4.0, 5.5]; // Changed indices 1 and 4

        let config = DeltaConfig::default();
        let delta = compute_delta(&old, &new, &config, 1, 2);

        assert_eq!(delta.version, 2);
        assert_eq!(delta.base_version, 1);

        if let DeltaEncoding::Sparse(changes) = delta.encoding {
            assert_eq!(changes.len(), 2);
            assert!(changes
                .iter()
                .any(|c| c.index == 1 && (c.value - 2.5).abs() < 0.001));
            assert!(changes
                .iter()
                .any(|c| c.index == 4 && (c.value - 5.5).abs() < 0.001));
        } else {
            panic!("Expected sparse encoding");
        }
    }

    #[test]
    fn test_compute_delta_rle() {
        let old = vec![0.0; 10];
        let new = vec![0.0, 1.0, 1.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0]; // Run of 4 changes

        let config = DeltaConfig::default();
        let delta = compute_delta(&old, &new, &config, 1, 2);

        // Should use RLE for contiguous changes
        if let DeltaEncoding::Rle(runs) = delta.encoding {
            assert_eq!(runs.len(), 1);
            assert_eq!(runs[0].start, 1);
            assert_eq!(runs[0].values.len(), 4);
        } else if let DeltaEncoding::Sparse(changes) = delta.encoding {
            // RLE wasn't more efficient, that's ok
            assert_eq!(changes.len(), 4);
        } else {
            panic!("Expected RLE or Sparse encoding");
        }
    }

    #[test]
    fn test_compute_delta_full() {
        let old = vec![0.0; 10];
        let new = vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]; // All changed

        let config = DeltaConfig::default();
        let delta = compute_delta(&old, &new, &config, 1, 2);

        // Should use full encoding
        assert!(delta.is_full());

        if let DeltaEncoding::Full(values) = delta.encoding {
            assert_eq!(values, new);
        } else {
            panic!("Expected full encoding");
        }
    }

    #[test]
    fn test_apply_delta_sparse() {
        let mut values = vec![1.0, 2.0, 3.0, 4.0, 5.0];
        let delta = VectorDelta {
            version: 2,
            base_version: 1,
            timestamp: 0,
            encoding: DeltaEncoding::Sparse(vec![
                ComponentChange {
                    index: 1,
                    value: 10.0,
                },
                ComponentChange {
                    index: 3,
                    value: 20.0,
                },
            ]),
            size_bytes: 0,
        };

        apply_delta(&mut values, &delta);

        assert_eq!(values, vec![1.0, 10.0, 3.0, 20.0, 5.0]);
    }

    #[test]
    fn test_apply_delta_rle() {
        let mut values = vec![0.0; 10];
        let delta = VectorDelta {
            version: 2,
            base_version: 1,
            timestamp: 0,
            encoding: DeltaEncoding::Rle(vec![RleChange {
                start: 2,
                values: vec![1.0, 2.0, 3.0],
            }]),
            size_bytes: 0,
        };

        apply_delta(&mut values, &delta);

        assert_eq!(values[2], 1.0);
        assert_eq!(values[3], 2.0);
        assert_eq!(values[4], 3.0);
    }

    #[test]
    fn test_versioned_vector() {
        let v = make_vector("v1", vec![1.0, 2.0, 3.0]);
        let versioned = VersionedVector::new(&v);

        assert_eq!(versioned.current_version, 1);
        assert_eq!(versioned.base_version, 1);
        assert_eq!(versioned.deltas.len(), 0);
        assert_eq!(versioned.reconstruct(), vec![1.0, 2.0, 3.0]);
    }

    #[test]
    fn test_namespace_delta_store_upsert_new() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default());

        let v = make_vector("v1", vec![1.0, 2.0, 3.0]);
        let result = store.upsert(&v);

        assert!(result.is_new);
        assert_eq!(result.version, 1);
        assert_eq!(result.delta_size, 0);
    }

    #[test]
    fn test_namespace_delta_store_upsert_update() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default());

        // Insert
        let v1 = make_vector("v1", vec![1.0, 2.0, 3.0]);
        store.upsert(&v1);

        // Update
        let v2 = make_vector("v1", vec![1.0, 5.0, 3.0]);
        let result = store.upsert(&v2);

        assert!(!result.is_new);
        assert_eq!(result.version, 2);
        assert!(result.delta_size > 0);

        // Verify reconstruction
        let retrieved = store.get(&"v1".to_string()).unwrap();
        assert_eq!(retrieved.values, vec![1.0, 5.0, 3.0]);
    }

    #[test]
    fn test_namespace_delta_store_version_history() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default().without_auto_compact());

        // Insert and update
        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0])); // v1
        store.upsert(&make_vector("v1", vec![1.0, 5.0, 3.0])); // v2
        store.upsert(&make_vector("v1", vec![1.0, 5.0, 10.0])); // v3

        // Get current
        let current = store.get(&"v1".to_string()).unwrap();
        assert_eq!(current.values, vec![1.0, 5.0, 10.0]);

        // Get at version 2
        let v2 = store.get_at_version(&"v1".to_string(), 2).unwrap();
        assert_eq!(v2.values, vec![1.0, 5.0, 3.0]);

        // Get at version 1
        let v1 = store.get_at_version(&"v1".to_string(), 1).unwrap();
        assert_eq!(v1.values, vec![1.0, 2.0, 3.0]);
    }

    #[test]
    fn test_namespace_delta_store_compact() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default().without_auto_compact());

        // Create delta chain
        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));
        store.upsert(&make_vector("v1", vec![1.0, 5.0, 3.0]));
        store.upsert(&make_vector("v1", vec![1.0, 5.0, 10.0]));

        // Verify chain exists
        let info_before = store.get_version_info(&"v1".to_string()).unwrap();
        assert_eq!(info_before.delta_count, 2);

        // Compact
        store.compact(&"v1".to_string());

        // Verify chain is gone
        let info_after = store.get_version_info(&"v1".to_string()).unwrap();
        assert_eq!(info_after.delta_count, 0);

        // Values should still be correct
        let v = store.get(&"v1".to_string()).unwrap();
        assert_eq!(v.values, vec![1.0, 5.0, 10.0]);
    }

    #[test]
    fn test_namespace_delta_store_auto_compact() {
        let config = DeltaConfig::default()
            .with_max_chain(10)
            .with_sparse_threshold(0.9); // Ensure sparse encoding

        // Low compact threshold for testing
        let store = NamespaceDeltaStore::new(DeltaConfig {
            compact_threshold: 3,
            auto_compact: true,
            ..config
        });

        // Create deltas
        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));
        store.upsert(&make_vector("v1", vec![1.1, 2.0, 3.0])); // delta 1
        store.upsert(&make_vector("v1", vec![1.2, 2.0, 3.0])); // delta 2
        let result = store.upsert(&make_vector("v1", vec![1.3, 2.0, 3.0])); // delta 3 -> compact

        // Should have compacted
        assert!(result.compacted);

        let info = store.get_version_info(&"v1".to_string()).unwrap();
        assert_eq!(info.delta_count, 0);
    }

    #[test]
    fn test_namespace_delta_store_delete() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default());

        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));
        assert!(store.delete(&"v1".to_string()));
        assert!(store.get(&"v1".to_string()).is_none());
        assert!(!store.delete(&"v1".to_string())); // Already deleted
    }

    #[test]
    fn test_namespace_delta_store_stats() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default().without_auto_compact());

        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));
        store.upsert(&make_vector("v1", vec![1.0, 5.0, 3.0]));
        store.upsert(&make_vector("v2", vec![4.0, 5.0, 6.0]));

        let stats = store.stats();
        assert_eq!(stats.total_vectors, 2);
        assert_eq!(stats.total_deltas, 1);
        assert!(stats.memory_bytes > 0);
    }

    #[test]
    fn test_delta_store_manager() {
        let manager = DeltaStoreManager::with_defaults();

        // Upsert
        let results = manager.upsert(
            &"ns1".to_string(),
            &[make_vector("v1", vec![1.0, 2.0, 3.0])],
        );
        assert_eq!(results.len(), 1);
        assert!(results[0].is_new);

        // Get
        let v = manager.get(&"ns1".to_string(), &"v1".to_string()).unwrap();
        assert_eq!(v.values, vec![1.0, 2.0, 3.0]);

        // List namespaces
        let namespaces = manager.list_namespaces();
        assert!(namespaces.contains(&"ns1".to_string()));

        // Delete namespace
        assert!(manager.delete_namespace(&"ns1".to_string()));
        assert!(manager.get(&"ns1".to_string(), &"v1".to_string()).is_none());
    }

    #[test]
    fn test_encode_rle_efficiency() {
        // Test that RLE is used for contiguous changes
        let changes: Vec<(usize, f32)> = (0..10).map(|i| (i, i as f32)).collect();

        let encoding = encode_rle(&changes);

        if let DeltaEncoding::Rle(runs) = encoding {
            // Should be one run
            assert_eq!(runs.len(), 1);
            assert_eq!(runs[0].values.len(), 10);
        } else {
            // Sparse might be used if RLE isn't more efficient
            // This is acceptable
        }
    }

    #[test]
    fn test_dimension_change() {
        let store = NamespaceDeltaStore::new(DeltaConfig::default());

        // Insert with 3 dimensions
        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));

        // Update with 5 dimensions
        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0, 4.0, 5.0]));

        let v = store.get(&"v1".to_string()).unwrap();
        assert_eq!(v.values, vec![1.0, 2.0, 3.0, 4.0, 5.0]);
    }

    #[test]
    fn test_epsilon_change_detection() {
        let config = DeltaConfig::default().with_epsilon(0.1);
        let store = NamespaceDeltaStore::new(config);

        store.upsert(&make_vector("v1", vec![1.0, 2.0, 3.0]));

        // Small change below epsilon
        store.upsert(&make_vector("v1", vec![1.05, 2.0, 3.0]));

        // Should not create a delta with changes below epsilon
        let info = store.get_version_info(&"v1".to_string()).unwrap();

        // Value should still be tracked even if no delta
        // The exact behavior depends on implementation details
        assert!(info.current_version >= 1);
    }
}