From 08b334ca4d98793418c7f5f69e6e6bcb5511807d Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Wed, 22 Jul 2026 22:42:38 -0700 Subject: [PATCH 1/4] feat: support segmented RTree indexes --- python/python/lance/dataset.py | 4 +- python/python/tests/test_geo.py | 54 ++++++ rust/lance-index/src/scalar/rtree.rs | 235 ++++++++++++++++++++++++++- rust/lance/src/index.rs | 20 +++ rust/lance/src/index/scalar.rs | 2 + rust/lance/src/index/scalar/rtree.rs | 73 +++++++++ 6 files changed, 379 insertions(+), 9 deletions(-) create mode 100644 rust/lance/src/index/scalar/rtree.rs diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py index c6b8bd9ebf3..602a248d07d 100644 --- a/python/python/lance/dataset.py +++ b/python/python/lance/dataset.py @@ -3308,6 +3308,7 @@ def _is_segment_native_scalar_index_type( "BITMAP", "INVERTED", "FTS", + "RTREE", "ZONEMAP", "BLOOMFILTER", } @@ -3320,6 +3321,7 @@ def _requires_uncommitted_scalar_index( return cls._normalized_index_type(index_type) in { "BTREE", "BITMAP", + "RTREE", "ZONEMAP", "BLOOMFILTER", } @@ -4294,7 +4296,7 @@ def create_index_uncommitted( 4. commit the final segment list with :meth:`commit_existing_index_segments` - BTREE, BITMAP, INVERTED, ZONEMAP, and BLOOMFILTER segments may + BTREE, BITMAP, INVERTED, RTREE, ZONEMAP, and BLOOMFILTER segments may be merged with :meth:`merge_existing_index_segments` before commit. Parameters are the same as :meth:`create_index`, with one additional requirement: diff --git a/python/python/tests/test_geo.py b/python/python/tests/test_geo.py index c011c2de3de..bc6f6449b81 100644 --- a/python/python/tests/test_geo.py +++ b/python/python/tests/test_geo.py @@ -153,3 +153,57 @@ def query(ds: lance.LanceDataset, has_index=False): table_with_index = query(ds, has_index=True) assert table_with_index == table_without_index + + +def test_rtree_segment_merge_and_commit(tmp_path: Path): + num_points = 120 + points_2d = points( + [ + np.arange(num_points, dtype=np.float64), + np.arange(num_points, dtype=np.float64), + ] + ) + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field(point("xy")).with_name("point"), + ] + ) + table = pa.Table.from_arrays( + [np.arange(num_points, dtype=np.int64), points_2d], schema=schema + ) + ds = lance.write_dataset( + table, + str(tmp_path / "segmented_rtree.lance"), + max_rows_per_file=40, + ) + fragments = ds.get_fragments() + segments = [ + ds.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree", + fragment_ids=[fragment.fragment_id], + ) + for fragment in fragments + ] + + merged = ds.merge_existing_index_segments(segments) + assert set(merged.fragment_ids) == {fragment.fragment_id for fragment in fragments} + ds = ds.commit_existing_index_segments("point_rtree", "point", [merged]) + + sql = """ + SELECT id, point + FROM dataset + WHERE St_Intersects(point, ST_GeomFromText('LINESTRING (10 10, 20 20)')) + """ + indexed = pa.Table.from_batches(ds.sql(sql).build().to_batch_records()) + assert indexed["id"].to_pylist() == list(range(10, 21)) + explain = ( + pa.Table.from_batches( + ds.sql("EXPLAIN ANALYZE " + sql).build().to_batch_records() + ) + .to_pandas() + .to_string() + ) + assert "ScalarIndexQuery" in explain diff --git a/rust/lance-index/src/scalar/rtree.rs b/rust/lance-index/src/scalar/rtree.rs index 0d6b54a02ca..600a509cf66 100644 --- a/rust/lance-index/src/scalar/rtree.rs +++ b/rust/lance-index/src/scalar/rtree.rs @@ -10,7 +10,8 @@ use crate::scalar::registry::{ use crate::scalar::rtree::sort::Sorter; use crate::scalar::{ AnyQuery, BuiltinIndexType, CreatedIndex, GeoQuery, IndexFile, IndexReader, IndexStore, - IndexWriter, RowIdRemapper, ScalarIndex, ScalarIndexParams, SearchResult, UpdateCriteria, + IndexWriter, OldIndexDataFilter, RowIdRemapper, ScalarIndex, ScalarIndexParams, SearchResult, + UpdateCriteria, }; use crate::{Index, IndexType, pb}; use arrow_array::UInt32Array; @@ -428,6 +429,136 @@ impl RTreeIndex { } } +fn filter_keeps_nothing(filter: &Option) -> bool { + match filter { + Some(OldIndexDataFilter::Fragments { to_keep, .. }) => to_keep.is_empty(), + Some(OldIndexDataFilter::RowIds(valid)) => valid.is_empty(), + None => false, + } +} + +fn filter_rtree_data( + data: SendableRecordBatchStream, + filter: OldIndexDataFilter, +) -> SendableRecordBatchStream { + let schema = data.schema(); + let filtered = data.map(move |batch_result| { + let batch = batch_result?; + let row_ids = batch + .column_by_name(ROW_ID) + .and_then(|column| column.as_any().downcast_ref::()) + .ok_or_else(|| Error::internal("expected UInt64Array for RTree row ids"))?; + let mask = filter.filter_row_ids(row_ids); + Ok(arrow_select::filter::filter_record_batch(&batch, &mask)?) + }); + Box::pin(RecordBatchStreamAdapter::new(schema, filtered)) +} + +fn remap_rtree_data( + data: SendableRecordBatchStream, + remapper: Arc, +) -> SendableRecordBatchStream { + let schema = data.schema(); + let remapped = data.map(move |batch_result| { + let batch = batch_result?; + Ok(remapper.remap_row_ids_record_batch(batch, 1)?) + }); + Box::pin(RecordBatchStreamAdapter::new(schema, remapped)) +} + +/// Merge caller-selected RTree segments into one self-contained segment. +pub async fn merge_rtree_indices( + source_indices: &[Arc], + dest_store: &dyn IndexStore, + old_data_filters: &[Option], +) -> Result { + if source_indices.is_empty() { + return Err(Error::invalid_input( + "merge_rtree_indices requires at least one source index", + )); + } + if source_indices.len() != old_data_filters.len() { + return Err(Error::invalid_input(format!( + "merge_rtree_indices received {} source indices but {} filters", + source_indices.len(), + old_data_filters.len() + ))); + } + + let first_contributing = source_indices + .iter() + .zip(old_data_filters) + .find(|(source, filter)| source.metadata.num_items > 0 && !filter_keeps_nothing(filter)) + .or_else(|| { + source_indices + .iter() + .zip(old_data_filters) + .find(|(_, filter)| !filter_keeps_nothing(filter)) + }) + .map(|(source, _)| source) + .unwrap_or(&source_indices[0]); + let page_size = first_contributing.metadata.page_size; + let mut data_streams = Vec::with_capacity(source_indices.len()); + let mut null_map = RowAddrTreeMap::new(); + let mut total_bbox = BoundingBox::new(); + + for (source, filter) in source_indices.iter().zip(old_data_filters) { + if filter_keeps_nothing(filter) { + continue; + } + if source.metadata.num_items > 0 && source.metadata.page_size != page_size { + return Err(Error::invalid_input(format!( + "cannot merge RTree segments with different page sizes: {} and {}", + page_size, source.metadata.page_size + ))); + } + if source.metadata.num_items > 0 { + total_bbox.add_rect(&source.metadata.bbox); + } + + let mut source_nulls = source.search_null(&NoOpMetricsCollector).await?; + if let Some(remapper) = &source.frag_reuse_index { + source_nulls = remapper.remap_row_addrs_tree_map(&source_nulls); + } + if let Some(filter) = filter { + filter.retain_old_rows(&mut source_nulls); + } + null_map |= &source_nulls; + + let mut data = source.as_ref().clone().into_data_stream().await?; + if let Some(remapper) = source.frag_reuse_index.clone() { + data = remap_rtree_data(data, remapper); + } + data_streams.push(match filter { + Some(filter) => filter_rtree_data(data, filter.clone()), + None => data, + }); + } + + let combined = Box::pin(RecordBatchStreamAdapter::new( + BBOX_ROWID_SCHEMA.clone(), + stream::select_all(data_streams), + )); + let tmpdir = Arc::new(TempDir::default()); + let spill_store = Arc::new(LanceIndexStore::new( + Arc::new(ObjectStore::local()), + tmpdir.obj_path(), + Arc::new(LanceCache::no_cache()), + )); + let (bbox_data, mut stats) = + RTreeIndexPlugin::process_and_analyze_bbox_stream(combined, page_size, spill_store).await?; + stats.null_map = null_map; + stats.total_bbox = total_bbox; + let files = + RTreeIndexPlugin::train_rtree_index(bbox_data, stats, page_size, dest_store).await?; + + Ok(CreatedIndex { + index_details: prost_types::Any::from_msg(&pb::RTreeIndexDetails::default())?, + index_version: RTREE_INDEX_VERSION, + files, + }) +} + impl DeepSizeOf for RTreeIndex { fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize { let mut total_size = 0; @@ -923,15 +1054,9 @@ impl BasicTrainer for RTreeIndexPlugin { data: SendableRecordBatchStream, index_store: &dyn IndexStore, request: Box, - fragment_ids: Option>, + _fragment_ids: Option>, _progress: Arc, ) -> Result { - if fragment_ids.is_some() { - return Err(Error::invalid_input_source( - "RTree index does not support fragment training".into(), - )); - } - Self::validate_schema(&data.schema())?; let request = request @@ -1012,6 +1137,7 @@ struct EncodedBatch { #[cfg(test)] mod tests { use super::*; + use crate::frag_reuse::{FragReuseIndex, FragReuseIndexDetails, FragReuseIndexHandle}; use crate::metrics::NoOpMetricsCollector; use crate::scalar::registry::VALUE_COLUMN_NAME; use arrow_array::ArrayRef; @@ -1178,6 +1304,99 @@ mod tests { assert_eq!(actual_nulls, expected_nulls); } + #[tokio::test] + async fn test_merge_rtree_indices_filters_rows_and_nulls() { + let point_type = PointType::new(Dimension::XY, Default::default()); + let mut first_builder = PointBuilder::new(point_type.clone()); + first_builder.push_point(Some(&geo_types::point!(x: 10.0, y: 10.0))); + first_builder.push_null(); + let first = first_builder.finish(); + + let mut second_builder = PointBuilder::new(point_type); + second_builder.push_point(Some(&geo_types::point!(x: 100.0, y: 100.0))); + second_builder.push_null(); + let second = second_builder.finish(); + let empty = PointBuilder::new(PointType::new(Dimension::XY, Default::default())).finish(); + + let (empty_index, _empty_store, _empty_tmpdir) = train_index(&empty, Some(8)).await; + let (first_index, _first_store, _first_tmpdir) = train_index(&first, Some(4)).await; + let (second_index, _second_store, _second_tmpdir) = train_index(&second, Some(4)).await; + let mut near_origin = BoundingBox::new(); + near_origin.add_rect(&Rect::new( + coord! { x: 9.0, y: 9.0 }, + coord! { x: 11.0, y: 11.0 }, + )); + let mut expected_geometry = RowAddrTreeMap::new(); + expected_geometry.insert(0); + assert_eq!( + first_index + .search_bbox(near_origin, &NoOpMetricsCollector) + .await + .unwrap(), + expected_geometry + ); + let merged_tmpdir = TempObjDir::default(); + let merged_store = Arc::new(LanceIndexStore::new( + Arc::new(ObjectStore::local()), + merged_tmpdir.clone(), + Arc::new(LanceCache::no_cache()), + )); + + let remapped_geometry = RowAddress::new_from_parts(2, 0).into(); + let remapped_null = RowAddress::new_from_parts(2, 1).into(); + expected_geometry = RowAddrTreeMap::new(); + expected_geometry.insert(remapped_geometry); + let remapper = FragReuseIndexHandle(Arc::new(FragReuseIndex::new( + uuid::Uuid::new_v4(), + vec![HashMap::from([ + (0, Some(remapped_geometry)), + (1, Some(remapped_null)), + ])], + FragReuseIndexDetails { versions: vec![] }, + ))); + let mut first_index = first_index.as_ref().clone(); + first_index.frag_reuse_index = Some(Arc::new(remapper)); + + let mut keep_first_rows = RowAddrTreeMap::new(); + keep_first_rows.insert(remapped_geometry); + keep_first_rows.insert(remapped_null); + merge_rtree_indices( + &[empty_index, Arc::new(first_index), second_index], + merged_store.as_ref(), + &[ + None, + Some(OldIndexDataFilter::RowIds(keep_first_rows)), + Some(OldIndexDataFilter::RowIds(RowAddrTreeMap::new())), + ], + ) + .await + .unwrap(); + + let merged = RTreeIndex::load(merged_store, None, &LanceCache::no_cache()) + .await + .unwrap(); + assert_eq!(merged.metadata.num_items, 1); + assert!( + merged.metadata.bbox.rect_intersects(&near_origin), + "merged bounds {:?} do not intersect {:?}", + merged.metadata.bbox, + near_origin + ); + assert_eq!( + merged + .search_bbox(near_origin, &NoOpMetricsCollector) + .await + .unwrap(), + expected_geometry + ); + let mut expected_null = RowAddrTreeMap::new(); + expected_null.insert(remapped_null); + assert_eq!( + merged.search_null(&NoOpMetricsCollector).await.unwrap(), + expected_null + ); + } + #[tokio::test] async fn test_update_and_search() { fn gen_data(num_items: u32, frag_id: u32, nulls_addrs: &mut RowAddrTreeMap) -> RectArray { diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index 65a2692414e..d53243d7889 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -531,6 +531,14 @@ fn segment_has_label_list_details(segment: &IndexMetadata) -> bool { .is_some_and(|details| details.type_url.ends_with("LabelListIndexDetails")) } +#[cfg(feature = "geo")] +fn segment_has_rtree_details(segment: &IndexMetadata) -> bool { + segment + .index_details + .as_ref() + .is_some_and(|details| details.type_url.ends_with("RTreeIndexDetails")) +} + // Cache keys for different index types #[derive(Debug, Clone)] pub(crate) struct LegacyVectorIndexCacheKey<'a> { @@ -1412,6 +1420,10 @@ impl DatasetIndexExt for Dataset { let all_fmindex = source_segments.iter().all(segment_has_fmindex_details); let all_zonemap = source_segments.iter().all(segment_has_zonemap_details); let all_label_list = source_segments.iter().all(segment_has_label_list_details); + #[cfg(feature = "geo")] + let all_rtree = source_segments.iter().all(segment_has_rtree_details); + #[cfg(not(feature = "geo"))] + let all_rtree = false; if !all_vector && !all_inverted && !all_bitmap @@ -1420,6 +1432,7 @@ impl DatasetIndexExt for Dataset { && !all_fmindex && !all_zonemap && !all_label_list + && !all_rtree { return Err(Error::invalid_input( "merge_existing_index_segments requires all segments to have the same supported index type" @@ -1446,6 +1459,13 @@ impl DatasetIndexExt for Dataset { crate::index::scalar::label_list::merge_segments(self, source_segments).await? } else if all_zonemap { crate::index::scalar::zonemap::merge_segments(self, source_segments).await? + } else if all_rtree { + #[cfg(feature = "geo")] + { + crate::index::scalar::rtree::merge_segments(self, source_segments).await? + } + #[cfg(not(feature = "geo"))] + unreachable!("RTree segments require the geo feature") } else { crate::index::scalar::btree::merge_segments(self, source_segments).await? }; diff --git a/rust/lance/src/index/scalar.rs b/rust/lance/src/index/scalar.rs index 1566b156036..ae4d61058b8 100644 --- a/rust/lance/src/index/scalar.rs +++ b/rust/lance/src/index/scalar.rs @@ -10,6 +10,8 @@ pub(crate) mod btree; pub(crate) mod fmindex; pub(crate) mod inverted; pub(crate) mod label_list; +#[cfg(feature = "geo")] +pub(crate) mod rtree; pub(crate) mod zonemap; pub use inverted::{load_segment_details, load_segment_params, load_segments}; diff --git a/rust/lance/src/index/scalar/rtree.rs b/rust/lance/src/index/scalar/rtree.rs new file mode 100644 index 00000000000..53e301b9757 --- /dev/null +++ b/rust/lance/src/index/scalar/rtree.rs @@ -0,0 +1,73 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +use std::sync::Arc; + +use lance_index::metrics::NoOpMetricsCollector; +use lance_index::scalar::index_files_to_table; +use lance_index::scalar::lance_format::LanceIndexStore; +use lance_index::scalar::rtree::RTreeIndex; +use lance_table::format::IndexMetadata; +use uuid::Uuid; + +use crate::{Dataset, Error, Result, dataset::index::LanceIndexStoreExt}; + +/// Merge one caller-defined group of source RTree segments into a single segment. +pub(in crate::index) async fn merge_segments( + dataset: &Dataset, + segments: Vec, +) -> Result { + if segments.is_empty() { + return Err(Error::index("No segment metadata was provided".to_string())); + } + + let field_id = *segments[0].fields.first().ok_or_else(|| { + Error::invalid_input(format!( + "CreateIndex: segment {} is missing field ids", + segments[0].uuid + )) + })?; + let field_path = dataset.schema().field_path(field_id)?; + let segment_refs = segments.iter().collect::>(); + let (fragment_bitmap, old_data_filters) = + crate::index::append::build_per_segment_filters(dataset, &segment_refs).await?; + + let mut source_indices = Vec::with_capacity(segments.len()); + for segment in &segments { + let scalar_index = + super::open_scalar_index(dataset, &field_path, segment, &NoOpMetricsCollector).await?; + let rtree_index = scalar_index + .as_any() + .downcast_ref::() + .ok_or_else(|| { + Error::index(format!( + "merge_existing_index_segments: expected RTree segment {}, got {:?}", + segment.uuid, + scalar_index.index_type() + )) + })?; + source_indices.push(Arc::new(rtree_index.clone())); + } + + let new_uuid = Uuid::new_v4(); + let new_store = LanceIndexStore::from_dataset_for_new(dataset, &new_uuid)?; + let created_index = lance_index::scalar::rtree::merge_rtree_indices( + &source_indices, + &new_store, + &old_data_filters, + ) + .await?; + + Ok(IndexMetadata { + uuid: new_uuid, + fields: vec![field_id], + dataset_version: dataset.manifest.version, + fragment_bitmap: Some(fragment_bitmap), + index_details: Some(Arc::new(created_index.index_details)), + index_version: created_index.index_version as i32, + created_at: Some(chrono::Utc::now()), + base_id: None, + files: Some(index_files_to_table(created_index.files)), + ..segments[0].clone() + }) +} From 0b5b104d7e3cd418e11af2958f4370043374c633 Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Wed, 22 Jul 2026 23:07:57 -0700 Subject: [PATCH 2/4] fix: preserve segmented index watermarks --- rust/lance/src/index/scalar/rtree.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/rust/lance/src/index/scalar/rtree.rs b/rust/lance/src/index/scalar/rtree.rs index 53e301b9757..d19376da3ab 100644 --- a/rust/lance/src/index/scalar/rtree.rs +++ b/rust/lance/src/index/scalar/rtree.rs @@ -28,6 +28,11 @@ pub(in crate::index) async fn merge_segments( )) })?; let field_path = dataset.schema().field_path(field_id)?; + let dataset_version = segments + .iter() + .map(|segment| segment.dataset_version) + .min() + .unwrap_or(dataset.manifest.version); let segment_refs = segments.iter().collect::>(); let (fragment_bitmap, old_data_filters) = crate::index::append::build_per_segment_filters(dataset, &segment_refs).await?; @@ -61,7 +66,7 @@ pub(in crate::index) async fn merge_segments( Ok(IndexMetadata { uuid: new_uuid, fields: vec![field_id], - dataset_version: dataset.manifest.version, + dataset_version, fragment_bitmap: Some(fragment_bitmap), index_details: Some(Arc::new(created_index.index_details)), index_version: created_index.index_version as i32, From 00c5a8d5f3acac1c5a910157c46b7141850742b0 Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Thu, 23 Jul 2026 14:02:19 -0700 Subject: [PATCH 3/4] fix(index): preserve segmented index provenance --- java/lance-jni/src/blocking_dataset.rs | 2 + python/src/dataset.rs | 2 + python/src/indices.rs | 2 + rust/lance-index/src/scalar/rtree.rs | 54 ++++++++++-- rust/lance/src/dataset/index.rs | 2 + rust/lance/src/index.rs | 109 +++++++++++++++++++++++-- rust/lance/src/index/api.rs | 77 +++++++++++++---- rust/lance/src/index/create.rs | 2 + 8 files changed, 222 insertions(+), 28 deletions(-) diff --git a/java/lance-jni/src/blocking_dataset.rs b/java/lance-jni/src/blocking_dataset.rs index caf837b371a..767f11de4df 100644 --- a/java/lance-jni/src/blocking_dataset.rs +++ b/java/lance-jni/src/blocking_dataset.rs @@ -1228,8 +1228,10 @@ fn index_metadata_to_segment(metadata: &IndexMetadata) -> Result { Ok(IndexSegment::new( metadata.uuid, fragment_bitmap, + metadata.fields.iter().copied(), index_details, metadata.index_version, + metadata.dataset_version, )) } diff --git a/python/src/dataset.rs b/python/src/dataset.rs index d163c65d6aa..b80fa3e51ae 100644 --- a/python/src/dataset.rs +++ b/python/src/dataset.rs @@ -608,8 +608,10 @@ fn index_metadata_to_segment(metadata: IndexMetadata) -> PyResult Ok(IndexSegment::new( metadata.uuid, fragment_bitmap.iter(), + metadata.fields, index_details, metadata.index_version, + metadata.dataset_version, )) } diff --git a/python/src/indices.rs b/python/src/indices.rs index 1d5c476d795..cdf52b37fa8 100644 --- a/python/src/indices.rs +++ b/python/src/indices.rs @@ -550,12 +550,14 @@ async fn do_load_shuffled_vectors( .as_ref() .expect("vector metadata should include fragment coverage") .iter(), + metadata.fields.iter().copied(), metadata .index_details .as_ref() .expect("vector metadata should include index details") .clone(), metadata.index_version, + metadata.dataset_version, ); ds.commit_existing_index_segments(index_name, column, vec![segment]) .await diff --git a/rust/lance-index/src/scalar/rtree.rs b/rust/lance-index/src/scalar/rtree.rs index 600a509cf66..d5eb762ec5b 100644 --- a/rust/lance-index/src/scalar/rtree.rs +++ b/rust/lance-index/src/scalar/rtree.rs @@ -55,6 +55,15 @@ const RTREE_INDEX_VERSION: u32 = 0; const RTREE_PAGES_NAME: &str = "page_data.lance"; const RTREE_NULLS_NAME: &str = "nulls.lance"; +fn validate_page_size(page_size: u32) -> Result<()> { + if page_size < 2 { + return Err(Error::invalid_input( + "RTree page_size must be at least 2".to_string(), + )); + } + Ok(()) +} + static BBOX_FIELD: LazyLock> = LazyLock::new(|| { let bbox_type = RectType::new(Dimension::XY, Default::default()); Arc::new(bbox_type.to_field("bbox", false)) @@ -141,7 +150,9 @@ pub struct RTreeMetadata { impl RTreeMetadata { pub fn new(page_size: u32, num_pages: u64, num_items: usize, bbox: BoundingBox) -> Self { let page_offsets = Self::calculate_page_offsets(num_items, page_size); - debug_assert_eq!(page_offsets.len(), num_pages as usize); + if page_size >= 2 { + debug_assert_eq!(page_offsets.len(), num_pages as usize); + } Self { page_size, num_pages, @@ -152,6 +163,9 @@ impl RTreeMetadata { } fn calculate_page_offsets(num_items: usize, page_size: u32) -> Vec { + if page_size < 2 { + return Vec::new(); + } let mut page_offsets = vec![]; let mut cur_level_items = num_items; let mut cur_offset = 0; @@ -282,6 +296,7 @@ impl RTreeIndex { ) -> Result> { let pages_reader = store.open_index_file(RTREE_PAGES_NAME).await?; let metadata = RTreeMetadata::from(&pages_reader.schema().metadata); + validate_page_size(metadata.page_size)?; let nulls_reader = store.open_index_file(RTREE_NULLS_NAME).await?; Ok(Arc::new(Self { @@ -461,12 +476,22 @@ fn remap_rtree_data( let schema = data.schema(); let remapped = data.map(move |batch_result| { let batch = batch_result?; + // The row ID is column 1 in BBOX_ROWID_SCHEMA. Ok(remapper.remap_row_ids_record_batch(batch, 1)?) }); Box::pin(RecordBatchStreamAdapter::new(schema, remapped)) } /// Merge caller-selected RTree segments into one self-contained segment. +/// +/// Each source may supply a filter for rows that are still live. The merged index recomputes its +/// bounding box from the retained, remapped rows and preserves retained null row IDs. +/// +/// # Examples +/// +/// ```ignore +/// let merged = merge_rtree_indices(&segments, destination.as_ref(), &filters).await?; +/// ``` pub async fn merge_rtree_indices( source_indices: &[Arc], dest_store: &dyn IndexStore, @@ -498,9 +523,9 @@ pub async fn merge_rtree_indices( .map(|(source, _)| source) .unwrap_or(&source_indices[0]); let page_size = first_contributing.metadata.page_size; + validate_page_size(page_size)?; let mut data_streams = Vec::with_capacity(source_indices.len()); let mut null_map = RowAddrTreeMap::new(); - let mut total_bbox = BoundingBox::new(); for (source, filter) in source_indices.iter().zip(old_data_filters) { if filter_keeps_nothing(filter) { @@ -512,10 +537,6 @@ pub async fn merge_rtree_indices( page_size, source.metadata.page_size ))); } - if source.metadata.num_items > 0 { - total_bbox.add_rect(&source.metadata.bbox); - } - let mut source_nulls = source.search_null(&NoOpMetricsCollector).await?; if let Some(remapper) = &source.frag_reuse_index { source_nulls = remapper.remap_row_addrs_tree_map(&source_nulls); @@ -548,7 +569,6 @@ pub async fn merge_rtree_indices( let (bbox_data, mut stats) = RTreeIndexPlugin::process_and_analyze_bbox_stream(combined, page_size, spill_store).await?; stats.null_map = null_map; - stats.total_bbox = total_bbox; let files = RTreeIndexPlugin::train_rtree_index(bbox_data, stats, page_size, dest_store).await?; @@ -955,6 +975,7 @@ impl RTreeIndexPlugin { store: &dyn IndexStore, page_size: u32, ) -> Result { + validate_page_size(page_size)?; let mut page_idx: u64 = 0; let mut writer = store .new_index_file(RTREE_PAGES_NAME, RTREE_PAGE_SCHEMA.clone()) @@ -1046,6 +1067,9 @@ impl BasicTrainer for RTreeIndexPlugin { _field: &ArrowField, ) -> Result> { let params = serde_json::from_str::(params)?; + if let Some(page_size) = params.page_size { + validate_page_size(page_size)?; + } Ok(Box::new(RTreeTrainingRequest::new(params))) } @@ -1067,6 +1091,7 @@ impl BasicTrainer for RTreeIndexPlugin { .parameters .page_size .unwrap_or(DEFAULT_RTREE_PAGE_SIZE); + validate_page_size(page_size)?; let bbox_data = Self::convert_bbox_stream(data)?; let tmpdir = Arc::new(TempDir::default()); @@ -1152,6 +1177,17 @@ mod tests { RTreeMetadata::calculate_page_offsets(num_items, page_size).len() as u64 } + #[test] + fn test_rejects_page_size_that_cannot_reduce_tree_levels() { + let plugin = RTreeIndexPlugin; + let field = ArrowField::new("geometry", DataType::Null, true); + let error = plugin + .new_training_request(r#"{"page_size":1}"#, &field) + .err() + .unwrap(); + assert!(error.to_string().contains("page_size must be at least 2")); + } + fn convert_bbox_rowid_batch_stream( geo_array: &dyn GeoArrowArray, row_id_array: ArrayRef, @@ -1376,6 +1412,10 @@ mod tests { .await .unwrap(); assert_eq!(merged.metadata.num_items, 1); + assert_eq!(merged.metadata.bbox.minx(), 10.0); + assert_eq!(merged.metadata.bbox.miny(), 10.0); + assert_eq!(merged.metadata.bbox.maxx(), 10.0); + assert_eq!(merged.metadata.bbox.maxy(), 10.0); assert!( merged.metadata.bbox.rect_intersects(&near_origin), "merged bounds {:?} do not intersect {:?}", diff --git a/rust/lance/src/dataset/index.rs b/rust/lance/src/dataset/index.rs index 29c82075fec..c8f23e9720d 100644 --- a/rust/lance/src/dataset/index.rs +++ b/rust/lance/src/dataset/index.rs @@ -330,12 +330,14 @@ mod tests { .as_ref() .expect("test segment metadata should have fragment coverage") .iter(), + segment.fields.iter().copied(), segment .index_details .as_ref() .expect("test segment metadata should have index details") .clone(), segment.index_version, + segment.dataset_version, ) }) .collect::>(); diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index d53243d7889..c8e37a5ab98 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -146,6 +146,22 @@ pub(crate) async fn build_index_metadata_from_segments( let mut seen_segment_ids = HashSet::with_capacity(segments.len()); let mut covered_fragments = RoaringBitmap::new(); for segment in &segments { + if segment.dataset_version() > dataset.manifest.version { + return Err(Error::invalid_input(format!( + "CreateIndex: segment {} was built at future dataset version {} (current version {})", + segment.uuid(), + segment.dataset_version(), + dataset.manifest.version + ))); + } + if segment.fields() != [field_id] { + return Err(Error::invalid_input(format!( + "CreateIndex: segment {} was built for fields {:?}, expected [{}]", + segment.uuid(), + segment.fields(), + field_id + ))); + } if !seen_segment_ids.insert(segment.uuid()) { return Err(Error::invalid_input(format!( "CreateIndex: duplicate segment uuid {} for index '{}'", @@ -164,16 +180,14 @@ pub(crate) async fn build_index_metadata_from_segments( let mut new_indices = Vec::with_capacity(segments.len()); for segment in segments { - let dataset_version = segment - .dataset_version() - .unwrap_or(dataset.manifest.version); - let (uuid, fragment_bitmap, index_details, index_version) = segment.into_parts(); + let (uuid, fragment_bitmap, fields, index_details, index_version, dataset_version) = + segment.into_parts(); let is_inverted_index = index_details.type_url.ends_with("InvertedIndexDetails"); if is_inverted_index { let metadata = IndexMetadata { uuid, name: index_name.to_string(), - fields: vec![field_id], + fields: fields.clone(), dataset_version, fragment_bitmap: Some(fragment_bitmap.clone()), index_details: Some(index_details.clone()), @@ -193,7 +207,7 @@ pub(crate) async fn build_index_metadata_from_segments( new_indices.push(IndexMetadata { uuid, name: index_name.to_string(), - fields: vec![field_id], + fields, dataset_version, fragment_bitmap: Some(fragment_bitmap), index_details: Some(index_details), @@ -1465,7 +1479,9 @@ impl DatasetIndexExt for Dataset { crate::index::scalar::rtree::merge_segments(self, source_segments).await? } #[cfg(not(feature = "geo"))] - unreachable!("RTree segments require the geo feature") + return Err(Error::not_supported( + "RTree segment merge requires the `geo` feature".to_string(), + )); } else { crate::index::scalar::btree::merge_segments(self, source_segments).await? }; @@ -3010,12 +3026,14 @@ mod tests { .as_ref() .expect("test segment metadata should have fragment coverage") .iter(), + metadata.fields.iter().copied(), metadata .index_details .as_ref() .expect("test segment metadata should have index details") .clone(), metadata.index_version, + metadata.dataset_version, ) } @@ -7191,6 +7209,83 @@ mod tests { assert!(err.to_string().contains("at least one index segment")); } + #[tokio::test] + async fn test_commit_existing_index_segments_rejects_wrong_field_provenance() { + use lance_datagen::{BatchCount, RowCount, array}; + + let test_dir = tempfile::tempdir().unwrap(); + let reader = lance_datagen::gen_batch() + .col("id", array::step::()) + .col( + "vector", + array::rand_vec::(8.into()), + ) + .into_reader_rows(RowCount::from(10), BatchCount::from(1)); + let mut dataset = Dataset::write(reader, test_dir.path().to_str().unwrap(), None) + .await + .unwrap(); + + let vector_field_id = dataset.schema().field("vector").unwrap().id; + let mut metadata = write_vector_segment_metadata( + &dataset, + "vector_idx", + vector_field_id, + Uuid::new_v4(), + [0_u32], + b"segment", + ) + .await; + metadata.fields = vec![dataset.schema().field("id").unwrap().id]; + + let error = dataset + .commit_existing_index_segments( + "vector_idx", + "vector", + vec![segment_from_metadata(&metadata)], + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("was built for fields")); + } + + #[tokio::test] + async fn test_commit_existing_index_segments_rejects_future_dataset_version() { + use lance_datagen::{BatchCount, RowCount, array}; + + let test_dir = tempfile::tempdir().unwrap(); + let reader = lance_datagen::gen_batch() + .col( + "vector", + array::rand_vec::(8.into()), + ) + .into_reader_rows(RowCount::from(10), BatchCount::from(1)); + let mut dataset = Dataset::write(reader, test_dir.path().to_str().unwrap(), None) + .await + .unwrap(); + + let field_id = dataset.schema().field("vector").unwrap().id; + let mut metadata = write_vector_segment_metadata( + &dataset, + "vector_idx", + field_id, + Uuid::new_v4(), + [0_u32], + b"segment", + ) + .await; + metadata.dataset_version = dataset.manifest.version + 1; + + let error = dataset + .commit_existing_index_segments( + "vector_idx", + "vector", + vec![segment_from_metadata(&metadata)], + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("future dataset version")); + } + #[tokio::test] async fn test_commit_existing_index_segments_rejects_overlapping_fragment_coverage() { use lance_datagen::{BatchCount, RowCount, array}; diff --git a/rust/lance/src/index/api.rs b/rust/lance/src/index/api.rs index 29cb10fea04..9cf7507203d 100644 --- a/rust/lance/src/index/api.rs +++ b/rust/lance/src/index/api.rs @@ -15,40 +15,45 @@ use crate::{Error, Result}; /// A single physical segment of a logical index. /// /// Each segment is stored independently and will become one manifest entry when committed. -/// The logical index identity (name / target column / dataset version) is provided separately -/// by the commit API. +/// The logical index name is provided separately by the commit API, while physical field and +/// dataset-version provenance travel with the segment. #[derive(Debug, Clone, PartialEq)] pub struct IndexSegment { /// Unique ID of the physical segment. uuid: Uuid, /// The fragments covered by this segment. fragment_bitmap: RoaringBitmap, + /// Field IDs whose physical values are encoded in this segment. + fields: Vec, /// Metadata specific to the index type. index_details: Arc, /// The on-disk index version for this segment. index_version: i32, /// Dataset version at which this segment's physical contents were built. - dataset_version: Option, + dataset_version: u64, } impl IndexSegment { - /// Create a fully described segment with the given UUID, fragment coverage, and index - /// metadata. - pub fn new( + /// Create a fully described segment with its physical build provenance. + pub fn new( uuid: Uuid, fragment_bitmap: I, + fields: F, index_details: Arc, index_version: i32, + dataset_version: u64, ) -> Self where I: IntoIterator, + F: IntoIterator, { Self { uuid, fragment_bitmap: fragment_bitmap.into_iter().collect(), + fields: fields.into_iter().collect(), index_details, index_version, - dataset_version: None, + dataset_version, } } @@ -62,6 +67,11 @@ impl IndexSegment { &self.fragment_bitmap } + /// Return the field IDs whose values are encoded in this segment. + pub fn fields(&self) -> &[i32] { + &self.fields + } + /// Return the serialized index details for this segment. pub fn index_details(&self) -> &Arc { &self.index_details @@ -72,18 +82,29 @@ impl IndexSegment { self.index_version } - /// Return the source dataset version when this segment came from index metadata. - pub fn dataset_version(&self) -> Option { + /// Return the source dataset version for this segment. + pub fn dataset_version(&self) -> u64 { self.dataset_version } /// Consume the segment and return its component parts. - pub fn into_parts(self) -> (Uuid, RoaringBitmap, Arc, i32) { + pub fn into_parts( + self, + ) -> ( + Uuid, + RoaringBitmap, + Vec, + Arc, + i32, + u64, + ) { ( self.uuid, self.fragment_bitmap, + self.fields, self.index_details, self.index_version, + self.dataset_version, ) } } @@ -115,14 +136,14 @@ impl IntoIndexSegment for IndexMetadata { )) })?; - let mut segment = IndexSegment::new( + Ok(IndexSegment::new( self.uuid, fragment_bitmap.iter(), + self.fields, index_details, self.index_version, - ); - segment.dataset_version = Some(self.dataset_version); - Ok(segment) + self.dataset_version, + )) } } @@ -283,3 +304,31 @@ pub trait DatasetIndexExt { with_vector: bool, ) -> Result; } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_index_metadata_conversion_preserves_provenance() { + let metadata = IndexMetadata { + uuid: Uuid::new_v4(), + name: "test".to_string(), + fields: vec![3, 7], + dataset_version: 42, + fragment_bitmap: Some(RoaringBitmap::from_iter([1, 2])), + index_details: Some(Arc::new(prost_types::Any { + type_url: "example.IndexDetails".to_string(), + value: vec![1, 2, 3], + })), + index_version: 5, + created_at: None, + base_id: None, + files: None, + }; + + let segment = metadata.into_index_segment().unwrap(); + assert_eq!(segment.fields(), [3, 7]); + assert_eq!(segment.dataset_version(), 42); + } +} diff --git a/rust/lance/src/index/create.rs b/rust/lance/src/index/create.rs index e8129257c67..f6e5a7729f1 100644 --- a/rust/lance/src/index/create.rs +++ b/rust/lance/src/index/create.rs @@ -3110,8 +3110,10 @@ mod tests { vec![IndexSegment::new( uuid, dataset.fragment_bitmap.as_ref().clone(), + [dataset.schema().field("vector").unwrap().id], Arc::new(vector_index_details(¶ms)), IndexType::IvfHnswFlat.version(), + dataset.manifest.version, )], ) .await From db65cc542b7eed51f49b348c86b82a59f1211121 Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Fri, 24 Jul 2026 00:41:16 -0700 Subject: [PATCH 4/4] fix(index): harden segmented RTree safety --- python/python/lance/dataset.py | 4 +- python/python/tests/test_geo.py | 235 ++++++++++++++++++++++++++- rust/lance-index/src/scalar/rtree.rs | 37 ++++- rust/lance/src/dataset/overlay.rs | 103 ++++++++++-- rust/lance/src/dataset/scanner.rs | 21 ++- rust/lance/src/index.rs | 134 ++++++++++++++- rust/lance/src/index/api.rs | 4 + rust/lance/src/index/create.rs | 47 ++++-- rust/lance/src/index/scalar/rtree.rs | 21 ++- 9 files changed, 558 insertions(+), 48 deletions(-) diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py index 602a248d07d..d8115cb4a45 100644 --- a/python/python/lance/dataset.py +++ b/python/python/lance/dataset.py @@ -4279,8 +4279,8 @@ def create_index_uncommitted( Create one segment without publishing it and return its metadata. This is the public distributed-build API for vector, BTREE scalar, - canonical bitmap scalar, INVERTED scalar, ZONEMAP scalar, and - BLOOMFILTER scalar index construction. Unlike + canonical bitmap scalar, INVERTED scalar, RTREE scalar, ZONEMAP scalar, + and BLOOMFILTER scalar index construction. Unlike :meth:`create_index`, this method does not publish the index into the dataset manifest. Instead, it writes one segment under ``_indices//`` and returns the resulting diff --git a/python/python/tests/test_geo.py b/python/python/tests/test_geo.py index bc6f6449b81..c2dd599e432 100644 --- a/python/python/tests/test_geo.py +++ b/python/python/tests/test_geo.py @@ -19,6 +19,19 @@ ) +def _query_point_ids(dataset: lance.LanceDataset, wkt: str) -> list[int]: + sql = f""" + SELECT id, point + FROM dataset + WHERE St_Intersects(point, ST_GeomFromText('{wkt}')) + """ + return [ + value + for batch in dataset.sql(sql).build().to_batch_records() + for value in batch.column("id").to_pylist() + ] + + def test_geo_types(tmp_path: Path): uri = str(tmp_path / "test_geo_types.lance") # Points @@ -178,6 +191,7 @@ def test_rtree_segment_merge_and_commit(tmp_path: Path): max_rows_per_file=40, ) fragments = ds.get_fragments() + assert len(fragments) == 3 segments = [ ds.create_index_uncommitted( column="point", @@ -195,10 +209,10 @@ def test_rtree_segment_merge_and_commit(tmp_path: Path): sql = """ SELECT id, point FROM dataset - WHERE St_Intersects(point, ST_GeomFromText('LINESTRING (10 10, 20 20)')) + WHERE St_Intersects(point, ST_GeomFromText('LINESTRING (10 10, 110 110)')) """ indexed = pa.Table.from_batches(ds.sql(sql).build().to_batch_records()) - assert indexed["id"].to_pylist() == list(range(10, 21)) + assert indexed["id"].to_pylist() == list(range(10, 111)) explain = ( pa.Table.from_batches( ds.sql("EXPLAIN ANALYZE " + sql).build().to_batch_records() @@ -207,3 +221,220 @@ def test_rtree_segment_merge_and_commit(tmp_path: Path): .to_string() ) assert "ScalarIndexQuery" in explain + + +def test_staged_rtree_after_rewrite_columns(tmp_path: Path): + uri = str(tmp_path / "stale_rtree.lance") + point_type = point("xy") + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field(point_type).with_name("point"), + ] + ) + dataset = lance.write_dataset( + pa.Table.from_arrays( + [ + pa.array([0, 1], type=pa.int64()), + points([np.array([0.0, 1.0]), np.array([0.0, 1.0])]), + ], + schema=schema, + ), + uri, + ) + segment = dataset.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree", + fragment_ids=[0], + ) + + update_schema = pa.schema( + [ + pa.field("_rowid", pa.uint64()), + pa.field(point_type).with_name("point"), + ] + ) + update = pa.Table.from_arrays( + [ + pa.array([0], type=pa.uint64()), + points([np.array([10.0]), np.array([10.0])]), + ], + schema=update_schema, + ) + fragment, fields = dataset.get_fragment(0).update_columns(update) + updated = lance.LanceDataset.commit( + uri, + lance.LanceOperation.Update( + updated_fragments=[fragment], + fields_modified=fields, + ), + read_version=dataset.version, + ) + + assert _query_point_ids(updated, "POINT (10 10)") == [0] + committed = updated.commit_existing_index_segments( + "point_rtree", + "point", + [segment], + ) + assert _query_point_ids(committed, "POINT (10 10)") == [0] + + +def test_rtree_rejects_distributed_uuid_reuse(tmp_path: Path): + uri = str(tmp_path / "uuid_reuse.lance") + num_points = 120 + point_type = point("xy") + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field(point_type).with_name("point"), + ] + ) + dataset = lance.write_dataset( + pa.Table.from_arrays( + [ + pa.array(range(num_points), type=pa.int64()), + points( + [ + np.arange(num_points, dtype=np.float64), + np.arange(num_points, dtype=np.float64), + ] + ), + ], + schema=schema, + ), + uri, + max_rows_per_file=40, + ) + dataset.create_scalar_index("point", "RTREE") + index_uuid = dataset.describe_indices()[0].segments[0].uuid + + with pytest.raises( + ValueError, + match="index_uuid is no longer accepted for RTree distributed index builds", + ): + dataset.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree_reuse", + fragment_ids=[0], + index_uuid=index_uuid, + ) + + assert _query_point_ids( + lance.dataset(uri), + "LINESTRING (100 100, 110 110)", + ) == list(range(100, 111)) + + +def test_rtree_merge_all_deleted_stable_row_ids(tmp_path: Path): + uri = str(tmp_path / "all_deleted.lance") + point_type = point("xy") + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field(point_type).with_name("point"), + ] + ) + dataset = lance.write_dataset( + pa.Table.from_arrays( + [ + pa.array([0, 1], type=pa.int64()), + points([np.array([0.0, 1.0]), np.array([0.0, 1.0])]), + ], + schema=schema, + ), + uri, + enable_stable_row_ids=True, + ) + segment = dataset.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree", + fragment_ids=[0], + ) + + dataset.delete("true") + merged = dataset.merge_existing_index_segments([segment]) + assert merged.fragment_ids == set() + committed = dataset.commit_existing_index_segments( + "point_rtree", + "point", + [merged], + ) + assert _query_point_ids(committed, "POINT (0 0)") == [] + + +def test_rtree_merge_preserves_newer_fragment_coverage(tmp_path: Path): + uri = str(tmp_path / "mixed_versions.lance") + point_type = point("xy") + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field(point_type).with_name("point"), + ] + ) + + def batch(start: int, stop: int) -> pa.Table: + values = np.arange(start, stop, dtype=np.float64) + return pa.Table.from_arrays( + [ + pa.array(range(start, stop), type=pa.int64()), + points([values, values]), + ], + schema=schema, + ) + + dataset = lance.write_dataset(batch(0, 40), uri) + first = dataset.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree", + fragment_ids=[0], + ) + dataset = lance.write_dataset(batch(40, 80), uri, mode="append") + second = dataset.create_index_uncommitted( + column="point", + index_type="RTREE", + name="point_rtree", + fragment_ids=[1], + ) + + merged = dataset.merge_existing_index_segments([first, second]) + assert merged.fragment_ids == {0, 1} + assert merged.dataset_version == dataset.version + + update_schema = pa.schema( + [ + pa.field("_rowid", pa.uint64()), + pa.field(point_type).with_name("point"), + ] + ) + update = pa.Table.from_arrays( + [ + pa.array([1 << 32], type=pa.uint64()), + points([np.array([100.0]), np.array([100.0])]), + ], + schema=update_schema, + ) + fragment, fields = dataset.get_fragment(1).update_columns(update) + dataset = lance.LanceDataset.commit( + uri, + lance.LanceOperation.Update( + updated_fragments=[fragment], + fields_modified=fields, + ), + read_version=dataset.version, + ) + + committed = dataset.commit_existing_index_segments( + "point_rtree", + "point", + [merged], + ) + assert committed.describe_indices()[0].segments[0].fragment_ids == {0} + assert _query_point_ids( + committed, + "POINT (100 100)", + ) == [40] diff --git a/rust/lance-index/src/scalar/rtree.rs b/rust/lance-index/src/scalar/rtree.rs index d5eb762ec5b..a83f2250ac5 100644 --- a/rust/lance-index/src/scalar/rtree.rs +++ b/rust/lance-index/src/scalar/rtree.rs @@ -64,6 +64,15 @@ fn validate_page_size(page_size: u32) -> Result<()> { Ok(()) } +fn validate_stored_page_size(page_size: u32, num_items: usize) -> Result<()> { + if page_size == 0 || (page_size == 1 && num_items > 1) { + return Err(Error::invalid_input(format!( + "stored RTree page_size {page_size} cannot represent {num_items} items" + ))); + } + Ok(()) +} + static BBOX_FIELD: LazyLock> = LazyLock::new(|| { let bbox_type = RectType::new(Dimension::XY, Default::default()); Arc::new(bbox_type.to_field("bbox", false)) @@ -296,7 +305,7 @@ impl RTreeIndex { ) -> Result> { let pages_reader = store.open_index_file(RTREE_PAGES_NAME).await?; let metadata = RTreeMetadata::from(&pages_reader.schema().metadata); - validate_page_size(metadata.page_size)?; + validate_stored_page_size(metadata.page_size, metadata.num_items)?; let nulls_reader = store.open_index_file(RTREE_NULLS_NAME).await?; Ok(Arc::new(Self { @@ -489,8 +498,22 @@ fn remap_rtree_data( /// /// # Examples /// -/// ```ignore -/// let merged = merge_rtree_indices(&segments, destination.as_ref(), &filters).await?; +/// ```no_run +/// use std::sync::Arc; +/// +/// use lance_core::Result; +/// use lance_index::scalar::OldIndexDataFilter; +/// use lance_index::scalar::lance_format::LanceIndexStore; +/// use lance_index::scalar::rtree::{RTreeIndex, merge_rtree_indices}; +/// +/// async fn merge( +/// segments: &[Arc], +/// destination: &LanceIndexStore, +/// filters: &[Option], +/// ) -> Result<()> { +/// merge_rtree_indices(segments, destination, filters).await?; +/// Ok(()) +/// } /// ``` pub async fn merge_rtree_indices( source_indices: &[Arc], @@ -1188,6 +1211,14 @@ mod tests { assert!(error.to_string().contains("page_size must be at least 2")); } + #[test] + fn test_stored_page_size_preserves_single_item_compatibility() { + assert!(validate_stored_page_size(1, 1).is_ok()); + assert!(validate_stored_page_size(1, 0).is_ok()); + assert!(validate_stored_page_size(1, 2).is_err()); + assert!(validate_stored_page_size(0, 0).is_err()); + } + fn convert_bbox_rowid_batch_stream( geo_array: &dyn GeoArrowArray, row_id_array: ArrayRef, diff --git a/rust/lance/src/dataset/overlay.rs b/rust/lance/src/dataset/overlay.rs index 6afd8d56eca..83d4bc17938 100644 --- a/rust/lance/src/dataset/overlay.rs +++ b/rust/lance/src/dataset/overlay.rs @@ -65,6 +65,7 @@ pub fn overlay_exclusion_offsets( overlays: &[DataOverlayFile], indexed_field_ids: &[i32], index_version: u64, + schema: &Schema, ) -> Result { let mut excluded = RoaringBitmap::new(); for overlay in overlays { @@ -72,7 +73,21 @@ pub fn overlay_exclusion_offsets( continue; } for (field_pos, field_id) in overlay.data_file.fields.iter().enumerate() { - if indexed_field_ids.contains(field_id) { + let overlay_ancestry = schema.field_ancestry_by_id(*field_id); + let affects_index = indexed_field_ids.iter().any(|indexed_field_id| { + indexed_field_id == field_id + || overlay_ancestry.as_ref().is_some_and(|ancestry| { + ancestry + .iter() + .any(|ancestor| ancestor.id == *indexed_field_id) + }) + || schema + .field_ancestry_by_id(*indexed_field_id) + .is_some_and(|ancestry| { + ancestry.iter().any(|ancestor| ancestor.id == *field_id) + }) + }); + if affects_index { excluded |= &*overlay.coverage_for_field(field_pos)?; } } @@ -87,6 +102,7 @@ fn stale_offsets_for_fragment( fragment: &Fragment, fields: &[i32], index_version: u64, + schema: &Schema, ) -> Result { if fragment .overlays @@ -95,7 +111,7 @@ fn stale_offsets_for_fragment( { return Ok(RoaringBitmap::new()); } - overlay_exclusion_offsets(&fragment.overlays, fields, index_version) + overlay_exclusion_offsets(&fragment.overlays, fields, index_version, schema) } // A missing `fragment_bitmap` means the index predates fragment-bitmap tracking; treat it as @@ -126,13 +142,14 @@ pub fn collect_overlay_stale_frags( segment: &IndexMetadata, overlaid_frags: &HashMap, stale: &mut RoaringBitmap, + schema: &Schema, ) -> Result<()> { let coverage = segment.fragment_bitmap.as_ref(); for (&frag_id, fragment) in overlaid_frags { if stale.contains(frag_id) || !covers_fragment(coverage, frag_id) { continue; } - if !stale_offsets_for_fragment(fragment, &segment.fields, segment.dataset_version)? + if !stale_offsets_for_fragment(fragment, &segment.fields, segment.dataset_version, schema)? .is_empty() { stale.insert(frag_id); @@ -152,6 +169,7 @@ pub fn collect_overlay_stale_rows_for_segment( segment: &IndexMetadata, overlaid_frags: &HashMap, stale: &mut HashMap, + schema: &Schema, ) -> Result<()> { let coverage = segment.fragment_bitmap.as_ref(); for (&frag_id, fragment) in overlaid_frags { @@ -159,7 +177,7 @@ pub fn collect_overlay_stale_rows_for_segment( continue; } let excluded = - stale_offsets_for_fragment(fragment, &segment.fields, segment.dataset_version)?; + stale_offsets_for_fragment(fragment, &segment.fields, segment.dataset_version, schema)?; if !excluded.is_empty() { *stale.entry(frag_id).or_default() |= &excluded; } @@ -888,6 +906,19 @@ mod tests { RoaringBitmap::from_iter(offsets) } + fn flat_test_schema() -> Schema { + use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema}; + + let mut schema = Schema::try_from(&ArrowSchema::new( + (0..5) + .map(|id| ArrowField::new(format!("field_{id}"), DataType::Int32, true)) + .collect::>(), + )) + .unwrap(); + schema.set_field_id(None); + schema + } + /// Physical offsets for a contiguous range `[start, start + len)`. fn offsets(start: u32, len: usize) -> Vec { (start..start + len as u32).collect() @@ -1096,7 +1127,8 @@ mod tests { DataType::Struct(outer_fields), true, )]); - let schema = Schema::try_from(&arrow_schema).unwrap(); + let mut schema = Schema::try_from(&arrow_schema).unwrap(); + schema.set_field_id(None); let middle = StructArray::from(vec![ ( Arc::new(ArrowField::new("a", DataType::Int32, true)), @@ -1196,17 +1228,18 @@ mod tests { #[test] fn test_exclusion_offsets_version_gate() { + let schema = flat_test_schema(); // index built at version 5; only overlays committed > 5 are excluded. let overlays = vec![ dense_overlay(vec![3], [0, 1], 4), dense_overlay(vec![3], [2, 7], 6), ]; - let excluded = overlay_exclusion_offsets(&overlays, &[3], 5).unwrap(); + let excluded = overlay_exclusion_offsets(&overlays, &[3], 5, &schema).unwrap(); assert_eq!(excluded, bitmap([2, 7])); // An overlay exactly at the index version is already incorporated. let overlays = vec![dense_overlay(vec![3], [9], 5)]; assert!( - overlay_exclusion_offsets(&overlays, &[3], 5) + overlay_exclusion_offsets(&overlays, &[3], 5, &schema) .unwrap() .is_empty() ); @@ -1214,23 +1247,51 @@ mod tests { #[test] fn test_exclusion_offsets_is_field_aware() { + let schema = flat_test_schema(); // An overlay touching only an unrelated field excludes nothing. let overlays = vec![dense_overlay(vec![2], [0, 1, 2], 9)]; assert!( - overlay_exclusion_offsets(&overlays, &[3], 1) + overlay_exclusion_offsets(&overlays, &[3], 1, &schema) .unwrap() .is_empty() ); // The union spans only the indexed fields the overlay actually carries. let overlays = vec![dense_overlay(vec![2, 3], [4], 9)]; assert_eq!( - overlay_exclusion_offsets(&overlays, &[3], 1).unwrap(), + overlay_exclusion_offsets(&overlays, &[3], 1, &schema).unwrap(), bitmap([4]) ); } + #[test] + fn test_exclusion_offsets_matches_nested_fields() { + let (schema, _) = nested_struct(); + let outer = &schema.fields[0]; + let middle = &outer.children[0]; + let a = &middle.children[0]; + let b = &middle.children[1]; + + let overlays = vec![dense_overlay(vec![a.id], [1], 9)]; + assert_eq!( + overlay_exclusion_offsets(&overlays, &[outer.id], 1, &schema).unwrap(), + bitmap([1]) + ); + assert!( + overlay_exclusion_offsets(&overlays, &[b.id], 1, &schema) + .unwrap() + .is_empty() + ); + + let overlays = vec![dense_overlay(vec![middle.id], [2], 9)]; + assert_eq!( + overlay_exclusion_offsets(&overlays, &[a.id], 1, &schema).unwrap(), + bitmap([2]) + ); + } + #[test] fn test_exclusion_offsets_sparse_per_field() { + let schema = flat_test_schema(); // Sparse overlay: field 2 covers {2,3}, field 4 covers {1}. let overlay = DataOverlayFile { data_file: DataFile::new_legacy_from_fields("o.lance", vec![2, 4], None), @@ -1240,23 +1301,24 @@ mod tests { let overlays = vec![overlay]; // Only the bitmap for the indexed field (4) contributes. assert_eq!( - overlay_exclusion_offsets(&overlays, &[4], 1).unwrap(), + overlay_exclusion_offsets(&overlays, &[4], 1, &schema).unwrap(), bitmap([1]) ); assert_eq!( - overlay_exclusion_offsets(&overlays, &[2], 1).unwrap(), + overlay_exclusion_offsets(&overlays, &[2], 1, &schema).unwrap(), bitmap([2, 3]) ); } #[test] fn test_exclusion_offsets_unions_multiple_overlays() { + let schema = flat_test_schema(); let overlays = vec![ dense_overlay(vec![3], [1], 6), dense_overlay(vec![3], [4, 5], 7), ]; assert_eq!( - overlay_exclusion_offsets(&overlays, &[3], 1).unwrap(), + overlay_exclusion_offsets(&overlays, &[3], 1, &schema).unwrap(), bitmap([1, 4, 5]) ); } @@ -1290,13 +1352,15 @@ mod tests { #[test] fn test_collect_frags_missing_bitmap_covers_all() { + let schema = flat_test_schema(); // A segment with no fragment_bitmap (legacy index predating bitmap tracking) must treat // every overlaid fragment as covered so stale rows can't leak past the index unmasked. let fragment = fragment_with_overlay(3, dense_overlay(vec![3], [1, 2], 9)); let overlaid: HashMap = HashMap::from([(3u32, &fragment)]); let mut stale = RoaringBitmap::new(); - collect_overlay_stale_frags(&segment(vec![3], 1, None), &overlaid, &mut stale).unwrap(); + collect_overlay_stale_frags(&segment(vec![3], 1, None), &overlaid, &mut stale, &schema) + .unwrap(); assert_eq!(stale, bitmap([3]), "missing bitmap must cover fragment 3"); // A present bitmap that excludes fragment 3 leaves it untouched. @@ -1305,6 +1369,7 @@ mod tests { &segment(vec![3], 1, Some(bitmap([0]))), &overlaid, &mut stale, + &schema, ) .unwrap(); assert!( @@ -1318,6 +1383,7 @@ mod tests { &segment(vec![3], 1, Some(bitmap([3]))), &overlaid, &mut stale, + &schema, ) .unwrap(); assert_eq!(stale, bitmap([3])); @@ -1325,13 +1391,19 @@ mod tests { #[test] fn test_collect_rows_missing_bitmap_covers_all() { + let schema = flat_test_schema(); // Same covers-all guarantee at row-level granularity. let fragment = fragment_with_overlay(3, dense_overlay(vec![3], [1, 2], 9)); let overlaid: HashMap = HashMap::from([(3u32, &fragment)]); let mut stale = HashMap::new(); - collect_overlay_stale_rows_for_segment(&segment(vec![3], 1, None), &overlaid, &mut stale) - .unwrap(); + collect_overlay_stale_rows_for_segment( + &segment(vec![3], 1, None), + &overlaid, + &mut stale, + &schema, + ) + .unwrap(); assert_eq!( stale.get(&3), Some(&bitmap([1, 2])), @@ -1344,6 +1416,7 @@ mod tests { &segment(vec![3], 1, Some(bitmap([0]))), &overlaid, &mut stale, + &schema, ) .unwrap(); assert!( diff --git a/rust/lance/src/dataset/scanner.rs b/rust/lance/src/dataset/scanner.rs index 804129ad690..0496f0c942e 100644 --- a/rust/lance/src/dataset/scanner.rs +++ b/rust/lance/src/dataset/scanner.rs @@ -4345,7 +4345,12 @@ impl Scanner { ) .await?; for segment in &segments { - collect_overlay_stale_rows_for_segment(segment, &overlaid_frags, &mut stale)?; + collect_overlay_stale_rows_for_segment( + segment, + &overlaid_frags, + &mut stale, + self.dataset.schema(), + )?; } } Ok(stale) @@ -4372,7 +4377,12 @@ impl Scanner { } let mut stale: HashMap = HashMap::new(); for segment in segments { - collect_overlay_stale_rows_for_segment(segment, &overlaid_frags, &mut stale)?; + collect_overlay_stale_rows_for_segment( + segment, + &overlaid_frags, + &mut stale, + self.dataset.schema(), + )?; } Ok(stale) } @@ -4405,7 +4415,12 @@ impl Scanner { let overlaid_frags = overlaid_fragments(target_fragments); let mut stale_frag_ids = RoaringBitmap::new(); for seg in &segments { - collect_overlay_stale_frags(seg, &overlaid_frags, &mut stale_frag_ids)?; + collect_overlay_stale_frags( + seg, + &overlaid_frags, + &mut stale_frag_ids, + self.dataset.schema(), + )?; } if stale_frag_ids.is_empty() { diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index c8e37a5ab98..8b871001acf 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -131,11 +131,109 @@ fn validate_segment_metadata(index_name: &str, segments: &[IndexMetadata]) -> Re Ok(()) } +fn collect_subtree_field_ids(field: &Field, field_ids: &mut HashSet) { + field_ids.insert(field.id); + for child in &field.children { + collect_subtree_field_ids(child, field_ids); + } +} + +fn fragment_field_paths<'a>( + fragment: &'a Fragment, + indexed_field_ids: &HashSet, +) -> HashMap { + fragment + .files + .iter() + .flat_map(|file| { + file.fields + .iter() + .filter(|field_id| indexed_field_ids.contains(field_id)) + .map(|field_id| (*field_id, file.path.as_str())) + }) + .collect() +} + +async fn prune_stale_segment_coverage( + dataset: &Dataset, + field_id: i32, + segments: &mut [IndexSegment], + prune_historically_missing: bool, + prune_newer_overlays: bool, +) -> Result<()> { + let field = dataset.schema().field_by_id(field_id).ok_or_else(|| { + Error::invalid_input(format!( + "CreateIndex: field id {field_id} does not exist in the current schema" + )) + })?; + let mut indexed_field_ids = HashSet::new(); + collect_subtree_field_ids(field, &mut indexed_field_ids); + + let current_fragments = dataset + .fragments() + .iter() + .map(|fragment| (fragment.id as u32, fragment)) + .collect::>(); + let historical_versions = segments + .iter() + .map(IndexSegment::dataset_version) + .filter(|version| *version < dataset.manifest.version) + .collect::>(); + + for version in historical_versions { + let historical = dataset.checkout_version(version).await.map_err(|error| { + Error::invalid_input(format!( + "CreateIndex: cannot validate segment coverage built at dataset version {version}: {error}" + )) + })?; + let historical_fragments = historical + .fragments() + .iter() + .map(|fragment| (fragment.id as u32, fragment)) + .collect::>(); + + for segment in segments + .iter_mut() + .filter(|segment| segment.dataset_version() == version) + { + let stale_fragments = segment + .fragment_bitmap() + .iter() + .filter(|fragment_id| { + let Some(historical_fragment) = historical_fragments.get(fragment_id) else { + return prune_historically_missing; + }; + let Some(current_fragment) = current_fragments.get(fragment_id) else { + return true; + }; + let changed_files = + fragment_field_paths(historical_fragment, &indexed_field_ids) + != fragment_field_paths(current_fragment, &indexed_field_ids); + let changed_overlays = prune_newer_overlays + && current_fragment.overlays.iter().any(|overlay| { + overlay.committed_version > version + && overlay + .data_file + .fields + .iter() + .any(|field_id| indexed_field_ids.contains(field_id)) + }); + changed_files || changed_overlays + }) + .collect::>(); + for fragment_id in stale_fragments { + segment.fragment_bitmap_mut().remove(fragment_id); + } + } + } + Ok(()) +} + pub(crate) async fn build_index_metadata_from_segments( dataset: &Dataset, index_name: &str, field_id: i32, - segments: Vec, + mut segments: Vec, ) -> Result> { if segments.is_empty() { return Err(Error::invalid_input( @@ -178,6 +276,8 @@ pub(crate) async fn build_index_metadata_from_segments( covered_fragments |= segment.fragment_bitmap().clone(); } + prune_stale_segment_coverage(dataset, field_id, &mut segments, false, false).await?; + let mut new_indices = Vec::with_capacity(segments.len()); for segment in segments { let (uuid, fragment_bitmap, fields, index_details, index_version, dataset_version) = @@ -545,7 +645,6 @@ fn segment_has_label_list_details(segment: &IndexMetadata) -> bool { .is_some_and(|details| details.type_url.ends_with("LabelListIndexDetails")) } -#[cfg(feature = "geo")] fn segment_has_rtree_details(segment: &IndexMetadata) -> bool { segment .index_details @@ -1404,9 +1503,18 @@ impl DatasetIndexExt for Dataset { async fn merge_existing_index_segments( &self, - source_segments: Vec, + mut source_segments: Vec, ) -> Result { validate_segment_metadata("uncommitted", &source_segments)?; + if let Some(segment) = source_segments + .iter() + .find(|segment| segment.dataset_version > self.manifest.version) + { + return Err(Error::invalid_input(format!( + "merge_existing_index_segments: segment {} was built at future dataset version {} (current version {})", + segment.uuid, segment.dataset_version, self.manifest.version + ))); + } let source_dataset_version = source_segments .iter() .map(|segment| segment.dataset_version) @@ -1434,10 +1542,7 @@ impl DatasetIndexExt for Dataset { let all_fmindex = source_segments.iter().all(segment_has_fmindex_details); let all_zonemap = source_segments.iter().all(segment_has_zonemap_details); let all_label_list = source_segments.iter().all(segment_has_label_list_details); - #[cfg(feature = "geo")] let all_rtree = source_segments.iter().all(segment_has_rtree_details); - #[cfg(not(feature = "geo"))] - let all_rtree = false; if !all_vector && !all_inverted && !all_bitmap @@ -1454,6 +1559,21 @@ impl DatasetIndexExt for Dataset { )); } + let merged_dataset_version = if all_rtree { + let mut source_coverage = source_segments + .iter() + .cloned() + .map(IntoIndexSegment::into_index_segment) + .collect::>>()?; + prune_stale_segment_coverage(self, field_id, &mut source_coverage, true, true).await?; + for (source, coverage) in source_segments.iter_mut().zip(source_coverage) { + source.fragment_bitmap = Some(coverage.fragment_bitmap().clone()); + } + self.manifest.version + } else { + source_dataset_version + }; + let mut merged_segment = if all_vector { crate::index::vector::ivf::merge_segments( self.object_store.as_ref(), @@ -1485,7 +1605,7 @@ impl DatasetIndexExt for Dataset { } else { crate::index::scalar::btree::merge_segments(self, source_segments).await? }; - merged_segment.dataset_version = source_dataset_version; + merged_segment.dataset_version = merged_dataset_version; merged_segment.fields = vec![field_id]; Ok(merged_segment) } diff --git a/rust/lance/src/index/api.rs b/rust/lance/src/index/api.rs index 9cf7507203d..431c9fb2330 100644 --- a/rust/lance/src/index/api.rs +++ b/rust/lance/src/index/api.rs @@ -67,6 +67,10 @@ impl IndexSegment { &self.fragment_bitmap } + pub(crate) fn fragment_bitmap_mut(&mut self) -> &mut RoaringBitmap { + &mut self.fragment_bitmap + } + /// Return the field IDs whose values are encoded in this segment. pub fn fields(&self) -> &[i32] { &self.fields diff --git a/rust/lance/src/index/create.rs b/rust/lance/src/index/create.rs index f6e5a7729f1..0f7a1d51471 100644 --- a/rust/lance/src/index/create.rs +++ b/rust/lance/src/index/create.rs @@ -855,20 +855,30 @@ fn ensure_index_uuid_allowed( fragments: Option<&Vec>, index_uuid: Option<&Uuid>, ) -> Result<()> { - let is_btree = index_type == IndexType::BTree - || params - .as_any() - .downcast_ref::() - .map(|params| params.index_type.eq_ignore_ascii_case("btree")) - .unwrap_or(false); - - if index_uuid.is_some() && fragments.is_some_and(|fragments| !fragments.is_empty()) && is_btree + let scalar_index_type = params + .as_any() + .downcast_ref::() + .map(|params| params.index_type.as_str()); + let unsafe_distributed_type = if index_type == IndexType::BTree + || scalar_index_type.is_some_and(|index_type| index_type.eq_ignore_ascii_case("btree")) + { + Some("BTree") + } else if index_type == IndexType::RTree + || scalar_index_type.is_some_and(|index_type| index_type.eq_ignore_ascii_case("rtree")) + { + Some("RTree") + } else { + None + }; + + if index_uuid.is_some() + && fragments.is_some_and(|fragments| !fragments.is_empty()) + && let Some(index_type) = unsafe_distributed_type { - return Err(Error::invalid_input( - "index_uuid is no longer accepted for BTree distributed index builds; segment UUIDs \ - are generated by Lance and returned in the index metadata." - .to_string(), - )); + return Err(Error::invalid_input(format!( + "index_uuid is no longer accepted for {index_type} distributed index builds; \ + segment UUIDs are generated by Lance and returned in the index metadata." + ))); } Ok(()) @@ -1859,6 +1869,17 @@ mod tests { ); } + let mut future_segment = staged[1].clone(); + future_segment.dataset_version = dataset.manifest.version + 1; + let error = dataset + .merge_existing_index_segments(vec![staged[0].clone(), future_segment]) + .await + .unwrap_err(); + assert!( + error.to_string().contains("future dataset version"), + "unexpected merge error: {error}" + ); + dataset .commit_existing_index_segments("value_bloom_segments", "value", staged.clone()) .await diff --git a/rust/lance/src/index/scalar/rtree.rs b/rust/lance/src/index/scalar/rtree.rs index d19376da3ab..de9fdece137 100644 --- a/rust/lance/src/index/scalar/rtree.rs +++ b/rust/lance/src/index/scalar/rtree.rs @@ -4,14 +4,23 @@ use std::sync::Arc; use lance_index::metrics::NoOpMetricsCollector; -use lance_index::scalar::index_files_to_table; use lance_index::scalar::lance_format::LanceIndexStore; use lance_index::scalar::rtree::RTreeIndex; +use lance_index::scalar::{OldIndexDataFilter, index_files_to_table}; +use lance_select::RowSetOps; use lance_table::format::IndexMetadata; use uuid::Uuid; use crate::{Dataset, Error, Result, dataset::index::LanceIndexStoreExt}; +fn filter_keeps_nothing(filter: &Option) -> bool { + match filter { + Some(OldIndexDataFilter::Fragments { to_keep, .. }) => to_keep.is_empty(), + Some(OldIndexDataFilter::RowIds(valid)) => valid.is_empty(), + None => false, + } +} + /// Merge one caller-defined group of source RTree segments into a single segment. pub(in crate::index) async fn merge_segments( dataset: &Dataset, @@ -38,7 +47,12 @@ pub(in crate::index) async fn merge_segments( crate::index::append::build_per_segment_filters(dataset, &segment_refs).await?; let mut source_indices = Vec::with_capacity(segments.len()); - for segment in &segments { + let mut source_filters = Vec::with_capacity(old_data_filters.len()); + let all_keep_nothing = old_data_filters.iter().all(filter_keeps_nothing); + for (position, (segment, filter)) in segments.iter().zip(&old_data_filters).enumerate() { + if filter_keeps_nothing(filter) && !(all_keep_nothing && position == 0) { + continue; + } let scalar_index = super::open_scalar_index(dataset, &field_path, segment, &NoOpMetricsCollector).await?; let rtree_index = scalar_index @@ -52,6 +66,7 @@ pub(in crate::index) async fn merge_segments( )) })?; source_indices.push(Arc::new(rtree_index.clone())); + source_filters.push(filter.clone()); } let new_uuid = Uuid::new_v4(); @@ -59,7 +74,7 @@ pub(in crate::index) async fn merge_segments( let created_index = lance_index::scalar::rtree::merge_rtree_indices( &source_indices, &new_store, - &old_data_filters, + &source_filters, ) .await?;