feat: add progress monitoring via callbacks for inverted indexes

rust/lance-index/benches/geo.rs

@@ -78,6 +78,7 @@ async fn build_rtree(
             store.as_ref(),
             Box::new(RTreeTrainingRequest::default()),
             None,
+            lance_index::progress::noop_progress(),
         )
         .await?;
 

rust/lance-index/src/scalar/bitmap.rs

@@ -803,6 +803,7 @@ impl ScalarIndexPlugin for BitmapIndexPlugin {
         index_store: &dyn IndexStore,
         _request: Box<dyn TrainingRequest>,
         fragment_ids: Option<Vec<u32>>,
+        _progress: Arc<dyn crate::progress::IndexBuildProgress>,
     ) -> Result<CreatedIndex> {
         if fragment_ids.is_some() {
             return Err(Error::InvalidInput {

rust/lance-index/src/scalar/bloomfilter.rs

@@ -1110,6 +1110,7 @@ impl ScalarIndexPlugin for BloomFilterIndexPlugin {
         index_store: &dyn IndexStore,
         request: Box<dyn TrainingRequest>,
         fragment_ids: Option<Vec<u32>>,
+        _progress: Arc<dyn crate::progress::IndexBuildProgress>,
     ) -> Result<CreatedIndex> {
         if fragment_ids.is_some() {
             return Err(Error::InvalidInput {

rust/lance-index/src/scalar/btree.rs

@@ -2543,6 +2543,7 @@ impl ScalarIndexPlugin for BTreeIndexPlugin {
         index_store: &dyn IndexStore,
         request: Box<dyn TrainingRequest>,
         fragment_ids: Option<Vec<u32>>,
+        _progress: Arc<dyn crate::progress::IndexBuildProgress>,
     ) -> Result<CreatedIndex> {
         let request = request
             .as_any()

rust/lance-index/src/scalar/inverted.rs

@@ -28,6 +28,7 @@ use lance_core::Error;
 use snafu::location;
 
 use crate::pbold;
+use crate::progress::IndexBuildProgress;
 use crate::{
     frag_reuse::FragReuseIndex,
     scalar::{
@@ -48,6 +49,7 @@ impl InvertedIndexPlugin {
         index_store: &dyn IndexStore,
         params: InvertedIndexParams,
         fragment_ids: Option<Vec<u32>>,
+        progress: Arc<dyn IndexBuildProgress>,
     ) -> Result<CreatedIndex> {
         let fragment_mask = fragment_ids.as_ref().and_then(|frag_ids| {
             if !frag_ids.is_empty() {
@@ -62,7 +64,8 @@ impl InvertedIndexPlugin {
 
         let details = pbold::InvertedIndexDetails::try_from(&params)?;
         let mut inverted_index =
-            InvertedIndexBuilder::new_with_fragment_mask(params, fragment_mask);
+            InvertedIndexBuilder::new_with_fragment_mask(params, fragment_mask)
+                .with_progress(progress);
         inverted_index.update(data, index_store).await?;
         Ok(CreatedIndex {
             index_details: prost_types::Any::from_msg(&details).unwrap(),
@@ -173,15 +176,22 @@ impl ScalarIndexPlugin for InvertedIndexPlugin {
         index_store: &dyn IndexStore,
         request: Box<dyn TrainingRequest>,
         fragment_ids: Option<Vec<u32>>,
+        progress: Arc<dyn IndexBuildProgress>,
     ) -> Result<CreatedIndex> {
         let request = (request as Box<dyn std::any::Any>)
             .downcast::<InvertedIndexTrainingRequest>()
             .map_err(|_| Error::InvalidInput {
                 source: "must provide training request created by new_training_request".into(),
                 location: location!(),
             })?;
-        Self::train_inverted_index(data, index_store, request.parameters.clone(), fragment_ids)
-            .await
+        Self::train_inverted_index(
+            data,
+            index_store,
+            request.parameters.clone(),
+            fragment_ids,
+            progress,
+        )
+        .await
     }
 
     /// Load an index from storage