featherdb_storage/

buffer_pool.rs

//! Buffer pool with pluggable eviction algorithms
//!
//! Supports multiple eviction policies:
//! - Clock: Simple second-chance algorithm (default)
//! - LRU-2: Tracks last 2 access times for scan resistance
//! - LIRS: Low Inter-reference Recency Set for workload adaptation

use crate::eviction::{EvictionPolicy, EvictionPolicyFactory, EvictionPolicyType};
use crate::{FileManager, Page, PageType};
use featherdb_core::{Error, PageId, Result, TransactionId};
use parking_lot::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;

/// Statistics for the buffer pool
#[derive(Debug, Default)]
pub struct BufferPoolStats {
    pub cache_hits: AtomicUsize,
    pub cache_misses: AtomicUsize,
    pub pages_evicted: AtomicUsize,
    pub pages_flushed: AtomicUsize,
}

impl BufferPoolStats {
    /// Calculate hit ratio (0.0 to 1.0)
    pub fn hit_ratio(&self) -> f64 {
        let hits = self.cache_hits.load(Ordering::Relaxed);
        let misses = self.cache_misses.load(Ordering::Relaxed);
        let total = hits + misses;
        if total == 0 {
            1.0
        } else {
            hits as f64 / total as f64
        }
    }

    /// Reset all statistics
    pub fn reset(&self) {
        self.cache_hits.store(0, Ordering::Relaxed);
        self.cache_misses.store(0, Ordering::Relaxed);
        self.pages_evicted.store(0, Ordering::Relaxed);
        self.pages_flushed.store(0, Ordering::Relaxed);
    }

    /// Get a snapshot of current statistics
    pub fn snapshot(&self) -> BufferPoolStatsSnapshot {
        BufferPoolStatsSnapshot {
            cache_hits: self.cache_hits.load(Ordering::Relaxed),
            cache_misses: self.cache_misses.load(Ordering::Relaxed),
            pages_evicted: self.pages_evicted.load(Ordering::Relaxed),
            pages_flushed: self.pages_flushed.load(Ordering::Relaxed),
        }
    }
}

/// Immutable snapshot of buffer pool statistics
#[derive(Debug, Clone)]
pub struct BufferPoolStatsSnapshot {
    pub cache_hits: usize,
    pub cache_misses: usize,
    pub pages_evicted: usize,
    pub pages_flushed: usize,
}

impl BufferPoolStatsSnapshot {
    /// Calculate hit ratio (0.0 to 1.0)
    pub fn hit_ratio(&self) -> f64 {
        let total = self.cache_hits + self.cache_misses;
        if total == 0 {
            1.0
        } else {
            self.cache_hits as f64 / total as f64
        }
    }
}

/// A frame in the buffer pool holding a page
pub struct BufferFrame {
    /// The page ID
    page_id: PageId,
    /// The page data
    page: RwLock<Page>,
    /// Number of active references (pinned when > 0)
    pin_count: AtomicU32,
    /// Recently accessed (for clock algorithm, kept for compatibility)
    reference_bit: AtomicBool,
    /// Modified since last flush
    dirty: AtomicBool,
    /// Transaction that dirtied this page (0 if clean or already committed/flushed)
    /// Uses AtomicU64 for lock-free access; 0 means no transaction owns this page
    dirty_by_txn: AtomicU64,
}

impl BufferFrame {
    fn new(page_id: PageId, page: Page) -> Self {
        BufferFrame {
            page_id,
            page: RwLock::new(page),
            pin_count: AtomicU32::new(0),
            reference_bit: AtomicBool::new(true),
            dirty: AtomicBool::new(false),
            dirty_by_txn: AtomicU64::new(0),
        }
    }

    #[allow(dead_code)]
    pub fn page_id(&self) -> PageId {
        self.page_id
    }

    #[allow(dead_code)]
    pub fn is_dirty(&self) -> bool {
        self.dirty.load(Ordering::Acquire)
    }

    pub fn mark_dirty(&self) {
        self.dirty.store(true, Ordering::Release);
    }

    /// Mark page as dirty with transaction ownership
    pub fn mark_dirty_by(&self, txn_id: TransactionId) {
        self.dirty_by_txn.store(txn_id.0, Ordering::Release);
        self.dirty.store(true, Ordering::Release);
    }

    /// Clear transaction ownership (called after commit/flush)
    pub fn clear_txn_ownership(&self) {
        self.dirty_by_txn.store(0, Ordering::Release);
    }

    /// Check if this page was dirtied by a specific transaction
    #[allow(dead_code)]
    pub fn is_dirty_by(&self, txn_id: TransactionId) -> bool {
        self.dirty_by_txn.load(Ordering::Acquire) == txn_id.0
    }

    /// Get the transaction that owns this dirty page (None if clean or committed)
    pub fn dirty_txn(&self) -> Option<TransactionId> {
        let txn_id = self.dirty_by_txn.load(Ordering::Acquire);
        if txn_id == 0 {
            None
        } else {
            Some(TransactionId(txn_id))
        }
    }

    #[allow(dead_code)]
    pub fn pin_count(&self) -> u32 {
        self.pin_count.load(Ordering::Acquire)
    }
}

/// The buffer pool - caches pages in memory
pub struct BufferPool {
    /// Maximum number of pages to cache
    capacity: usize,
    /// Page cache: page_id -> frame
    frames: RwLock<HashMap<PageId, Arc<BufferFrame>>>,
    /// Eviction policy (behind a mutex for exclusive access during eviction)
    eviction_policy: Mutex<Box<dyn EvictionPolicy>>,
    /// File manager for disk I/O
    file_manager: Arc<FileManager>,
    /// Statistics
    pub stats: BufferPoolStats,
    /// Eviction policy type (for informational purposes)
    policy_type: EvictionPolicyType,
    /// Active (uncommitted) transactions - pages owned by these can't be evicted
    active_txns: RwLock<HashSet<TransactionId>>,
}

impl BufferPool {
    /// Create a new buffer pool with default Clock eviction
    pub fn new(capacity: usize, file_manager: Arc<FileManager>) -> Self {
        Self::with_policy(capacity, file_manager, EvictionPolicyType::Clock)
    }

    /// Create a new buffer pool with specified eviction policy
    pub fn with_policy(
        capacity: usize,
        file_manager: Arc<FileManager>,
        policy_type: EvictionPolicyType,
    ) -> Self {
        BufferPool {
            capacity,
            frames: RwLock::new(HashMap::new()),
            eviction_policy: Mutex::new(policy_type.create(capacity)),
            file_manager,
            stats: BufferPoolStats::default(),
            policy_type,
            active_txns: RwLock::new(HashSet::new()),
        }
    }

    /// Get the eviction policy type
    pub fn eviction_policy_type(&self) -> EvictionPolicyType {
        self.policy_type
    }

    /// Get the eviction policy name
    pub fn eviction_policy_name(&self) -> &'static str {
        self.eviction_policy.lock().name()
    }

    /// Get a page for reading
    pub fn get_page(&self, page_id: PageId) -> Result<PageGuard> {
        self.get_page_internal(page_id, false)
    }

    /// Get a page for writing
    pub fn get_page_for_write(&self, page_id: PageId) -> Result<PageGuardMut> {
        let guard = self.get_page_internal(page_id, true)?;
        Ok(PageGuardMut { inner: guard })
    }

    fn get_page_internal(&self, page_id: PageId, for_write: bool) -> Result<PageGuard> {
        // Get current active transaction for ownership tracking
        let current_txn = if for_write {
            self.get_single_active_txn()
        } else {
            None
        };

        // Fast path: page in cache
        {
            let frames = self.frames.read();
            if let Some(frame) = frames.get(&page_id) {
                frame.pin_count.fetch_add(1, Ordering::AcqRel);
                frame.reference_bit.store(true, Ordering::Relaxed);
                self.stats.cache_hits.fetch_add(1, Ordering::Relaxed);

                // For Clock policy, the reference_bit set above is sufficient.
                // For other policies (LRU-2, LIRS), we need to update their internal state.
                if self.policy_type != EvictionPolicyType::Clock {
                    let mut policy = self.eviction_policy.lock();
                    policy.on_access(page_id);
                }

                if for_write {
                    if let Some(txn_id) = current_txn {
                        frame.mark_dirty_by(txn_id);
                    } else {
                        frame.mark_dirty();
                    }
                }

                return Ok(PageGuard {
                    frame: frame.clone(),
                });
            }
        }

        // Slow path: load from disk
        self.stats.cache_misses.fetch_add(1, Ordering::Relaxed);
        self.load_page_with_txn(page_id, for_write, current_txn)
    }

    fn load_page_with_txn(
        &self,
        page_id: PageId,
        for_write: bool,
        txn_id: Option<TransactionId>,
    ) -> Result<PageGuard> {
        // Evict if at capacity
        {
            let frames = self.frames.read();
            if frames.len() >= self.capacity {
                drop(frames);
                self.evict_one()?;
            }
        }

        // Read from disk
        let page_data = self.file_manager.read_page(page_id)?;
        let page = Page::from_bytes(page_data)?;

        // Verify checksum
        if !page.verify_checksum() {
            return Err(Error::CorruptedPage(page_id));
        }

        // Create frame
        let frame = Arc::new(BufferFrame::new(page_id, page));
        frame.pin_count.fetch_add(1, Ordering::AcqRel);

        if for_write {
            if let Some(tid) = txn_id {
                frame.mark_dirty_by(tid);
            } else {
                frame.mark_dirty();
            }
        }

        // Insert into cache
        {
            let mut frames = self.frames.write();
            let mut policy = self.eviction_policy.lock();

            // Double-check another thread didn't load it
            if let Some(existing) = frames.get(&page_id) {
                existing.pin_count.fetch_add(1, Ordering::AcqRel);
                return Ok(PageGuard {
                    frame: existing.clone(),
                });
            }

            frames.insert(page_id, frame.clone());
            policy.on_load(page_id);
        }

        Ok(PageGuard { frame })
    }

    /// Allocate a new page
    pub fn allocate_page(&self, page_type: PageType) -> Result<(PageId, PageGuardMut)> {
        // Get current active transaction for ownership tracking
        let current_txn = self.get_single_active_txn();

        // Evict if at capacity
        {
            let frames = self.frames.read();
            if frames.len() >= self.capacity {
                drop(frames);
                self.evict_one()?;
            }
        }

        // Extend file
        let page_id = self.file_manager.extend()?;

        // Create new page
        let page = Page::new(page_id, page_type, self.file_manager.page_size());

        // Create frame
        let frame = Arc::new(BufferFrame::new(page_id, page));
        frame.pin_count.fetch_add(1, Ordering::AcqRel);

        // Mark dirty with transaction ownership if we have an active transaction
        if let Some(txn_id) = current_txn {
            frame.mark_dirty_by(txn_id);
        } else {
            frame.mark_dirty();
        }

        // Insert into cache
        {
            let mut frames = self.frames.write();
            let mut policy = self.eviction_policy.lock();

            frames.insert(page_id, frame.clone());
            policy.on_load(page_id);
        }

        Ok((
            page_id,
            PageGuardMut {
                inner: PageGuard { frame },
            },
        ))
    }

    /// Copy a page for modification (copy-on-write)
    pub fn copy_page(&self, page_id: PageId) -> Result<(PageId, PageGuardMut)> {
        // Get current active transaction for ownership tracking
        let current_txn = self.get_single_active_txn();

        // Read the original page
        let original = self.get_page(page_id)?;
        let page_data = {
            let page = original.read();
            page.as_bytes().to_vec()
        };
        drop(original);

        // Allocate new page
        let new_page_id = self.file_manager.extend()?;
        let mut new_page = Page::from_bytes(page_data)?;
        new_page.set_page_id(new_page_id);

        // Create frame
        let frame = Arc::new(BufferFrame::new(new_page_id, new_page));
        frame.pin_count.fetch_add(1, Ordering::AcqRel);

        // Mark dirty with transaction ownership if we have an active transaction
        if let Some(txn_id) = current_txn {
            frame.mark_dirty_by(txn_id);
        } else {
            frame.mark_dirty();
        }

        // Insert into cache
        {
            let mut frames = self.frames.write();
            let mut policy = self.eviction_policy.lock();

            frames.insert(new_page_id, frame.clone());
            policy.on_load(new_page_id);
        }

        Ok((
            new_page_id,
            PageGuardMut {
                inner: PageGuard { frame },
            },
        ))
    }

    /// Evict one page using the configured eviction policy
    fn evict_one(&self) -> Result<()> {
        // Try to find a victim using the eviction policy
        let max_attempts = self.capacity * 2;

        for _ in 0..max_attempts {
            let victim_id = {
                let mut policy = self.eviction_policy.lock();
                policy.select_victim()
            };

            let Some(page_id) = victim_id else {
                // No victim available
                return Err(Error::BufferPoolFull {
                    capacity: self.capacity,
                    pinned: self.count_pinned(),
                });
            };

            // Check if the page can be evicted
            let frame_opt = {
                let frames = self.frames.read();
                frames.get(&page_id).cloned()
            };

            if let Some(frame) = frame_opt {
                // Check if page can be evicted (not pinned, not dirty by active txn)
                if !self.can_evict_frame(&frame) {
                    // Remove from policy and try again
                    let mut policy = self.eviction_policy.lock();
                    policy.remove(page_id);
                    continue;
                }

                // Found victim! Flush if dirty (only committed pages reach here)
                if frame.dirty.load(Ordering::Acquire) {
                    self.flush_frame(&frame)?;
                }

                // Remove from cache and policy
                {
                    let mut frames = self.frames.write();
                    let mut policy = self.eviction_policy.lock();

                    frames.remove(&page_id);
                    policy.remove(page_id);
                }

                self.stats.pages_evicted.fetch_add(1, Ordering::Relaxed);
                return Ok(());
            } else {
                // Page not in cache, remove from policy
                let mut policy = self.eviction_policy.lock();
                policy.remove(page_id);
            }
        }

        Err(Error::BufferPoolFull {
            capacity: self.capacity,
            pinned: self.count_pinned(),
        })
    }

    fn flush_frame(&self, frame: &BufferFrame) -> Result<()> {
        let mut page = frame.page.write();

        // Compute checksum
        page.compute_checksum();

        // Write to disk
        self.file_manager
            .write_page(frame.page_id, page.as_bytes())?;

        frame.dirty.store(false, Ordering::Release);
        self.stats.pages_flushed.fetch_add(1, Ordering::Relaxed);

        Ok(())
    }

    /// Flush all dirty pages
    ///
    /// Skips pages owned by active (uncommitted) transactions to prevent
    /// uncommitted data from being persisted.
    pub fn flush_all(&self) -> Result<()> {
        let frames = self.frames.read();
        let active = self.active_txns.read();

        for frame in frames.values() {
            if frame.dirty.load(Ordering::Acquire) {
                // Skip pages owned by active transactions
                if let Some(txn_id) = frame.dirty_txn() {
                    if active.contains(&txn_id) {
                        continue;
                    }
                }
                self.flush_frame(frame)?;
            }
        }

        Ok(())
    }

    /// Flush a specific page
    pub fn flush_page(&self, page_id: PageId) -> Result<()> {
        let frames = self.frames.read();

        if let Some(frame) = frames.get(&page_id) {
            if frame.dirty.load(Ordering::Acquire) {
                self.flush_frame(frame)?;
            }
        }

        Ok(())
    }

    // ==================== Transaction-Aware Methods ====================

    /// Register a transaction as active (its dirty pages can't be evicted)
    pub fn register_active_txn(&self, txn_id: TransactionId) {
        self.active_txns.write().insert(txn_id);
    }

    /// Unregister a transaction (called on commit or abort)
    pub fn unregister_active_txn(&self, txn_id: TransactionId) {
        self.active_txns.write().remove(&txn_id);
    }

    /// Get a page for write within a transaction context
    ///
    /// This marks the page as dirty and records the transaction that modified it.
    /// Pages owned by uncommitted transactions cannot be evicted.
    pub fn get_page_for_write_in_txn(
        &self,
        page_id: PageId,
        txn_id: TransactionId,
    ) -> Result<PageGuardMut> {
        let guard = self.get_page_internal(page_id, true)?;
        // Mark with transaction ownership
        guard.frame.mark_dirty_by(txn_id);
        Ok(PageGuardMut { inner: guard })
    }

    /// Allocate a new page within a transaction context
    pub fn allocate_page_in_txn(
        &self,
        page_type: PageType,
        txn_id: TransactionId,
    ) -> Result<(PageId, PageGuardMut)> {
        // Evict if at capacity
        {
            let frames = self.frames.read();
            if frames.len() >= self.capacity {
                drop(frames);
                self.evict_one()?;
            }
        }

        // Extend file
        let page_id = self.file_manager.extend()?;

        // Create new page
        let page = Page::new(page_id, page_type, self.file_manager.page_size());

        // Create frame
        let frame = Arc::new(BufferFrame::new(page_id, page));
        frame.pin_count.fetch_add(1, Ordering::AcqRel);
        frame.mark_dirty_by(txn_id);

        // Insert into cache
        {
            let mut frames = self.frames.write();
            let mut policy = self.eviction_policy.lock();

            frames.insert(page_id, frame.clone());
            policy.on_load(page_id);
        }

        Ok((
            page_id,
            PageGuardMut {
                inner: PageGuard { frame },
            },
        ))
    }

    /// Flush only pages dirtied by a specific transaction
    ///
    /// Called during commit to persist the transaction's changes.
    /// After flushing, clears the transaction ownership from the pages.
    pub fn flush_transaction(&self, txn_id: TransactionId) -> Result<()> {
        let frames = self.frames.read();

        for frame in frames.values() {
            if frame.is_dirty_by(txn_id) {
                self.flush_frame(frame)?;
                frame.clear_txn_ownership();
            }
        }

        Ok(())
    }

    /// Discard dirty pages from an aborted transaction
    ///
    /// This clears the dirty flag and transaction ownership for pages
    /// that were modified by the aborted transaction, effectively
    /// discarding their uncommitted changes.
    ///
    /// Note: The in-memory page data still contains the aborted writes,
    /// but since the dirty flag is cleared, they won't be persisted.
    /// When the page is evicted and re-read, the original data is restored.
    pub fn discard_transaction(&self, txn_id: TransactionId) -> Result<()> {
        // First pass: collect pages to discard
        let pages_to_discard: Vec<PageId> = {
            let frames = self.frames.read();
            frames
                .iter()
                .filter(|(_, frame)| frame.is_dirty_by(txn_id))
                .map(|(page_id, _)| *page_id)
                .collect()
        };

        // Second pass: for each page, clear dirty flag and remove if unpinned
        for page_id in pages_to_discard {
            // Get write lock and process the page
            let mut frames_write = self.frames.write();

            if let Some(frame) = frames_write.get(&page_id) {
                // Only process if still owned by this transaction
                if frame.is_dirty_by(txn_id) {
                    // Clear the dirty flag so the page won't be flushed
                    frame.dirty.store(false, Ordering::Release);
                    frame.clear_txn_ownership();

                    // If unpinned, remove from cache so it gets reloaded from disk
                    if frame.pin_count.load(Ordering::Acquire) == 0 {
                        frames_write.remove(&page_id);
                        let mut policy = self.eviction_policy.lock();
                        policy.remove(page_id);
                    }
                }
            }
        }

        Ok(())
    }

    /// Get the single active transaction, if exactly one exists
    ///
    /// This is used to automatically associate dirty pages with transactions
    /// when B-tree operations use the standard get_page_for_write() path.
    /// If there's exactly one active transaction, we can infer that any
    /// write operation belongs to it.
    fn get_single_active_txn(&self) -> Option<TransactionId> {
        let active = self.active_txns.read();
        if active.len() == 1 {
            active.iter().next().copied()
        } else {
            None
        }
    }

    /// Check if a page can be evicted
    ///
    /// A page can be evicted if:
    /// 1. It's not pinned (pin_count == 0)
    /// 2. It's not dirty by an active (uncommitted) transaction
    fn can_evict_frame(&self, frame: &BufferFrame) -> bool {
        // Can't evict pinned pages
        if frame.pin_count.load(Ordering::Acquire) > 0 {
            return false;
        }

        // Can't evict pages dirty by active transactions
        if let Some(txn_id) = frame.dirty_txn() {
            let active = self.active_txns.read();
            if active.contains(&txn_id) {
                return false;
            }
        }

        true
    }

    /// Count the number of currently pinned pages
    fn count_pinned(&self) -> usize {
        let frames = self.frames.read();
        frames
            .values()
            .filter(|f| f.pin_count.load(Ordering::Relaxed) > 0)
            .count()
    }

    /// Get the number of pages in the buffer pool
    pub fn size(&self) -> usize {
        self.frames.read().len()
    }

    /// Get the capacity of the buffer pool
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// Get a snapshot of current statistics
    pub fn stats_snapshot(&self) -> BufferPoolStatsSnapshot {
        self.stats.snapshot()
    }
}

impl Drop for BufferPool {
    fn drop(&mut self) {
        // Best-effort: try to flush remaining dirty pages
        let _ = self.flush_all();
    }
}

/// RAII guard for reading a page
pub struct PageGuard {
    frame: Arc<BufferFrame>,
}

impl PageGuard {
    pub fn read(&self) -> RwLockReadGuard<'_, Page> {
        self.frame.reference_bit.store(true, Ordering::Relaxed);
        self.frame.page.read()
    }

    pub fn page_id(&self) -> PageId {
        self.frame.page_id
    }
}

impl Drop for PageGuard {
    fn drop(&mut self) {
        self.frame.pin_count.fetch_sub(1, Ordering::AcqRel);
    }
}

/// RAII guard for writing a page
pub struct PageGuardMut {
    inner: PageGuard,
}

impl PageGuardMut {
    pub fn read(&self) -> RwLockReadGuard<'_, Page> {
        self.inner.read()
    }

    pub fn write(&self) -> RwLockWriteGuard<'_, Page> {
        self.inner
            .frame
            .reference_bit
            .store(true, Ordering::Relaxed);
        self.inner.frame.mark_dirty();
        self.inner.frame.page.write()
    }

    pub fn page_id(&self) -> PageId {
        self.inner.page_id()
    }

    pub fn apply_redo(&mut self, data: &[u8]) -> Result<()> {
        let mut page = self.write();
        page.apply_redo(data)
    }
}

impl Drop for PageGuardMut {
    fn drop(&mut self) {
        // inner's drop will handle unpin
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::FileManager;
    use featherdb_core::Config;
    use tempfile::TempDir;

    fn create_test_buffer_pool() -> (TempDir, BufferPool) {
        create_test_buffer_pool_with_policy(EvictionPolicyType::Clock)
    }

    fn create_test_buffer_pool_with_policy(policy: EvictionPolicyType) -> (TempDir, BufferPool) {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("test.db");
        let config = Config::new(&path);

        let fm = Arc::new(FileManager::open(&config).unwrap());
        fm.init_if_needed(&config).unwrap();

        let bp = BufferPool::with_policy(10, fm, policy);
        (tmp, bp)
    }

    #[test]
    fn test_buffer_pool_allocate() {
        let (_tmp, bp) = create_test_buffer_pool();

        let (page_id, guard) = bp.allocate_page(PageType::Leaf).unwrap();
        assert!(page_id.0 >= 3); // First data page

        {
            let page = guard.read();
            assert_eq!(page.page_type(), PageType::Leaf);
        }
    }

    #[test]
    fn test_buffer_pool_read_write() {
        let (_tmp, bp) = create_test_buffer_pool();

        // Allocate and write
        let (page_id, guard) = bp.allocate_page(PageType::Leaf).unwrap();
        {
            let mut page = guard.write();
            page.insert_cell(b"hello");
        }
        drop(guard);

        // Flush
        bp.flush_page(page_id).unwrap();

        // Read back
        let guard = bp.get_page(page_id).unwrap();
        let page = guard.read();
        assert_eq!(page.get_cell(0), Some(b"hello".as_slice()));
    }

    #[test]
    fn test_buffer_pool_eviction_clock() {
        let (_tmp, bp) = create_test_buffer_pool_with_policy(EvictionPolicyType::Clock);

        // Allocate more pages than capacity
        for _ in 0..15 {
            let (_, guard) = bp.allocate_page(PageType::Leaf).unwrap();
            drop(guard); // Unpin immediately
        }

        // Should have evicted some pages
        assert!(bp.stats.pages_evicted.load(Ordering::Relaxed) > 0);
        assert_eq!(bp.eviction_policy_name(), "Clock");
    }

    #[test]
    fn test_buffer_pool_eviction_lru2() {
        let (_tmp, bp) = create_test_buffer_pool_with_policy(EvictionPolicyType::Lru2);

        assert_eq!(bp.eviction_policy_name(), "LRU-2");

        // Allocate more pages than capacity
        for _ in 0..15 {
            let (_, guard) = bp.allocate_page(PageType::Leaf).unwrap();
            drop(guard); // Unpin immediately
        }

        // Should have evicted some pages
        assert!(bp.stats.pages_evicted.load(Ordering::Relaxed) > 0);
    }

    #[test]
    fn test_buffer_pool_eviction_lirs() {
        let (_tmp, bp) = create_test_buffer_pool_with_policy(EvictionPolicyType::Lirs);

        assert_eq!(bp.eviction_policy_name(), "LIRS");

        // Allocate more pages than capacity
        for _ in 0..15 {
            let (_, guard) = bp.allocate_page(PageType::Leaf).unwrap();
            drop(guard); // Unpin immediately
        }

        // Should have evicted some pages
        assert!(bp.stats.pages_evicted.load(Ordering::Relaxed) > 0);
    }

    #[test]
    fn test_buffer_pool_stats() {
        let (_tmp, bp) = create_test_buffer_pool();

        // Allocate a page
        let (page_id, guard) = bp.allocate_page(PageType::Leaf).unwrap();
        drop(guard);

        // Access the cached page (should be a hit)
        let _guard = bp.get_page(page_id).unwrap();

        let stats = bp.stats_snapshot();
        assert!(stats.cache_hits > 0);
        assert!(stats.hit_ratio() > 0.0);
    }

    #[test]
    fn test_buffer_pool_hit_ratio() {
        let (_tmp, bp) = create_test_buffer_pool();

        // Create some pages
        let mut page_ids = Vec::new();
        for _ in 0..5 {
            let (page_id, guard) = bp.allocate_page(PageType::Leaf).unwrap();
            page_ids.push(page_id);
            drop(guard);
        }

        // Access them multiple times (cache hits)
        for _ in 0..10 {
            for &page_id in &page_ids {
                let _guard = bp.get_page(page_id).unwrap();
            }
        }

        let stats = bp.stats_snapshot();
        assert!(
            stats.hit_ratio() > 0.9,
            "Hit ratio should be high: {}",
            stats.hit_ratio()
        );
    }
}