radixdb-storage 1.1.0

// Copyright 2026 RadixDB Contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Bitmap Index implementation for low-cardinality columns
//!
//! This module provides a bitmap-based index optimized for columns with
//! few distinct values (< 1000). Uses RoaringBitmap for compression.
//!
//! ## Performance characteristics:
//! - INSERT: O(1)
//! - DELETE: O(1)
//! - FIND exact: O(1) bitmap lookup
//! - AND/OR operations: O(n/64) bitwise operations
//!
//! ## When to use BitmapIndex:
//! - BOOLEAN columns (only 2 values)
//! - Status/state columns (pending, active, completed, etc.)
//! - Category columns with few values
//! - Any column with < 1000 distinct values
//!
//! ## When NOT to use BitmapIndex:
//! - High-cardinality columns (use Hash or BTree)
//! - Range queries on numeric data (use BTree)
//! - Unique columns (use Hash or BTree)
//!
//! ## Key advantage:
//! Multi-predicate queries (WHERE a = 1 AND b = 2 AND c = 3) can be
//! answered with O(n/64) bitwise AND operations instead of O(n) scans.
//!
//! ## Implementation:
//! Uses `roaring` crate (RoaringBitmap) - same as Lucene, Druid, Spark.
//! Automatic compression: array for sparse, bitmap for dense, RLE for runs.

use parking_lot::RwLock;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering};

use ahash::AHashMap;
use roaring::RoaringTreemap;

use crate::expression::Expression;
use crate::traits::Index;
use radixdb_core::{CompactArc, I64Map};
use radixdb_core::{DataType, Error, IndexEntry, IndexType, Operator, Result, RowIdVec, Value};

/// Warning threshold for cardinality
const HIGH_CARDINALITY_WARNING_THRESHOLD: usize = 1000;

/// Bitmap Index for low-cardinality columns
///
/// Optimized for BOOLEAN, status, and category columns where the number
/// of distinct values is small (< 1000).
///
/// ## Key features:
/// - One RoaringBitmap per distinct value
/// - O(n/64) AND/OR/NOT operations for multi-predicate queries
/// - Automatic compression (array, bitmap, or RLE per 8KB chunk)
/// - Thread-safe with RwLock
///
/// ## Memory efficiency:
/// For 1M rows with 10 distinct values:
/// - B-tree: ~80 MB
/// - Bitmap: ~500KB - 2.5MB (30-160x savings)
///
/// ## Row ID Support:
/// Uses RoaringTreemap which supports full i64 row IDs (up to u64::MAX).
/// This is implemented as a BTreeMap of RoaringBitmaps for efficient
/// storage while supporting the full 64-bit range.
pub struct BitmapIndex {
    name: String,
    table_name: String,
    column_names: Vec<String>,
    column_ids: Vec<i32>,
    data_types: Vec<DataType>,
    is_unique: bool,
    closed: AtomicBool,

    /// One bitmap per distinct value
    /// Maps `CompactArc<Value>` -> `RoaringTreemap` of row IDs (supports full u64 range)
    /// Uses `CompactArc<Value>` keys for memory efficiency (8 bytes per key)
    /// AHash for HashDoS resistance (user-controlled indexed values)
    bitmaps: RwLock<AHashMap<CompactArc<Value>, RoaringTreemap>>,

    /// Reverse mapping: row_id -> `CompactArc<Value>` for efficient removal
    /// A normal signed-key map is required here: `I64Map` reserves i64::MIN,
    /// while public row identity covers the complete i64 domain.
    row_to_value: RwLock<AHashMap<i64, CompactArc<Value>>>,

    /// Track cardinality for warnings
    distinct_count: AtomicUsize,
}

impl std::fmt::Debug for BitmapIndex {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BitmapIndex")
            .field("name", &self.name)
            .field("table_name", &self.table_name)
            .field("column_names", &self.column_names)
            .field("column_ids", &self.column_ids)
            .field("is_unique", &self.is_unique)
            .field(
                "distinct_count",
                &self.distinct_count.load(AtomicOrdering::Relaxed),
            )
            .field("closed", &self.closed.load(AtomicOrdering::Relaxed))
            .finish_non_exhaustive()
    }
}

impl BitmapIndex {
    /// Create a new BitmapIndex
    ///
    /// # Arguments
    /// * `expected_rows` - Hint for initial capacity of row_to_value map.
    ///   Pass 0 if unknown; the map will grow automatically.
    #[doc(hidden)]
    pub fn new(
        name: String,
        table_name: String,
        column_names: Vec<String>,
        column_ids: Vec<i32>,
        data_types: Vec<DataType>,
        is_unique: bool,
        expected_rows: usize,
    ) -> Self {
        Self::try_new(
            name,
            table_name,
            column_names,
            column_ids,
            data_types,
            is_unique,
            expected_rows,
        )
        .expect("internal BitmapIndex metadata must be coherent")
    }

    pub fn try_new(
        name: String,
        table_name: String,
        column_names: Vec<String>,
        column_ids: Vec<i32>,
        data_types: Vec<DataType>,
        is_unique: bool,
        expected_rows: usize,
    ) -> Result<Self> {
        super::validate_index_metadata_shape(
            &name,
            &table_name,
            &column_names,
            &column_ids,
            &data_types,
        )?;
        if column_names.len() != 1 {
            return Err(Error::invalid_argument(
                "BitmapIndex supports exactly one structural Value column",
            ));
        }
        Ok(Self {
            name,
            table_name,
            column_names,
            column_ids,
            data_types,
            is_unique,
            closed: AtomicBool::new(false),
            bitmaps: RwLock::new(AHashMap::default()),
            row_to_value: RwLock::new(AHashMap::with_capacity(expected_rows)),
            distinct_count: AtomicUsize::new(0),
        })
    }

    /// Get the current cardinality (number of distinct values)
    pub fn cardinality(&self) -> usize {
        self.distinct_count.load(AtomicOrdering::Relaxed)
    }

    /// Check if cardinality is too high for efficient bitmap operations
    pub fn is_high_cardinality(&self) -> bool {
        self.cardinality() > HIGH_CARDINALITY_WARNING_THRESHOLD
    }

    /// Get the bitmap for a specific value (for AND/OR operations)
    pub fn get_bitmap(&self, value: &Value) -> Option<RoaringTreemap> {
        // Intern value to get Arc for lookup
        let arc_key = self.value_to_arc_key(std::slice::from_ref(value));
        let bitmaps = self.bitmaps.read();
        bitmaps.get(&arc_key).cloned()
    }

    /// Perform AND operation on multiple values (for multi-predicate queries)
    /// Returns row IDs that match ALL values
    pub fn and_values(&self, values: &[Value]) -> RoaringTreemap {
        let bitmaps = self.bitmaps.read();
        let mut result: Option<RoaringTreemap> = None;

        for value in values {
            let arc_key = self.value_to_arc_key(std::slice::from_ref(value));
            if let Some(bitmap) = bitmaps.get(&arc_key) {
                result = Some(match result {
                    Some(r) => r & bitmap,
                    None => bitmap.clone(),
                });
            } else {
                // Value not found - result is empty
                return RoaringTreemap::new();
            }
        }

        result.unwrap_or_default()
    }

    /// Perform OR operation on multiple values
    /// Returns row IDs that match ANY value
    pub fn or_values(&self, values: &[Value]) -> RoaringTreemap {
        let bitmaps = self.bitmaps.read();
        let mut result = RoaringTreemap::new();

        for value in values {
            let arc_key = self.value_to_arc_key(std::slice::from_ref(value));
            if let Some(bitmap) = bitmaps.get(&arc_key) {
                result |= bitmap;
            }
        }

        result
    }

    /// Perform NOT operation on a value
    /// Returns row IDs that do NOT match the value
    /// Note: Requires knowing all row IDs in the table
    pub fn not_value(&self, value: &Value) -> RoaringTreemap {
        let arc_key = self.value_to_arc_key(std::slice::from_ref(value));
        let bitmaps = self.bitmaps.read();

        // Get all row IDs (union of all bitmaps)
        let mut all_rows = RoaringTreemap::new();
        for bitmap in bitmaps.values() {
            all_rows |= bitmap;
        }

        // Subtract the matching bitmap
        if let Some(bitmap) = bitmaps.get(&arc_key) {
            all_rows - bitmap
        } else {
            all_rows
        }
    }

    /// Convert values to an CompactArc<Value> key
    /// For single-column indexes, wraps the value in Arc
    /// For multi-column indexes, creates a composite key
    fn value_to_arc_key(&self, values: &[Value]) -> CompactArc<Value> {
        if values.len() == 1 {
            CompactArc::new(values[0].clone())
        } else {
            // For multi-column bitmap index, create a composite key
            // This is less common but supported
            let composite = Value::Text(
                values
                    .iter()
                    .map(|v| format!("{:?}", v))
                    .collect::<Vec<_>>()
                    .join("||")
                    .into(),
            );
            CompactArc::new(composite)
        }
    }

    /// Slow path for multi-column indexes (rare case)
    /// Creates composite key and handles add operation
    #[cold]
    fn add_multi_column_slow(
        &self,
        values: &[Value],
        row_id: i64,
        row_id_u64: u64,
        mut bitmaps: parking_lot::RwLockWriteGuard<'_, AHashMap<CompactArc<Value>, RoaringTreemap>>,
        mut row_to_value: parking_lot::RwLockWriteGuard<'_, AHashMap<i64, CompactArc<Value>>>,
    ) -> Result<()> {
        // Create composite key for multi-column lookup
        let arc_key = self.value_to_arc_key(values);

        // Check uniqueness constraint
        if self.is_unique {
            let has_null = values.iter().any(|v| v.is_null());
            if !has_null {
                if let Some(bitmap) = bitmaps.get(&arc_key) {
                    let existing_count = if bitmap.contains(row_id_u64) {
                        bitmap.len() - 1
                    } else {
                        bitmap.len()
                    };
                    if existing_count > 0 {
                        let values_str: Vec<String> =
                            values.iter().map(|v| format!("{:?}", v)).collect();
                        return Err(Error::unique_constraint(
                            &self.name,
                            self.column_names.join(", "),
                            format!("[{}]", values_str.join(", ")),
                        ));
                    }
                }
            }
        }

        // Check if row already exists with a different value
        if let Some(old_arc_key) = row_to_value.get(&row_id).cloned() {
            if !CompactArc::ptr_eq(&old_arc_key, &arc_key) {
                if let Some(old_bitmap) = bitmaps.get_mut(&old_arc_key) {
                    old_bitmap.remove(row_id_u64);
                    if old_bitmap.is_empty() {
                        bitmaps.remove(&old_arc_key);
                        self.distinct_count.fetch_sub(1, AtomicOrdering::Relaxed);
                    }
                }
            }
        }

        // Add to bitmap
        let is_new_value = !bitmaps.contains_key(&arc_key);
        let bitmap = bitmaps.entry(CompactArc::clone(&arc_key)).or_default();
        bitmap.insert(row_id_u64);

        // Update reverse mapping
        row_to_value.insert(row_id, arc_key);

        if is_new_value {
            self.distinct_count.fetch_add(1, AtomicOrdering::Relaxed);
        }

        Ok(())
    }
}

impl Index for BitmapIndex {
    fn name(&self) -> &str {
        &self.name
    }

    fn table_name(&self) -> &str {
        &self.table_name
    }

    fn build(&mut self) -> Result<()> {
        Ok(())
    }

    fn add(&self, values: &[Value], row_id: i64, _ref_id: i64) -> Result<()> {
        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        // `as u64` is a reversible bit-pattern mapping for the full i64
        // domain; result paths cast it back to i64.
        let row_id_u64 = row_id as u64;

        let num_cols = self.column_ids.len();
        if values.len() != num_cols {
            return Err(Error::internal(format!(
                "expected {} values, got {}",
                num_cols,
                values.len()
            )));
        }

        // Acquire write locks
        let mut bitmaps = self.bitmaps.write();
        let mut row_to_value = self.row_to_value.write();

        // Build lookup key (for single-column, just the first value)
        let lookup_value = if values.len() == 1 {
            &values[0]
        } else {
            // For multi-column, we need to create a composite key for lookup
            // This is less common but supported
            return self.add_multi_column_slow(values, row_id, row_id_u64, bitmaps, row_to_value);
        };

        // Check uniqueness constraint (using Borrow trait for &Value lookup)
        if self.is_unique && !lookup_value.is_null() {
            if let Some(bitmap) = bitmaps.get(lookup_value) {
                // Check if there's already a row with this value (excluding current row)
                let existing_count = if bitmap.contains(row_id_u64) {
                    bitmap.len() - 1
                } else {
                    bitmap.len()
                };
                if existing_count > 0 {
                    return Err(Error::unique_constraint(
                        &self.name,
                        self.column_names.join(", "),
                        format!("{:?}", lookup_value),
                    ));
                }
            }
        }

        // Try to reuse existing Arc if value already exists (O(1) clone)
        // Only create new Arc if this is a new unique value
        let (arc_key, is_new_value) =
            if let Some((existing_arc, _)) = bitmaps.get_key_value(lookup_value) {
                // Value exists - reuse the existing Arc (O(1) atomic refcount bump)
                (CompactArc::clone(existing_arc), false)
            } else {
                // New unique value - create Arc once
                (CompactArc::new(lookup_value.clone()), true)
            };

        // Check if row already exists with a different value (for updates)
        if let Some(old_arc_key) = row_to_value.get(&row_id).cloned() {
            // Compare Arc pointers - if same Arc, same value
            if !CompactArc::ptr_eq(&old_arc_key, &arc_key) {
                // Remove from old bitmap
                if let Some(old_bitmap) = bitmaps.get_mut(&old_arc_key) {
                    old_bitmap.remove(row_id_u64);
                    if old_bitmap.is_empty() {
                        bitmaps.remove(&old_arc_key);
                        self.distinct_count.fetch_sub(1, AtomicOrdering::Relaxed);
                    }
                }
            }
        }

        // Add to bitmap
        let bitmap = bitmaps.entry(CompactArc::clone(&arc_key)).or_default();
        bitmap.insert(row_id_u64);

        // Update reverse mapping with Arc reference
        row_to_value.insert(row_id, arc_key);

        // Update cardinality if this is a new distinct value
        if is_new_value {
            self.distinct_count.fetch_add(1, AtomicOrdering::Relaxed);
        }

        Ok(())
    }

    fn add_batch(&self, entries: &I64Map<Vec<Value>>) -> Result<()> {
        let borrowed: Vec<(i64, &[Value])> = entries
            .iter()
            .map(|(row_id, values)| (row_id, values.as_slice()))
            .collect();
        self.add_batch_slice(&borrowed)
    }

    fn remove(&self, values: &[Value], row_id: i64, _ref_id: i64) -> Result<()> {
        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        if values.len() != 1 {
            return Err(Error::invalid_argument("bitmap index requires one value"));
        }
        let row_id_u64 = row_id as u64;

        let mut bitmaps = self.bitmaps.write();
        let mut row_to_value = self.row_to_value.write();

        let Some(arc_key) = row_to_value.get(&row_id).cloned() else {
            return Ok(());
        };
        if arc_key.as_ref() != &values[0] {
            return Err(Error::invalid_argument(format!(
                "stale value supplied while removing row {row_id} from index '{}'",
                self.name
            )));
        }

        // Remove from bitmap
        if let Some(bitmap) = bitmaps.get_mut(&arc_key) {
            bitmap.remove(row_id_u64);
            if bitmap.is_empty() {
                bitmaps.remove(&arc_key);
                self.distinct_count.fetch_sub(1, AtomicOrdering::Relaxed);
            }
        }

        // Remove from reverse mapping
        row_to_value.remove(&row_id);

        Ok(())
    }

    fn remove_batch(&self, entries: &I64Map<Vec<Value>>) -> Result<()> {
        let borrowed: Vec<(i64, &[Value])> = entries
            .iter()
            .map(|(row_id, values)| (row_id, values.as_slice()))
            .collect();
        self.remove_batch_slice(&borrowed)
    }

    /// Optimized batch add with single lock acquisition
    fn add_batch_slice(&self, entries: &[(i64, &[Value])]) -> Result<()> {
        if entries.is_empty() {
            return Ok(());
        }

        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        let num_cols = self.column_ids.len();

        for &(_, values) in entries {
            if values.len() != num_cols {
                return Err(Error::internal(format!(
                    "expected {} values, got {}",
                    num_cols,
                    values.len()
                )));
            }
        }

        // Acquire write locks ONCE for entire batch
        let mut bitmaps = self.bitmaps.write();
        let mut row_to_value = self.row_to_value.write();
        row_to_value.reserve(entries.len());

        // PRE-CHECK PHASE: Validate all unique constraints BEFORE modifying anything
        // This prevents partial batch execution on failure
        if self.is_unique {
            // Track keys seen in this batch for intra-batch duplicate detection
            let mut batch_keys: AHashMap<CompactArc<Value>, i64> =
                AHashMap::with_capacity(entries.len());

            for &(row_id, values) in entries {
                // NULL values don't violate uniqueness
                if values.iter().any(|v| v.is_null()) {
                    continue;
                }

                let row_id_u64 = row_id as u64;
                let arc_key = self.value_to_arc_key(values);

                // Check intra-batch duplicates
                if let Some(&existing_row_id) = batch_keys.get(&arc_key) {
                    if existing_row_id != row_id {
                        return Err(Error::unique_constraint(
                            &self.name,
                            self.column_names.join(", "),
                            format!("{:?}", values),
                        ));
                    }
                }

                // Check against existing index
                if let Some(bitmap) = bitmaps.get(&arc_key) {
                    let existing_count = if bitmap.contains(row_id_u64) {
                        bitmap.len() - 1
                    } else {
                        bitmap.len()
                    };
                    if existing_count > 0 {
                        return Err(Error::unique_constraint(
                            &self.name,
                            self.column_names.join(", "),
                            format!("{:?}", values),
                        ));
                    }
                }

                batch_keys.insert(arc_key, row_id);
            }
        }

        // MODIFICATION PHASE: All constraints checked, now safe to modify
        for &(row_id, values) in entries {
            let row_id_u64 = row_id as u64;

            let arc_key = self.value_to_arc_key(values);

            // Try to reuse existing Arc
            let (final_arc_key, is_new_value) =
                if let Some((existing_arc, _)) = bitmaps.get_key_value(&arc_key) {
                    (CompactArc::clone(existing_arc), false)
                } else {
                    (arc_key, true)
                };

            // Handle update case - remove from old bitmap
            if let Some(old_arc_key) = row_to_value.get(&row_id).cloned() {
                if !CompactArc::ptr_eq(&old_arc_key, &final_arc_key) {
                    if let Some(old_bitmap) = bitmaps.get_mut(&old_arc_key) {
                        old_bitmap.remove(row_id_u64);
                        if old_bitmap.is_empty() {
                            bitmaps.remove(&old_arc_key);
                            self.distinct_count.fetch_sub(1, AtomicOrdering::Relaxed);
                        }
                    }
                }
            }

            // Add to bitmap
            let bitmap = bitmaps
                .entry(CompactArc::clone(&final_arc_key))
                .or_default();
            bitmap.insert(row_id_u64);

            // Update reverse mapping
            row_to_value.insert(row_id, final_arc_key);

            if is_new_value {
                self.distinct_count.fetch_add(1, AtomicOrdering::Relaxed);
            }
        }

        Ok(())
    }

    /// Optimized batch remove with single lock acquisition
    fn remove_batch_slice(&self, entries: &[(i64, &[Value])]) -> Result<()> {
        if entries.is_empty() {
            return Ok(());
        }

        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        // Acquire write locks ONCE for entire batch
        let mut bitmaps = self.bitmaps.write();
        let mut row_to_value = self.row_to_value.write();

        let mut removals = Vec::with_capacity(entries.len());
        for &(row_id, values) in entries {
            if values.len() != 1 {
                return Err(Error::invalid_argument("bitmap index requires one value"));
            }
            let Some(arc_key) = row_to_value.get(&row_id).cloned() else {
                continue;
            };
            if arc_key.as_ref() != &values[0] {
                return Err(Error::invalid_argument(format!(
                    "stale value supplied while removing row {row_id} from index '{}'",
                    self.name
                )));
            }
            removals.push((row_id, arc_key));
        }

        for (row_id, arc_key) in removals {
            let row_id_u64 = row_id as u64;

            // Remove from bitmap
            if let Some(bitmap) = bitmaps.get_mut(&arc_key) {
                bitmap.remove(row_id_u64);
                if bitmap.is_empty() {
                    bitmaps.remove(&arc_key);
                    self.distinct_count.fetch_sub(1, AtomicOrdering::Relaxed);
                }
            }

            // Remove from reverse mapping
            row_to_value.remove(&row_id);
        }

        Ok(())
    }

    fn column_ids(&self) -> &[i32] {
        &self.column_ids
    }

    fn column_names(&self) -> &[String] {
        &self.column_names
    }

    fn data_types(&self) -> &[DataType] {
        &self.data_types
    }

    fn index_type(&self) -> IndexType {
        IndexType::Bitmap
    }

    fn is_unique(&self) -> bool {
        self.is_unique
    }

    fn find(&self, values: &[Value]) -> Result<Vec<IndexEntry>> {
        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        if values.len() != self.column_ids.len() {
            return Err(Error::internal(
                "bitmap index requires exact match on all columns",
            ));
        }

        // Intern value to get Arc key for lookup
        let arc_key = self.value_to_arc_key(values);
        let bitmaps = self.bitmaps.read();

        if let Some(bitmap) = bitmaps.get(&arc_key) {
            Ok(bitmap
                .iter()
                .map(|row_id| IndexEntry {
                    row_id: row_id as i64,
                    ref_id: 0,
                })
                .collect())
        } else {
            Ok(vec![])
        }
    }

    fn find_range(
        &self,
        _min: &[Value],
        _max: &[Value],
        _min_inclusive: bool,
        _max_inclusive: bool,
    ) -> Result<Vec<IndexEntry>> {
        // Bitmap index doesn't efficiently support range queries
        // For ordered values, we could iterate through bitmaps, but it's not optimal
        Err(Error::internal(
            "bitmap index does not efficiently support range queries; use btree index instead",
        ))
    }

    fn find_with_operator(&self, op: Operator, values: &[Value]) -> Result<Vec<IndexEntry>> {
        match op {
            Operator::Eq => self.find(values),
            Operator::Ne => {
                // Use NOT operation
                if values.len() != self.column_ids.len() {
                    return Err(Error::internal(
                        "bitmap index requires exact match on all columns",
                    ));
                }
                // not_value() takes &Value, not the key directly
                // For single column, pass the value; for multi, create composite
                let result = if values.len() == 1 {
                    self.not_value(&values[0])
                } else {
                    let composite = Value::Text(
                        values
                            .iter()
                            .map(|v| format!("{:?}", v))
                            .collect::<Vec<_>>()
                            .join("||")
                            .into(),
                    );
                    self.not_value(&composite)
                };
                Ok(result
                    .iter()
                    .map(|row_id| IndexEntry {
                        row_id: row_id as i64,
                        ref_id: 0,
                    })
                    .collect())
            }
            _ => Err(Error::internal(format!(
                "bitmap index only supports = and != operators, not {:?}",
                op
            ))),
        }
    }

    fn get_row_ids_equal_into(&self, values: &[Value], buffer: &mut Vec<i64>) -> Result<()> {
        if self.closed.load(AtomicOrdering::Acquire) {
            return Err(Error::IndexClosed);
        }

        if values.len() != self.column_ids.len() {
            return Err(Error::invalid_argument(
                "bitmap index requires exact values for every indexed column",
            ));
        }

        // Intern value to get Arc key for lookup
        let arc_key = self.value_to_arc_key(values);
        let bitmaps = self.bitmaps.read();

        if let Some(bitmap) = bitmaps.get(&arc_key) {
            // RoaringTreemap iteration is efficient
            buffer.extend(bitmap.iter().map(|row_id| row_id as i64));
        }
        Ok(())
    }

    fn get_row_ids_in_range_into(
        &self,
        _min_value: &[Value],
        _max_value: &[Value],
        _include_min: bool,
        _include_max: bool,
        _buffer: &mut Vec<i64>,
    ) -> Result<()> {
        Err(Error::NotSupported(
            "bitmap index does not support range queries".to_string(),
        ))
    }

    fn get_filtered_row_ids(&self, expr: &dyn Expression) -> Result<RowIdVec> {
        // For complex expressions, return all row IDs and let caller filter
        let bitmaps = self.bitmaps.read();
        let mut all_rows = RoaringTreemap::new();
        for bitmap in bitmaps.values() {
            all_rows |= bitmap;
        }
        let _ = expr;
        let collected: Vec<i64> = all_rows.iter().map(|id| id as i64).collect();
        Ok(RowIdVec::from_vec(collected))
    }

    fn get_all_values(&self) -> Vec<Value> {
        let bitmaps = self.bitmaps.read();
        // Dereference CompactArc<Value> to clone inner Value
        bitmaps.keys().map(|arc| (**arc).clone()).collect()
    }

    fn clear(&self) {
        *self.bitmaps.write() = AHashMap::default();
        *self.row_to_value.write() = AHashMap::default();
        self.distinct_count.store(0, AtomicOrdering::Relaxed);
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn close(&mut self) -> Result<()> {
        self.closed.store(true, AtomicOrdering::Release);
        self.clear();
        Ok(())
    }
}

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

    #[test]
    fn test_bitmap_index_basic() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        // Add entries
        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("pending".into())], 2, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 3, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 4, 0).unwrap();
        index.add(&[Value::Text("delivered".into())], 5, 0).unwrap();

        // Check cardinality
        assert_eq!(index.cardinality(), 3);

        // Find by status
        let results = index.find(&[Value::Text("pending".into())]).unwrap();
        assert_eq!(results.len(), 2);

        let results = index.find(&[Value::Text("shipped".into())]).unwrap();
        assert_eq!(results.len(), 2);

        let results = index.find(&[Value::Text("delivered".into())]).unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_bitmap_index_boolean() {
        let index = BitmapIndex::new(
            "idx_active".to_string(),
            "users".to_string(),
            vec!["active".to_string()],
            vec![1],
            vec![DataType::Boolean],
            false,
            0,
        );

        // Add boolean values
        index.add(&[Value::Boolean(true)], 1, 0).unwrap();
        index.add(&[Value::Boolean(true)], 2, 0).unwrap();
        index.add(&[Value::Boolean(true)], 3, 0).unwrap();
        index.add(&[Value::Boolean(false)], 4, 0).unwrap();
        index.add(&[Value::Boolean(false)], 5, 0).unwrap();

        // Only 2 distinct values
        assert_eq!(index.cardinality(), 2);

        let active = index.find(&[Value::Boolean(true)]).unwrap();
        assert_eq!(active.len(), 3);

        let inactive = index.find(&[Value::Boolean(false)]).unwrap();
        assert_eq!(inactive.len(), 2);
    }

    #[test]
    fn test_bitmap_index_and_operation() {
        // Note: AND operation is for combining multiple bitmap indexes
        // Here we test the single-index AND which isn't as useful
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 2, 0).unwrap();

        // AND of different values in same column = empty (row can't have two values)
        let result =
            index.and_values(&[Value::Text("pending".into()), Value::Text("shipped".into())]);
        assert!(result.is_empty());
    }

    #[test]
    fn test_bitmap_index_or_operation() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("pending".into())], 2, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 3, 0).unwrap();
        index.add(&[Value::Text("delivered".into())], 4, 0).unwrap();

        // OR of pending and shipped
        let result =
            index.or_values(&[Value::Text("pending".into()), Value::Text("shipped".into())]);
        assert_eq!(result.len(), 3); // rows 1, 2, 3
    }

    #[test]
    fn test_bitmap_index_not_operation() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("pending".into())], 2, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 3, 0).unwrap();
        index.add(&[Value::Text("delivered".into())], 4, 0).unwrap();

        // NOT pending = shipped + delivered
        let result = index.not_value(&Value::Text("pending".into()));
        assert_eq!(result.len(), 2); // rows 3, 4
    }

    #[test]
    fn test_bitmap_index_remove() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("pending".into())], 2, 0).unwrap();

        assert_eq!(index.cardinality(), 1);

        // Remove one
        index
            .remove(&[Value::Text("pending".into())], 1, 0)
            .unwrap();

        let results = index.find(&[Value::Text("pending".into())]).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].row_id, 2);

        // Remove last one - cardinality should decrease
        index
            .remove(&[Value::Text("pending".into())], 2, 0)
            .unwrap();
        assert_eq!(index.cardinality(), 0);
    }

    #[test]
    fn test_bitmap_index_update() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        // Initial value
        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        assert_eq!(index.cardinality(), 1);

        // Update to new value
        index.add(&[Value::Text("shipped".into())], 1, 0).unwrap();
        assert_eq!(index.cardinality(), 1); // Old empty bitmap removed

        // Old value should be gone
        let pending = index.find(&[Value::Text("pending".into())]).unwrap();
        assert!(pending.is_empty());

        // New value should be found
        let shipped = index.find(&[Value::Text("shipped".into())]).unwrap();
        assert_eq!(shipped.len(), 1);
        assert_eq!(shipped[0].row_id, 1);
    }

    #[test]
    fn test_bitmap_index_not_equal() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();
        index.add(&[Value::Text("shipped".into())], 2, 0).unwrap();
        index.add(&[Value::Text("delivered".into())], 3, 0).unwrap();

        // != pending should return shipped and delivered
        let results = index
            .find_with_operator(Operator::Ne, &[Value::Text("pending".into())])
            .unwrap();
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_bitmap_index_high_cardinality_check() {
        let index = BitmapIndex::new(
            "idx_id".to_string(),
            "items".to_string(),
            vec!["id".to_string()],
            vec![1],
            vec![DataType::Integer],
            false,
            0,
        );

        // Add many distinct values
        for i in 0..HIGH_CARDINALITY_WARNING_THRESHOLD + 100 {
            index.add(&[Value::Integer(i as i64)], i as i64, 0).unwrap();
        }

        assert!(index.is_high_cardinality());
    }

    #[test]
    fn test_bitmap_index_null_handling() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        index.add(&[Value::Null(DataType::Text)], 1, 0).unwrap();
        index.add(&[Value::Null(DataType::Text)], 2, 0).unwrap();
        index.add(&[Value::Text("active".into())], 3, 0).unwrap();

        let nulls = index.find(&[Value::Null(DataType::Text)]).unwrap();
        assert_eq!(nulls.len(), 2);
    }

    #[test]
    fn test_bitmap_index_unique_constraint() {
        let index = BitmapIndex::new(
            "idx_status_unique".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            true, // unique
            0,
        );

        index.add(&[Value::Text("pending".into())], 1, 0).unwrap();

        // Try to add duplicate - should fail
        let result = index.add(&[Value::Text("pending".into())], 2, 0);
        assert!(result.is_err());

        // NULL doesn't violate uniqueness
        index.add(&[Value::Null(DataType::Text)], 3, 0).unwrap();
        index.add(&[Value::Null(DataType::Text)], 4, 0).unwrap();
    }

    #[test]
    fn test_bitmap_index_large_row_ids() {
        // Test that bitmap index correctly handles row IDs > u32::MAX
        // This uses RoaringTreemap (64-bit) instead of RoaringBitmap (32-bit)
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        // Use row IDs beyond u32::MAX
        let large_row_id_1: i64 = (u32::MAX as i64) + 1; // 4,294,967,296
        let large_row_id_2: i64 = (u32::MAX as i64) + 1000; // 4,294,968,295
        let large_row_id_3: i64 = i64::MAX / 2; // Very large value

        index
            .add(&[Value::Text("active".into())], large_row_id_1, 0)
            .unwrap();
        index
            .add(&[Value::Text("active".into())], large_row_id_2, 0)
            .unwrap();
        index
            .add(&[Value::Text("inactive".into())], large_row_id_3, 0)
            .unwrap();

        // Verify we can find them
        let results = index.find(&[Value::Text("active".into())]).unwrap();
        assert_eq!(results.len(), 2);
        assert!(results.iter().any(|e| e.row_id == large_row_id_1));
        assert!(results.iter().any(|e| e.row_id == large_row_id_2));

        let results = index.find(&[Value::Text("inactive".into())]).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].row_id, large_row_id_3);

        // Verify removal works
        index
            .remove(&[Value::Text("active".into())], large_row_id_1, 0)
            .unwrap();
        let results = index.find(&[Value::Text("active".into())]).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].row_id, large_row_id_2);
    }

    #[test]
    fn test_bitmap_index_supports_full_signed_row_id_domain() {
        let index = BitmapIndex::new(
            "idx_status".to_string(),
            "orders".to_string(),
            vec!["status".to_string()],
            vec![1],
            vec![DataType::Text],
            false,
            0,
        );

        let key = [Value::Text("pending".into())];
        index.add(&key, -1, 0).unwrap();
        index.add(&key, i64::MIN, 0).unwrap();
        let mut row_ids = index.get_row_ids_equal(&key).unwrap().into_vec();
        row_ids.sort_unstable();
        assert_eq!(row_ids, vec![i64::MIN, -1]);

        index.remove(&key, -1, 0).unwrap();
        index.remove(&key, i64::MIN, 0).unwrap();
        assert!(index.get_row_ids_equal(&key).unwrap().is_empty());
    }

    #[test]
    fn v2_r3_stale_remove_is_rejected_without_losing_bitmap_ownership() {
        let index = BitmapIndex::new(
            "idx".into(),
            "items".into(),
            vec!["state".into()],
            vec![0],
            vec![DataType::Text],
            false,
            0,
        );
        let a = [Value::text("a")];
        let b = [Value::text("b")];
        let c = [Value::text("c")];
        index.add(&a, 1, 1).unwrap();
        index.add(&c, 2, 2).unwrap();

        assert!(index.remove(&b, 1, 1).is_err());
        assert_eq!(index.find(&a).unwrap()[0].row_id, 1);

        let removals = [(1, a.as_slice()), (2, b.as_slice())];
        assert!(index.remove_batch_slice(&removals).is_err());
        assert_eq!(index.find(&a).unwrap()[0].row_id, 1);
        assert_eq!(index.find(&c).unwrap()[0].row_id, 2);

        index.add(&b, 1, 1).unwrap();
        assert!(index.find(&a).unwrap().is_empty());
        assert_eq!(index.find(&b).unwrap()[0].row_id, 1);
    }
}