Skip to main content

lance_index/scalar/
bitmap.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4use std::{
5    any::Any,
6    collections::{BTreeMap, HashMap},
7    fmt::Debug,
8    ops::Bound,
9    sync::Arc,
10};
11
12use arrow::array::BinaryBuilder;
13use arrow_array::{Array, BinaryArray, RecordBatch, UInt64Array, new_null_array};
14use arrow_schema::{DataType, Field, Schema};
15use async_trait::async_trait;
16use bytes::Bytes;
17use datafusion::physical_plan::SendableRecordBatchStream;
18use datafusion_common::ScalarValue;
19use deepsize::DeepSizeOf;
20use futures::{StreamExt, TryStreamExt, stream};
21use lance_core::utils::mask::RowSetOps;
22use lance_core::{
23    Error, ROW_ID, Result,
24    cache::{CacheKey, LanceCache, WeakLanceCache},
25    error::LanceOptionExt,
26    utils::{
27        mask::{NullableRowAddrSet, RowAddrTreeMap},
28        tokio::get_num_compute_intensive_cpus,
29    },
30};
31use roaring::RoaringBitmap;
32use serde::Serialize;
33use tracing::instrument;
34
35use super::{AnyQuery, IndexStore, ScalarIndex};
36use super::{
37    BuiltinIndexType, SargableQuery, ScalarIndexParams, SearchResult, btree::OrderableScalarValue,
38};
39use crate::pbold;
40use crate::{Index, IndexType, metrics::MetricsCollector};
41use crate::{
42    frag_reuse::FragReuseIndex,
43    scalar::{
44        CreatedIndex, UpdateCriteria,
45        expression::SargableQueryParser,
46        registry::{
47            DefaultTrainingRequest, ScalarIndexPlugin, TrainingCriteria, TrainingOrdering,
48            TrainingRequest, VALUE_COLUMN_NAME,
49        },
50    },
51};
52use crate::{scalar::IndexReader, scalar::expression::ScalarQueryParser};
53
54pub const BITMAP_LOOKUP_NAME: &str = "bitmap_page_lookup.lance";
55pub const INDEX_STATS_METADATA_KEY: &str = "lance:index_stats";
56
57const MAX_BITMAP_ARRAY_LENGTH: usize = i32::MAX as usize - 1024 * 1024; // leave headroom
58
59const MAX_ROWS_PER_CHUNK: usize = 2 * 1024;
60
61const BITMAP_INDEX_VERSION: u32 = 0;
62
63// We only need to open a file reader if we need to load a bitmap. If all
64// bitmaps are cached we don't open it. If we do open it we should only open it once.
65#[derive(Clone)]
66struct LazyIndexReader {
67    index_reader: Arc<tokio::sync::Mutex<Option<Arc<dyn IndexReader>>>>,
68    store: Arc<dyn IndexStore>,
69}
70
71impl std::fmt::Debug for LazyIndexReader {
72    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73        f.debug_struct("LazyIndexReader")
74            .field("store", &self.store)
75            .finish()
76    }
77}
78
79impl LazyIndexReader {
80    fn new(store: Arc<dyn IndexStore>) -> Self {
81        Self {
82            index_reader: Arc::new(tokio::sync::Mutex::new(None)),
83            store,
84        }
85    }
86
87    async fn get(&self) -> Result<Arc<dyn IndexReader>> {
88        let mut reader = self.index_reader.lock().await;
89        if reader.is_none() {
90            let index_reader = self.store.open_index_file(BITMAP_LOOKUP_NAME).await?;
91            *reader = Some(index_reader);
92        }
93        Ok(reader.as_ref().unwrap().clone())
94    }
95}
96
97/// A scalar index that stores a bitmap for each possible value
98///
99/// This index works best for low-cardinality columns, where the number of unique values is small.
100/// The bitmap stores a list of row ids where the value is present.
101#[derive(Clone, Debug)]
102pub struct BitmapIndex {
103    /// Maps each unique value to its bitmap location in the index file
104    /// The usize value is the row offset in the bitmap_page_lookup.lance file
105    /// for quickly locating the row and reading it out
106    index_map: BTreeMap<OrderableScalarValue, usize>,
107
108    null_map: Arc<RowAddrTreeMap>,
109
110    value_type: DataType,
111
112    store: Arc<dyn IndexStore>,
113
114    index_cache: WeakLanceCache,
115
116    frag_reuse_index: Option<Arc<FragReuseIndex>>,
117
118    lazy_reader: LazyIndexReader,
119}
120
121#[derive(Debug, Clone)]
122pub struct BitmapKey {
123    value: OrderableScalarValue,
124}
125
126impl CacheKey for BitmapKey {
127    type ValueType = RowAddrTreeMap;
128
129    fn key(&self) -> std::borrow::Cow<'_, str> {
130        format!("{}", self.value.0).into()
131    }
132}
133
134impl BitmapIndex {
135    fn new(
136        index_map: BTreeMap<OrderableScalarValue, usize>,
137        null_map: Arc<RowAddrTreeMap>,
138        value_type: DataType,
139        store: Arc<dyn IndexStore>,
140        index_cache: WeakLanceCache,
141        frag_reuse_index: Option<Arc<FragReuseIndex>>,
142    ) -> Self {
143        let lazy_reader = LazyIndexReader::new(store.clone());
144        Self {
145            index_map,
146            null_map,
147            value_type,
148            store,
149            index_cache,
150            frag_reuse_index,
151            lazy_reader,
152        }
153    }
154
155    pub(crate) async fn load(
156        store: Arc<dyn IndexStore>,
157        frag_reuse_index: Option<Arc<FragReuseIndex>>,
158        index_cache: &LanceCache,
159    ) -> Result<Arc<Self>> {
160        let page_lookup_file = store.open_index_file(BITMAP_LOOKUP_NAME).await?;
161        let total_rows = page_lookup_file.num_rows();
162
163        if total_rows == 0 {
164            let schema = page_lookup_file.schema();
165            let data_type = schema.fields[0].data_type();
166            return Ok(Arc::new(Self::new(
167                BTreeMap::new(),
168                Arc::new(RowAddrTreeMap::default()),
169                data_type,
170                store,
171                WeakLanceCache::from(index_cache),
172                frag_reuse_index,
173            )));
174        }
175
176        let mut index_map: BTreeMap<OrderableScalarValue, usize> = BTreeMap::new();
177        let mut null_map = Arc::new(RowAddrTreeMap::default());
178        let mut value_type: Option<DataType> = None;
179        let mut null_location: Option<usize> = None;
180        let mut row_offset = 0;
181
182        for start_row in (0..total_rows).step_by(MAX_ROWS_PER_CHUNK) {
183            let end_row = (start_row + MAX_ROWS_PER_CHUNK).min(total_rows);
184            let chunk = page_lookup_file
185                .read_range(start_row..end_row, Some(&["keys"]))
186                .await?;
187
188            if chunk.num_rows() == 0 {
189                continue;
190            }
191
192            if value_type.is_none() {
193                value_type = Some(chunk.schema().field(0).data_type().clone());
194            }
195
196            let dict_keys = chunk.column(0);
197
198            for idx in 0..chunk.num_rows() {
199                let key = OrderableScalarValue(ScalarValue::try_from_array(dict_keys, idx)?);
200
201                if key.0.is_null() {
202                    null_location = Some(row_offset);
203                } else {
204                    index_map.insert(key, row_offset);
205                }
206
207                row_offset += 1;
208            }
209        }
210
211        if let Some(null_loc) = null_location {
212            let batch = page_lookup_file
213                .read_range(null_loc..null_loc + 1, Some(&["bitmaps"]))
214                .await?;
215
216            let binary_bitmaps = batch
217                .column(0)
218                .as_any()
219                .downcast_ref::<BinaryArray>()
220                .ok_or_else(|| Error::internal("Invalid bitmap column type".to_string()))?;
221            let bitmap_bytes = binary_bitmaps.value(0);
222            let mut bitmap = RowAddrTreeMap::deserialize_from(bitmap_bytes).unwrap();
223
224            // Apply fragment remapping if needed
225            if let Some(fri) = &frag_reuse_index {
226                bitmap = fri.remap_row_addrs_tree_map(&bitmap);
227            }
228
229            null_map = Arc::new(bitmap);
230        }
231
232        let final_value_type = value_type.expect_ok()?;
233
234        Ok(Arc::new(Self::new(
235            index_map,
236            null_map,
237            final_value_type,
238            store,
239            WeakLanceCache::from(index_cache),
240            frag_reuse_index,
241        )))
242    }
243
244    async fn load_bitmap(
245        &self,
246        key: &OrderableScalarValue,
247        metrics: Option<&dyn MetricsCollector>,
248    ) -> Result<Arc<RowAddrTreeMap>> {
249        if key.0.is_null() {
250            return Ok(self.null_map.clone());
251        }
252
253        let cache_key = BitmapKey { value: key.clone() };
254
255        if let Some(cached) = self.index_cache.get_with_key(&cache_key).await {
256            return Ok(cached);
257        }
258
259        // Record that we're loading a partition from disk
260        if let Some(metrics) = metrics {
261            metrics.record_part_load();
262        }
263
264        let row_offset = match self.index_map.get(key) {
265            Some(loc) => *loc,
266            None => return Ok(Arc::new(RowAddrTreeMap::default())),
267        };
268
269        let page_lookup_file = self.lazy_reader.get().await?;
270        let batch = page_lookup_file
271            .read_range(row_offset..row_offset + 1, Some(&["bitmaps"]))
272            .await?;
273
274        let binary_bitmaps = batch
275            .column(0)
276            .as_any()
277            .downcast_ref::<BinaryArray>()
278            .ok_or_else(|| Error::internal("Invalid bitmap column type".to_string()))?;
279        let bitmap_bytes = binary_bitmaps.value(0); // First (and only) row
280        let mut bitmap = RowAddrTreeMap::deserialize_from(bitmap_bytes).unwrap();
281
282        if let Some(fri) = &self.frag_reuse_index {
283            bitmap = fri.remap_row_addrs_tree_map(&bitmap);
284        }
285
286        self.index_cache
287            .insert_with_key(&cache_key, Arc::new(bitmap.clone()))
288            .await;
289
290        Ok(Arc::new(bitmap))
291    }
292
293    pub(crate) fn value_type(&self) -> &DataType {
294        &self.value_type
295    }
296
297    /// Loads the current bitmap index into an in-memory value-to-row-id map.
298    pub(crate) async fn load_bitmap_index_state(
299        &self,
300    ) -> Result<HashMap<ScalarValue, RowAddrTreeMap>> {
301        let mut state = HashMap::new();
302
303        for key in self.index_map.keys() {
304            let bitmap = self.load_bitmap(key, None).await?;
305            state.insert(key.0.clone(), (*bitmap).clone());
306        }
307
308        if !self.null_map.is_empty() {
309            let existing_null = new_null_array(&self.value_type, 1);
310            let existing_null = ScalarValue::try_from_array(existing_null.as_ref(), 0)?;
311            state.insert(existing_null, (*self.null_map).clone());
312        }
313
314        Ok(state)
315    }
316}
317
318impl DeepSizeOf for BitmapIndex {
319    fn deep_size_of_children(&self, context: &mut deepsize::Context) -> usize {
320        let mut total_size = 0;
321
322        total_size += self.index_map.deep_size_of_children(context);
323        total_size += self.store.deep_size_of_children(context);
324
325        total_size
326    }
327}
328
329#[derive(Serialize)]
330struct BitmapStatistics {
331    num_bitmaps: usize,
332}
333
334#[async_trait]
335impl Index for BitmapIndex {
336    fn as_any(&self) -> &dyn Any {
337        self
338    }
339
340    fn as_index(self: Arc<Self>) -> Arc<dyn Index> {
341        self
342    }
343
344    fn as_vector_index(self: Arc<Self>) -> Result<Arc<dyn crate::vector::VectorIndex>> {
345        Err(Error::not_supported_source(
346            "BitmapIndex is not a vector index".into(),
347        ))
348    }
349
350    async fn prewarm(&self) -> Result<()> {
351        let page_lookup_file = self.lazy_reader.get().await?;
352        let total_rows = page_lookup_file.num_rows();
353
354        if total_rows == 0 {
355            return Ok(());
356        }
357
358        for start_row in (0..total_rows).step_by(MAX_ROWS_PER_CHUNK) {
359            let end_row = (start_row + MAX_ROWS_PER_CHUNK).min(total_rows);
360            let chunk = page_lookup_file
361                .read_range(start_row..end_row, None)
362                .await?;
363
364            if chunk.num_rows() == 0 {
365                continue;
366            }
367
368            let dict_keys = chunk.column(0);
369            let binary_bitmaps = chunk.column(1);
370            let bitmap_binary_array = binary_bitmaps
371                .as_any()
372                .downcast_ref::<BinaryArray>()
373                .unwrap();
374
375            for idx in 0..chunk.num_rows() {
376                let key = OrderableScalarValue(ScalarValue::try_from_array(dict_keys, idx)?);
377
378                if key.0.is_null() {
379                    continue;
380                }
381
382                let bitmap_bytes = bitmap_binary_array.value(idx);
383                let mut bitmap = RowAddrTreeMap::deserialize_from(bitmap_bytes).unwrap();
384
385                if let Some(frag_reuse_index_ref) = self.frag_reuse_index.as_ref() {
386                    bitmap = frag_reuse_index_ref.remap_row_addrs_tree_map(&bitmap);
387                }
388
389                let cache_key = BitmapKey { value: key };
390                self.index_cache
391                    .insert_with_key(&cache_key, Arc::new(bitmap))
392                    .await;
393            }
394        }
395
396        Ok(())
397    }
398
399    fn index_type(&self) -> IndexType {
400        IndexType::Bitmap
401    }
402
403    fn statistics(&self) -> Result<serde_json::Value> {
404        let stats = BitmapStatistics {
405            num_bitmaps: self.index_map.len() + if !self.null_map.is_empty() { 1 } else { 0 },
406        };
407        serde_json::to_value(stats).map_err(|e| {
408            Error::internal(format!(
409                "failed to serialize bitmap index statistics: {}",
410                e
411            ))
412        })
413    }
414
415    async fn calculate_included_frags(&self) -> Result<RoaringBitmap> {
416        unimplemented!()
417    }
418}
419
420#[async_trait]
421impl ScalarIndex for BitmapIndex {
422    #[instrument(name = "bitmap_search", level = "debug", skip_all)]
423    async fn search(
424        &self,
425        query: &dyn AnyQuery,
426        metrics: &dyn MetricsCollector,
427    ) -> Result<SearchResult> {
428        let query = query.as_any().downcast_ref::<SargableQuery>().unwrap();
429
430        let (row_ids, null_row_ids) = match query {
431            SargableQuery::Equals(val) => {
432                metrics.record_comparisons(1);
433                if val.is_null() {
434                    // Querying FOR nulls - they are the TRUE result, not NULL result
435                    ((*self.null_map).clone(), None)
436                } else {
437                    let key = OrderableScalarValue(val.clone());
438                    let bitmap = self.load_bitmap(&key, Some(metrics)).await?;
439                    let null_rows = if !self.null_map.is_empty() {
440                        Some((*self.null_map).clone())
441                    } else {
442                        None
443                    };
444                    ((*bitmap).clone(), null_rows)
445                }
446            }
447            SargableQuery::Range(start, end) => {
448                let range_start = match start {
449                    Bound::Included(val) => Bound::Included(OrderableScalarValue(val.clone())),
450                    Bound::Excluded(val) => Bound::Excluded(OrderableScalarValue(val.clone())),
451                    Bound::Unbounded => Bound::Unbounded,
452                };
453
454                let range_end = match end {
455                    Bound::Included(val) => Bound::Included(OrderableScalarValue(val.clone())),
456                    Bound::Excluded(val) => Bound::Excluded(OrderableScalarValue(val.clone())),
457                    Bound::Unbounded => Bound::Unbounded,
458                };
459
460                // Empty range if lower > upper, or if any bound is excluded and lower >= upper.
461                let empty_range = match (&range_start, &range_end) {
462                    (Bound::Included(lower), Bound::Included(upper)) => lower > upper,
463                    (Bound::Included(lower), Bound::Excluded(upper))
464                    | (Bound::Excluded(lower), Bound::Included(upper))
465                    | (Bound::Excluded(lower), Bound::Excluded(upper)) => lower >= upper,
466                    _ => false,
467                };
468
469                let keys: Vec<_> = if empty_range {
470                    Vec::new()
471                } else {
472                    self.index_map
473                        .range((range_start, range_end))
474                        .map(|(k, _v)| k.clone())
475                        .collect()
476                };
477
478                metrics.record_comparisons(keys.len());
479
480                let result = if keys.is_empty() {
481                    RowAddrTreeMap::default()
482                } else {
483                    let bitmaps: Vec<_> = stream::iter(
484                        keys.into_iter()
485                            .map(|key| async move { self.load_bitmap(&key, None).await }),
486                    )
487                    .buffer_unordered(get_num_compute_intensive_cpus())
488                    .try_collect()
489                    .await?;
490
491                    let bitmap_refs: Vec<_> = bitmaps.iter().map(|b| b.as_ref()).collect();
492                    RowAddrTreeMap::union_all(&bitmap_refs)
493                };
494
495                let null_rows = if !self.null_map.is_empty() {
496                    Some((*self.null_map).clone())
497                } else {
498                    None
499                };
500                (result, null_rows)
501            }
502            SargableQuery::IsIn(values) => {
503                metrics.record_comparisons(values.len());
504
505                // Collect keys that exist in the index, tracking if we need nulls
506                let mut has_null = false;
507                let keys: Vec<_> = values
508                    .iter()
509                    .filter_map(|val| {
510                        if val.is_null() {
511                            has_null = true;
512                            None
513                        } else {
514                            let key = OrderableScalarValue(val.clone());
515                            if self.index_map.contains_key(&key) {
516                                Some(key)
517                            } else {
518                                None
519                            }
520                        }
521                    })
522                    .collect();
523
524                // Load bitmaps in parallel
525                let mut bitmaps: Vec<_> = stream::iter(
526                    keys.into_iter()
527                        .map(|key| async move { self.load_bitmap(&key, None).await }),
528                )
529                .buffer_unordered(get_num_compute_intensive_cpus())
530                .try_collect()
531                .await?;
532
533                // Add null bitmap if needed
534                if has_null && !self.null_map.is_empty() {
535                    bitmaps.push(self.null_map.clone());
536                }
537
538                let result = if bitmaps.is_empty() {
539                    RowAddrTreeMap::default()
540                } else {
541                    // Convert Arc<RowAddrTreeMap> to &RowAddrTreeMap for union_all
542                    let bitmap_refs: Vec<_> = bitmaps.iter().map(|b| b.as_ref()).collect();
543                    RowAddrTreeMap::union_all(&bitmap_refs)
544                };
545
546                // If the query explicitly includes null, then nulls are TRUE (not NULL)
547                // Otherwise, nulls remain NULL (unknown)
548                let null_rows = if !has_null && !self.null_map.is_empty() {
549                    Some((*self.null_map).clone())
550                } else {
551                    None
552                };
553                (result, null_rows)
554            }
555            SargableQuery::IsNull() => {
556                metrics.record_comparisons(1);
557                // Querying FOR nulls - they are the TRUE result, not NULL result
558                ((*self.null_map).clone(), None)
559            }
560            SargableQuery::FullTextSearch(_) => {
561                return Err(Error::not_supported_source(
562                    "full text search is not supported for bitmap indexes".into(),
563                ));
564            }
565            SargableQuery::LikePrefix(_) => {
566                return Err(Error::not_supported_source(
567                    "LIKE prefix queries are not supported for bitmap indexes".into(),
568                ));
569            }
570        };
571
572        let selection = NullableRowAddrSet::new(row_ids, null_row_ids.unwrap_or_default());
573        Ok(SearchResult::Exact(selection))
574    }
575
576    fn can_remap(&self) -> bool {
577        true
578    }
579
580    /// Remap the row ids, creating a new remapped version of this index in `dest_store`
581    async fn remap(
582        &self,
583        mapping: &HashMap<u64, Option<u64>>,
584        dest_store: &dyn IndexStore,
585    ) -> Result<CreatedIndex> {
586        let state = self.load_bitmap_index_state().await?;
587        let remapped_state = BitmapIndexPlugin::remap_bitmap_state(state, mapping);
588        BitmapIndexPlugin::write_bitmap_index(remapped_state, dest_store, &self.value_type).await?;
589
590        Ok(CreatedIndex {
591            index_details: prost_types::Any::from_msg(&pbold::BitmapIndexDetails::default())
592                .unwrap(),
593            index_version: BITMAP_INDEX_VERSION,
594            files: Some(dest_store.list_files_with_sizes().await?),
595        })
596    }
597
598    /// Add the new data into the index, creating an updated version of the index in `dest_store`
599    async fn update(
600        &self,
601        new_data: SendableRecordBatchStream,
602        dest_store: &dyn IndexStore,
603        _old_data_filter: Option<super::OldIndexDataFilter>,
604    ) -> Result<CreatedIndex> {
605        let state = self.load_bitmap_index_state().await?;
606        BitmapIndexPlugin::do_train_bitmap_index(new_data, state, dest_store).await?;
607
608        Ok(CreatedIndex {
609            index_details: prost_types::Any::from_msg(&pbold::BitmapIndexDetails::default())
610                .unwrap(),
611            index_version: BITMAP_INDEX_VERSION,
612            files: Some(dest_store.list_files_with_sizes().await?),
613        })
614    }
615
616    fn update_criteria(&self) -> UpdateCriteria {
617        UpdateCriteria::only_new_data(TrainingCriteria::new(TrainingOrdering::None).with_row_id())
618    }
619
620    fn derive_index_params(&self) -> Result<ScalarIndexParams> {
621        Ok(ScalarIndexParams::for_builtin(BuiltinIndexType::Bitmap))
622    }
623}
624
625#[derive(Debug, Default)]
626pub struct BitmapIndexPlugin;
627
628impl BitmapIndexPlugin {
629    fn get_batch_from_arrays(
630        keys: Arc<dyn Array>,
631        binary_bitmaps: Arc<dyn Array>,
632    ) -> Result<RecordBatch> {
633        let schema = Arc::new(Schema::new(vec![
634            Field::new("keys", keys.data_type().clone(), true),
635            Field::new("bitmaps", binary_bitmaps.data_type().clone(), true),
636        ]));
637
638        let columns = vec![keys, binary_bitmaps];
639
640        Ok(RecordBatch::try_new(schema, columns)?)
641    }
642
643    async fn write_bitmap_index(
644        state: HashMap<ScalarValue, RowAddrTreeMap>,
645        index_store: &dyn IndexStore,
646        value_type: &DataType,
647    ) -> Result<()> {
648        Self::write_bitmap_index_with_extras(
649            state,
650            index_store,
651            value_type,
652            HashMap::new(),
653            Vec::new(),
654        )
655        .await
656    }
657
658    /// Writes a bitmap index and attaches extra metadata and global buffers.
659    pub(crate) async fn write_bitmap_index_with_extras(
660        state: HashMap<ScalarValue, RowAddrTreeMap>,
661        index_store: &dyn IndexStore,
662        value_type: &DataType,
663        mut metadata: HashMap<String, String>,
664        global_buffers: Vec<(String, Bytes)>,
665    ) -> Result<()> {
666        let num_bitmaps = state.len();
667        let schema = Arc::new(Schema::new(vec![
668            Field::new("keys", value_type.clone(), true),
669            Field::new("bitmaps", DataType::Binary, true),
670        ]));
671
672        let mut bitmap_index_file = index_store
673            .new_index_file(BITMAP_LOOKUP_NAME, schema)
674            .await?;
675
676        for (metadata_key, data) in global_buffers {
677            let buffer_idx = bitmap_index_file.add_global_buffer(data).await?;
678            metadata.insert(metadata_key, buffer_idx.to_string());
679        }
680
681        let mut cur_keys = Vec::new();
682        let mut cur_bitmaps = Vec::new();
683        let mut cur_bytes = 0;
684
685        for (key, bitmap) in state.into_iter() {
686            let mut bytes = Vec::new();
687            bitmap.serialize_into(&mut bytes).unwrap();
688            let bitmap_size = bytes.len();
689
690            if cur_bytes + bitmap_size > MAX_BITMAP_ARRAY_LENGTH {
691                let keys_array = ScalarValue::iter_to_array(cur_keys.clone().into_iter()).unwrap();
692                let mut binary_builder = BinaryBuilder::new();
693                for b in &cur_bitmaps {
694                    binary_builder.append_value(b);
695                }
696                let bitmaps_array = Arc::new(binary_builder.finish()) as Arc<dyn Array>;
697
698                let record_batch = Self::get_batch_from_arrays(keys_array, bitmaps_array)?;
699                bitmap_index_file.write_record_batch(record_batch).await?;
700
701                cur_keys.clear();
702                cur_bitmaps.clear();
703                cur_bytes = 0;
704            }
705
706            cur_keys.push(key);
707            cur_bitmaps.push(bytes);
708            cur_bytes += bitmap_size;
709        }
710
711        // Flush any remaining
712        if !cur_keys.is_empty() {
713            let keys_array = ScalarValue::iter_to_array(cur_keys).unwrap();
714            let mut binary_builder = BinaryBuilder::new();
715            for b in &cur_bitmaps {
716                binary_builder.append_value(b);
717            }
718            let bitmaps_array = Arc::new(binary_builder.finish()) as Arc<dyn Array>;
719
720            let record_batch = Self::get_batch_from_arrays(keys_array, bitmaps_array)?;
721            bitmap_index_file.write_record_batch(record_batch).await?;
722        }
723
724        // Finish file with metadata that allows lightweight statistics reads
725        let stats_json = serde_json::to_string(&BitmapStatistics { num_bitmaps })
726            .map_err(|e| Error::internal(format!("failed to serialize bitmap statistics: {e}")))?;
727        metadata.insert(INDEX_STATS_METADATA_KEY.to_string(), stats_json);
728
729        bitmap_index_file.finish_with_metadata(metadata).await?;
730
731        Ok(())
732    }
733
734    /// Builds bitmap index state from a `(value, row_id)` stream without writing it.
735    pub(crate) async fn build_bitmap_index_state(
736        mut data_source: SendableRecordBatchStream,
737        mut state: HashMap<ScalarValue, RowAddrTreeMap>,
738    ) -> Result<(HashMap<ScalarValue, RowAddrTreeMap>, DataType)> {
739        let value_type = data_source.schema().field(0).data_type().clone();
740        while let Some(batch) = data_source.try_next().await? {
741            let values = batch.column_by_name(VALUE_COLUMN_NAME).expect_ok()?;
742            let row_ids = batch.column_by_name(ROW_ID).expect_ok()?;
743            debug_assert_eq!(row_ids.data_type(), &DataType::UInt64);
744
745            let row_id_column = row_ids.as_any().downcast_ref::<UInt64Array>().unwrap();
746
747            for i in 0..values.len() {
748                let row_id = row_id_column.value(i);
749                let key = ScalarValue::try_from_array(values.as_ref(), i)?;
750                state.entry(key.clone()).or_default().insert(row_id);
751            }
752        }
753
754        Ok((state, value_type))
755    }
756
757    async fn do_train_bitmap_index(
758        data_source: SendableRecordBatchStream,
759        state: HashMap<ScalarValue, RowAddrTreeMap>,
760        index_store: &dyn IndexStore,
761    ) -> Result<()> {
762        let (state, value_type) = Self::build_bitmap_index_state(data_source, state).await?;
763        Self::write_bitmap_index(state, index_store, &value_type).await
764    }
765
766    pub async fn train_bitmap_index(
767        data: SendableRecordBatchStream,
768        index_store: &dyn IndexStore,
769    ) -> Result<()> {
770        // mapping from item to list of the row ids where it is present
771        let dictionary: HashMap<ScalarValue, RowAddrTreeMap> = HashMap::new();
772
773        Self::do_train_bitmap_index(data, dictionary, index_store).await
774    }
775
776    /// Remaps every bitmap in a materialized bitmap-index state using row-id mappings.
777    pub(crate) fn remap_bitmap_state(
778        state: HashMap<ScalarValue, RowAddrTreeMap>,
779        mapping: &HashMap<u64, Option<u64>>,
780    ) -> HashMap<ScalarValue, RowAddrTreeMap> {
781        state
782            .into_iter()
783            .map(|(key, bitmap)| {
784                let remapped_bitmap =
785                    RowAddrTreeMap::from_iter(bitmap.row_addrs().unwrap().filter_map(|addr| {
786                        let addr_as_u64 = u64::from(addr);
787                        mapping
788                            .get(&addr_as_u64)
789                            .copied()
790                            .unwrap_or(Some(addr_as_u64))
791                    }));
792                (key, remapped_bitmap)
793            })
794            .collect()
795    }
796}
797
798#[async_trait]
799impl ScalarIndexPlugin for BitmapIndexPlugin {
800    fn name(&self) -> &str {
801        "Bitmap"
802    }
803
804    fn new_training_request(
805        &self,
806        _params: &str,
807        field: &Field,
808    ) -> Result<Box<dyn TrainingRequest>> {
809        if field.data_type().is_nested() {
810            return Err(Error::invalid_input_source(
811                "A bitmap index can only be created on a non-nested field.".into(),
812            ));
813        }
814        Ok(Box::new(DefaultTrainingRequest::new(
815            TrainingCriteria::new(TrainingOrdering::None).with_row_id(),
816        )))
817    }
818
819    fn provides_exact_answer(&self) -> bool {
820        true
821    }
822
823    fn version(&self) -> u32 {
824        BITMAP_INDEX_VERSION
825    }
826
827    fn new_query_parser(
828        &self,
829        index_name: String,
830        _index_details: &prost_types::Any,
831    ) -> Option<Box<dyn ScalarQueryParser>> {
832        Some(Box::new(SargableQueryParser::new(index_name, false)))
833    }
834
835    async fn train_index(
836        &self,
837        data: SendableRecordBatchStream,
838        index_store: &dyn IndexStore,
839        _request: Box<dyn TrainingRequest>,
840        fragment_ids: Option<Vec<u32>>,
841        _progress: Arc<dyn crate::progress::IndexBuildProgress>,
842    ) -> Result<CreatedIndex> {
843        if fragment_ids.is_some() {
844            return Err(Error::invalid_input_source(
845                "Bitmap index does not support fragment training".into(),
846            ));
847        }
848
849        Self::train_bitmap_index(data, index_store).await?;
850        Ok(CreatedIndex {
851            index_details: prost_types::Any::from_msg(&pbold::BitmapIndexDetails::default())
852                .unwrap(),
853            index_version: BITMAP_INDEX_VERSION,
854            files: Some(index_store.list_files_with_sizes().await?),
855        })
856    }
857
858    /// Load an index from storage
859    async fn load_index(
860        &self,
861        index_store: Arc<dyn IndexStore>,
862        _index_details: &prost_types::Any,
863        frag_reuse_index: Option<Arc<FragReuseIndex>>,
864        cache: &LanceCache,
865    ) -> Result<Arc<dyn ScalarIndex>> {
866        Ok(BitmapIndex::load(index_store, frag_reuse_index, cache).await? as Arc<dyn ScalarIndex>)
867    }
868
869    async fn load_statistics(
870        &self,
871        index_store: Arc<dyn IndexStore>,
872        _index_details: &prost_types::Any,
873    ) -> Result<Option<serde_json::Value>> {
874        let reader = index_store.open_index_file(BITMAP_LOOKUP_NAME).await?;
875        if let Some(value) = reader.schema().metadata.get(INDEX_STATS_METADATA_KEY) {
876            let stats = serde_json::from_str(value).map_err(|e| {
877                Error::internal(format!("failed to parse bitmap statistics metadata: {e}"))
878            })?;
879            Ok(Some(stats))
880        } else {
881            Ok(None)
882        }
883    }
884}
885
886#[cfg(test)]
887pub mod tests {
888    use super::*;
889    use crate::metrics::NoOpMetricsCollector;
890    use crate::scalar::lance_format::LanceIndexStore;
891    use arrow_array::{RecordBatch, StringArray, UInt64Array, record_batch};
892    use arrow_schema::{DataType, Field, Schema};
893    use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
894    use futures::stream;
895    use lance_core::utils::mask::RowSetOps;
896    use lance_core::utils::{address::RowAddress, tempfile::TempObjDir};
897    use lance_io::object_store::ObjectStore;
898    use std::collections::HashMap;
899
900    #[tokio::test]
901    async fn test_bitmap_lazy_loading_and_cache() {
902        // Create a temporary directory for the index
903        let tmpdir = TempObjDir::default();
904        let store = Arc::new(LanceIndexStore::new(
905            Arc::new(ObjectStore::local()),
906            tmpdir.clone(),
907            Arc::new(LanceCache::no_cache()),
908        ));
909
910        // Create test data with low cardinality column
911        let colors = vec![
912            "red", "blue", "green", "red", "yellow", "blue", "red", "green", "blue", "yellow",
913            "red", "red", "blue", "green", "yellow",
914        ];
915
916        let row_ids = (0u64..15u64).collect::<Vec<_>>();
917
918        let schema = Arc::new(Schema::new(vec![
919            Field::new("value", DataType::Utf8, false),
920            Field::new("_rowid", DataType::UInt64, false),
921        ]));
922
923        let batch = RecordBatch::try_new(
924            schema.clone(),
925            vec![
926                Arc::new(StringArray::from(colors.clone())),
927                Arc::new(UInt64Array::from(row_ids.clone())),
928            ],
929        )
930        .unwrap();
931
932        let stream = stream::once(async move { Ok(batch) });
933        let stream = Box::pin(RecordBatchStreamAdapter::new(schema, stream));
934
935        // Train and write the bitmap index
936        BitmapIndexPlugin::train_bitmap_index(stream, store.as_ref())
937            .await
938            .unwrap();
939
940        // Create a cache with limited capacity
941        let cache = LanceCache::with_capacity(1024 * 1024); // 1MB cache
942
943        // Load the index (should only load metadata, not bitmaps)
944        let index = BitmapIndex::load(store.clone(), None, &cache)
945            .await
946            .unwrap();
947
948        assert_eq!(index.index_map.len(), 4); // 4 non-null unique values (red, blue, green, yellow)
949        assert!(index.null_map.is_empty()); // No nulls in test data
950
951        // Test 1: Search for "red"
952        let query = SargableQuery::Equals(ScalarValue::Utf8(Some("red".to_string())));
953        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
954
955        // Verify results
956        let expected_red_rows = vec![0u64, 3, 6, 10, 11];
957        if let SearchResult::Exact(row_ids) = result {
958            let mut actual: Vec<u64> = row_ids
959                .true_rows()
960                .row_addrs()
961                .unwrap()
962                .map(|id| id.into())
963                .collect();
964            actual.sort();
965            assert_eq!(actual, expected_red_rows);
966        } else {
967            panic!("Expected exact search result");
968        }
969
970        // Test 2: Search for "red" again - should hit cache
971        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
972        if let SearchResult::Exact(row_ids) = result {
973            let mut actual: Vec<u64> = row_ids
974                .true_rows()
975                .row_addrs()
976                .unwrap()
977                .map(|id| id.into())
978                .collect();
979            actual.sort();
980            assert_eq!(actual, expected_red_rows);
981        }
982
983        // Test 3: Range query
984        let query = SargableQuery::Range(
985            std::ops::Bound::Included(ScalarValue::Utf8(Some("blue".to_string()))),
986            std::ops::Bound::Included(ScalarValue::Utf8(Some("green".to_string()))),
987        );
988        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
989
990        let expected_range_rows = vec![1u64, 2, 5, 7, 8, 12, 13];
991        if let SearchResult::Exact(row_ids) = result {
992            let mut actual: Vec<u64> = row_ids
993                .true_rows()
994                .row_addrs()
995                .unwrap()
996                .map(|id| id.into())
997                .collect();
998            actual.sort();
999            assert_eq!(actual, expected_range_rows);
1000        }
1001
1002        // Test 3b: Inverted range query should return empty result
1003        let query = SargableQuery::Range(
1004            std::ops::Bound::Included(ScalarValue::Utf8(Some("green".to_string()))),
1005            std::ops::Bound::Included(ScalarValue::Utf8(Some("blue".to_string()))),
1006        );
1007        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
1008        if let SearchResult::Exact(row_ids) = result {
1009            assert!(row_ids.true_rows().is_empty());
1010        } else {
1011            panic!("Expected exact search result");
1012        }
1013
1014        // Test 4: IsIn query
1015        let query = SargableQuery::IsIn(vec![
1016            ScalarValue::Utf8(Some("red".to_string())),
1017            ScalarValue::Utf8(Some("yellow".to_string())),
1018        ]);
1019        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
1020
1021        let expected_in_rows = vec![0u64, 3, 4, 6, 9, 10, 11, 14];
1022        if let SearchResult::Exact(row_ids) = result {
1023            let mut actual: Vec<u64> = row_ids
1024                .true_rows()
1025                .row_addrs()
1026                .unwrap()
1027                .map(|id| id.into())
1028                .collect();
1029            actual.sort();
1030            assert_eq!(actual, expected_in_rows);
1031        }
1032    }
1033
1034    #[tokio::test]
1035    #[ignore]
1036    async fn test_big_bitmap_index() {
1037        // WARNING: This test allocates a huge state to force overflow over int32 on BinaryArray
1038        // You must run it only on a machine with enough resources (or skip it normally).
1039        use super::{BITMAP_LOOKUP_NAME, BitmapIndex};
1040        use crate::scalar::IndexStore;
1041        use crate::scalar::lance_format::LanceIndexStore;
1042        use arrow_schema::DataType;
1043        use datafusion_common::ScalarValue;
1044        use lance_core::cache::LanceCache;
1045        use lance_core::utils::mask::RowAddrTreeMap;
1046        use lance_io::object_store::ObjectStore;
1047        use std::collections::HashMap;
1048        use std::sync::Arc;
1049
1050        // Adjust these numbers so that:
1051        //     m * (serialized size per bitmap) > 2^31 bytes.
1052        //
1053        // For example, if we assume each bitmap serializes to ~1000 bytes,
1054        // you need m > 2.1e6.
1055        let m: u32 = 2_500_000;
1056        let per_bitmap_size = 1000; // assumed bytes per bitmap
1057
1058        let mut state = HashMap::new();
1059        for i in 0..m {
1060            // Create a bitmap that contains, say, 1000 row IDs.
1061            let bitmap = RowAddrTreeMap::from_iter(0..per_bitmap_size);
1062
1063            let key = ScalarValue::UInt32(Some(i));
1064            state.insert(key, bitmap);
1065        }
1066
1067        // Create a temporary store.
1068        let tmpdir = TempObjDir::default();
1069        let test_store = LanceIndexStore::new(
1070            Arc::new(ObjectStore::local()),
1071            tmpdir.clone(),
1072            Arc::new(LanceCache::no_cache()),
1073        );
1074
1075        // This call should never trigger a "byte array offset overflow" error since now the code supports
1076        // read by chunks
1077        let result =
1078            BitmapIndexPlugin::write_bitmap_index(state, &test_store, &DataType::UInt32).await;
1079
1080        assert!(
1081            result.is_ok(),
1082            "Failed to write bitmap index: {:?}",
1083            result.err()
1084        );
1085
1086        // Verify the index file exists
1087        let index_file = test_store.open_index_file(BITMAP_LOOKUP_NAME).await;
1088        assert!(
1089            index_file.is_ok(),
1090            "Failed to open index file: {:?}",
1091            index_file.err()
1092        );
1093        let index_file = index_file.unwrap();
1094
1095        // Print stats about the index file
1096        tracing::info!(
1097            "Index file contains {} rows in total",
1098            index_file.num_rows()
1099        );
1100
1101        // Load the index using BitmapIndex::load
1102        tracing::info!("Loading index from disk...");
1103        let loaded_index = BitmapIndex::load(Arc::new(test_store), None, &LanceCache::no_cache())
1104            .await
1105            .expect("Failed to load bitmap index");
1106
1107        // Verify the loaded index has the correct number of entries
1108        assert_eq!(
1109            loaded_index.index_map.len(),
1110            m as usize,
1111            "Loaded index has incorrect number of keys (expected {}, got {})",
1112            m,
1113            loaded_index.index_map.len()
1114        );
1115
1116        // Manually verify specific keys without using search()
1117        let test_keys = [0, m / 2, m - 1]; // Beginning, middle, and end
1118        for &key_val in &test_keys {
1119            let key = OrderableScalarValue(ScalarValue::UInt32(Some(key_val)));
1120            // Load the bitmap for this key
1121            let bitmap = loaded_index
1122                .load_bitmap(&key, None)
1123                .await
1124                .unwrap_or_else(|_| panic!("Key {} should exist", key_val));
1125
1126            // Convert RowAddrTreeMap to a vector for easier assertion
1127            let row_addrs: Vec<u64> = bitmap.row_addrs().unwrap().map(u64::from).collect();
1128
1129            // Verify length
1130            assert_eq!(
1131                row_addrs.len(),
1132                per_bitmap_size as usize,
1133                "Bitmap for key {} has wrong size",
1134                key_val
1135            );
1136
1137            // Verify first few and last few elements
1138            for i in 0..5.min(per_bitmap_size) {
1139                assert!(
1140                    row_addrs.contains(&i),
1141                    "Bitmap for key {} should contain row_id {}",
1142                    key_val,
1143                    i
1144                );
1145            }
1146
1147            for i in (per_bitmap_size - 5)..per_bitmap_size {
1148                assert!(
1149                    row_addrs.contains(&i),
1150                    "Bitmap for key {} should contain row_id {}",
1151                    key_val,
1152                    i
1153                );
1154            }
1155
1156            // Verify exact range
1157            let expected_range: Vec<u64> = (0..per_bitmap_size).collect();
1158            assert_eq!(
1159                row_addrs, expected_range,
1160                "Bitmap for key {} doesn't contain expected values",
1161                key_val
1162            );
1163
1164            tracing::info!(
1165                "✓ Verified bitmap for key {}: {} rows as expected",
1166                key_val,
1167                row_addrs.len()
1168            );
1169        }
1170
1171        tracing::info!("Test successful! Index properly contains {} keys", m);
1172    }
1173
1174    #[tokio::test]
1175    async fn test_bitmap_prewarm() {
1176        // Create a temporary directory for the index
1177        let tmpdir = TempObjDir::default();
1178        let store = Arc::new(LanceIndexStore::new(
1179            Arc::new(ObjectStore::local()),
1180            tmpdir.clone(),
1181            Arc::new(LanceCache::no_cache()),
1182        ));
1183
1184        // Create test data with low cardinality
1185        let colors = vec![
1186            "red", "blue", "green", "red", "yellow", "blue", "red", "green", "blue", "yellow",
1187            "red", "red", "blue", "green", "yellow",
1188        ];
1189
1190        let row_ids = (0u64..15u64).collect::<Vec<_>>();
1191
1192        let schema = Arc::new(Schema::new(vec![
1193            Field::new("value", DataType::Utf8, false),
1194            Field::new("_rowid", DataType::UInt64, false),
1195        ]));
1196
1197        let batch = RecordBatch::try_new(
1198            schema.clone(),
1199            vec![
1200                Arc::new(StringArray::from(colors.clone())),
1201                Arc::new(UInt64Array::from(row_ids.clone())),
1202            ],
1203        )
1204        .unwrap();
1205
1206        let stream = stream::once(async move { Ok(batch) });
1207        let stream = Box::pin(RecordBatchStreamAdapter::new(schema, stream));
1208
1209        // Train and write the bitmap index
1210        BitmapIndexPlugin::train_bitmap_index(stream, store.as_ref())
1211            .await
1212            .unwrap();
1213
1214        // Create a cache with metrics tracking
1215        let cache = LanceCache::with_capacity(1024 * 1024); // 1MB cache
1216
1217        // Load the index (should only load metadata, not bitmaps)
1218        let index = BitmapIndex::load(store.clone(), None, &cache)
1219            .await
1220            .unwrap();
1221
1222        // Verify no bitmaps are cached yet
1223        let cache_key_red = BitmapKey {
1224            value: OrderableScalarValue(ScalarValue::Utf8(Some("red".to_string()))),
1225        };
1226        let cache_key_blue = BitmapKey {
1227            value: OrderableScalarValue(ScalarValue::Utf8(Some("blue".to_string()))),
1228        };
1229
1230        assert!(
1231            cache
1232                .get_with_key::<BitmapKey>(&cache_key_red)
1233                .await
1234                .is_none()
1235        );
1236        assert!(
1237            cache
1238                .get_with_key::<BitmapKey>(&cache_key_blue)
1239                .await
1240                .is_none()
1241        );
1242
1243        // Call prewarm
1244        index.prewarm().await.unwrap();
1245
1246        // Verify all bitmaps are now cached
1247        assert!(
1248            cache
1249                .get_with_key::<BitmapKey>(&cache_key_red)
1250                .await
1251                .is_some()
1252        );
1253        assert!(
1254            cache
1255                .get_with_key::<BitmapKey>(&cache_key_blue)
1256                .await
1257                .is_some()
1258        );
1259
1260        // Verify cached bitmaps have correct content
1261        let cached_red = cache
1262            .get_with_key::<BitmapKey>(&cache_key_red)
1263            .await
1264            .unwrap();
1265        let red_rows: Vec<u64> = cached_red.row_addrs().unwrap().map(u64::from).collect();
1266        assert_eq!(red_rows, vec![0, 3, 6, 10, 11]);
1267
1268        // Call prewarm again - should be idempotent
1269        index.prewarm().await.unwrap();
1270
1271        // Verify cache still contains the same items
1272        let cached_red_2 = cache
1273            .get_with_key::<BitmapKey>(&cache_key_red)
1274            .await
1275            .unwrap();
1276        let red_rows_2: Vec<u64> = cached_red_2.row_addrs().unwrap().map(u64::from).collect();
1277        assert_eq!(red_rows_2, vec![0, 3, 6, 10, 11]);
1278    }
1279
1280    #[tokio::test]
1281    async fn test_remap_bitmap_with_null() {
1282        use arrow_array::UInt32Array;
1283
1284        // Create a temporary store.
1285        let tmpdir = TempObjDir::default();
1286        let test_store = Arc::new(LanceIndexStore::new(
1287            Arc::new(ObjectStore::local()),
1288            tmpdir.clone(),
1289            Arc::new(LanceCache::no_cache()),
1290        ));
1291
1292        // Create test data that simulates:
1293        // frag 1 - { 0: null, 1: null, 2: 1 }
1294        // frag 2 - { 0: 1, 1: 2, 2: 2 }
1295        // We'll create this data with specific row addresses
1296        let values = vec![
1297            None,       // row 0: null (will be at address (1,0))
1298            None,       // row 1: null (will be at address (1,1))
1299            Some(1u32), // row 2: 1    (will be at address (1,2))
1300            Some(1u32), // row 3: 1    (will be at address (2,0))
1301            Some(2u32), // row 4: 2    (will be at address (2,1))
1302            Some(2u32), // row 5: 2    (will be at address (2,2))
1303        ];
1304
1305        // Create row IDs with specific fragment addresses
1306        let row_ids: Vec<u64> = vec![
1307            RowAddress::new_from_parts(1, 0).into(),
1308            RowAddress::new_from_parts(1, 1).into(),
1309            RowAddress::new_from_parts(1, 2).into(),
1310            RowAddress::new_from_parts(2, 0).into(),
1311            RowAddress::new_from_parts(2, 1).into(),
1312            RowAddress::new_from_parts(2, 2).into(),
1313        ];
1314
1315        let schema = Arc::new(Schema::new(vec![
1316            Field::new("value", DataType::UInt32, true),
1317            Field::new("_rowid", DataType::UInt64, false),
1318        ]));
1319
1320        let batch = RecordBatch::try_new(
1321            schema.clone(),
1322            vec![
1323                Arc::new(UInt32Array::from(values)),
1324                Arc::new(UInt64Array::from(row_ids)),
1325            ],
1326        )
1327        .unwrap();
1328
1329        let stream = stream::once(async move { Ok(batch) });
1330        let stream = Box::pin(RecordBatchStreamAdapter::new(schema, stream));
1331
1332        // Create the bitmap index
1333        BitmapIndexPlugin::train_bitmap_index(stream, test_store.as_ref())
1334            .await
1335            .unwrap();
1336
1337        // Load the index
1338        let index = BitmapIndex::load(test_store.clone(), None, &LanceCache::no_cache())
1339            .await
1340            .expect("Failed to load bitmap index");
1341
1342        // Verify initial state
1343        assert_eq!(index.index_map.len(), 2); // 2 non-null values (1 and 2)
1344        assert!(!index.null_map.is_empty()); // Should have null values
1345
1346        // Create a remap that simulates compaction of frags 1 and 2 into frag 3
1347        let mut row_addr_map = HashMap::<u64, Option<u64>>::new();
1348        row_addr_map.insert(
1349            RowAddress::new_from_parts(1, 0).into(),
1350            Some(RowAddress::new_from_parts(3, 0).into()),
1351        );
1352        row_addr_map.insert(
1353            RowAddress::new_from_parts(1, 1).into(),
1354            Some(RowAddress::new_from_parts(3, 1).into()),
1355        );
1356        row_addr_map.insert(
1357            RowAddress::new_from_parts(1, 2).into(),
1358            Some(RowAddress::new_from_parts(3, 2).into()),
1359        );
1360        row_addr_map.insert(
1361            RowAddress::new_from_parts(2, 0).into(),
1362            Some(RowAddress::new_from_parts(3, 3).into()),
1363        );
1364        row_addr_map.insert(
1365            RowAddress::new_from_parts(2, 1).into(),
1366            Some(RowAddress::new_from_parts(3, 4).into()),
1367        );
1368        row_addr_map.insert(
1369            RowAddress::new_from_parts(2, 2).into(),
1370            Some(RowAddress::new_from_parts(3, 5).into()),
1371        );
1372
1373        // Perform remap
1374        index
1375            .remap(&row_addr_map, test_store.as_ref())
1376            .await
1377            .unwrap();
1378
1379        // Reload and check
1380        let reloaded_idx = BitmapIndex::load(test_store, None, &LanceCache::no_cache())
1381            .await
1382            .expect("Failed to load remapped bitmap index");
1383
1384        // Verify the null bitmap was remapped correctly
1385        let expected_null_addrs: Vec<u64> = vec![
1386            RowAddress::new_from_parts(3, 0).into(),
1387            RowAddress::new_from_parts(3, 1).into(),
1388        ];
1389        let actual_null_addrs: Vec<u64> = reloaded_idx
1390            .null_map
1391            .row_addrs()
1392            .unwrap()
1393            .map(u64::from)
1394            .collect();
1395        assert_eq!(
1396            actual_null_addrs, expected_null_addrs,
1397            "Null bitmap not remapped correctly"
1398        );
1399
1400        // Search for value 1 and verify remapped addresses
1401        let query = SargableQuery::Equals(ScalarValue::UInt32(Some(1)));
1402        let result = reloaded_idx
1403            .search(&query, &NoOpMetricsCollector)
1404            .await
1405            .unwrap();
1406        if let crate::scalar::SearchResult::Exact(row_ids) = result {
1407            let mut actual: Vec<u64> = row_ids
1408                .true_rows()
1409                .row_addrs()
1410                .unwrap()
1411                .map(u64::from)
1412                .collect();
1413            actual.sort();
1414            let expected: Vec<u64> = vec![
1415                RowAddress::new_from_parts(3, 2).into(),
1416                RowAddress::new_from_parts(3, 3).into(),
1417            ];
1418            assert_eq!(actual, expected, "Value 1 bitmap not remapped correctly");
1419        }
1420
1421        // Search for value 2 and verify remapped addresses
1422        let query = SargableQuery::Equals(ScalarValue::UInt32(Some(2)));
1423        let result = reloaded_idx
1424            .search(&query, &NoOpMetricsCollector)
1425            .await
1426            .unwrap();
1427        if let crate::scalar::SearchResult::Exact(row_ids) = result {
1428            let mut actual: Vec<u64> = row_ids
1429                .true_rows()
1430                .row_addrs()
1431                .unwrap()
1432                .map(u64::from)
1433                .collect();
1434            actual.sort();
1435            let expected: Vec<u64> = vec![
1436                RowAddress::new_from_parts(3, 4).into(),
1437                RowAddress::new_from_parts(3, 5).into(),
1438            ];
1439            assert_eq!(actual, expected, "Value 2 bitmap not remapped correctly");
1440        }
1441
1442        // Search for null values
1443        let query = SargableQuery::IsNull();
1444        let result = reloaded_idx
1445            .search(&query, &NoOpMetricsCollector)
1446            .await
1447            .unwrap();
1448        if let crate::scalar::SearchResult::Exact(row_ids) = result {
1449            let mut actual: Vec<u64> = row_ids
1450                .true_rows()
1451                .row_addrs()
1452                .unwrap()
1453                .map(u64::from)
1454                .collect();
1455            actual.sort();
1456            assert_eq!(
1457                actual, expected_null_addrs,
1458                "Null search results not correct"
1459            );
1460        }
1461    }
1462
1463    #[tokio::test]
1464    async fn test_bitmap_null_handling_in_queries() {
1465        // Test that bitmap index correctly returns null_list for queries
1466        let tmpdir = TempObjDir::default();
1467        let store = Arc::new(LanceIndexStore::new(
1468            Arc::new(ObjectStore::local()),
1469            tmpdir.clone(),
1470            Arc::new(LanceCache::no_cache()),
1471        ));
1472
1473        // Create test data: [0, 5, null]
1474        let batch = record_batch!(
1475            ("value", Int64, [Some(0), Some(5), None]),
1476            ("_rowid", UInt64, [0, 1, 2])
1477        )
1478        .unwrap();
1479        let schema = batch.schema();
1480        let stream = stream::once(async move { Ok(batch) });
1481        let stream = Box::pin(RecordBatchStreamAdapter::new(schema, stream));
1482
1483        // Train and write the bitmap index
1484        BitmapIndexPlugin::train_bitmap_index(stream, store.as_ref())
1485            .await
1486            .unwrap();
1487
1488        let cache = LanceCache::with_capacity(1024 * 1024);
1489        let index = BitmapIndex::load(store.clone(), None, &cache)
1490            .await
1491            .unwrap();
1492
1493        // Test 1: Search for value 5 - should return allow=[1], null=[2]
1494        let query = SargableQuery::Equals(ScalarValue::Int64(Some(5)));
1495        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
1496
1497        match result {
1498            SearchResult::Exact(row_ids) => {
1499                let actual_rows: Vec<u64> = row_ids
1500                    .true_rows()
1501                    .row_addrs()
1502                    .unwrap()
1503                    .map(u64::from)
1504                    .collect();
1505                assert_eq!(actual_rows, vec![1], "Should find row 1 where value == 5");
1506
1507                let null_row_ids = row_ids.null_rows();
1508                // Check that null_row_ids contains row 2
1509                assert!(!null_row_ids.is_empty(), "null_row_ids should be Some");
1510                let null_rows: Vec<u64> =
1511                    null_row_ids.row_addrs().unwrap().map(u64::from).collect();
1512                assert_eq!(null_rows, vec![2], "Should report row 2 as null");
1513            }
1514            _ => panic!("Expected Exact search result"),
1515        }
1516
1517        // Test 2: Search for null values - should return allow=[2], null=None
1518        let query = SargableQuery::IsNull();
1519        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
1520
1521        match result {
1522            SearchResult::Exact(row_addrs) => {
1523                let actual_rows: Vec<u64> = row_addrs
1524                    .true_rows()
1525                    .row_addrs()
1526                    .unwrap()
1527                    .map(u64::from)
1528                    .collect();
1529                assert_eq!(
1530                    actual_rows,
1531                    vec![2],
1532                    "IsNull should find row 2 where value is null"
1533                );
1534
1535                let null_row_ids = row_addrs.null_rows();
1536                // When querying FOR nulls, null_row_ids should be None (nulls are the TRUE result)
1537                assert!(
1538                    null_row_ids.is_empty(),
1539                    "null_row_ids should be None for IsNull query"
1540                );
1541            }
1542            _ => panic!("Expected Exact search result"),
1543        }
1544
1545        // Test 3: Range query - should return matching rows and null_list
1546        let query = SargableQuery::Range(
1547            std::ops::Bound::Included(ScalarValue::Int64(Some(0))),
1548            std::ops::Bound::Included(ScalarValue::Int64(Some(3))),
1549        );
1550        let result = index.search(&query, &NoOpMetricsCollector).await.unwrap();
1551
1552        match result {
1553            SearchResult::Exact(row_addrs) => {
1554                let actual_rows: Vec<u64> = row_addrs
1555                    .true_rows()
1556                    .row_addrs()
1557                    .unwrap()
1558                    .map(u64::from)
1559                    .collect();
1560                assert_eq!(actual_rows, vec![0], "Should find row 0 where value == 0");
1561
1562                // Should report row 2 as null
1563                let null_row_ids = row_addrs.null_rows();
1564                assert!(!null_row_ids.is_empty(), "null_row_ids should be Some");
1565                let null_rows: Vec<u64> =
1566                    null_row_ids.row_addrs().unwrap().map(u64::from).collect();
1567                assert_eq!(null_rows, vec![2], "Should report row 2 as null");
1568            }
1569            _ => panic!("Expected Exact search result"),
1570        }
1571    }
1572}