Fix bftree direct streaming: size provider to dataset ID space

The direct (non-Managed) streaming path uses absolute runbook tag IDs
directly as provider slot IDs. The provider was sized to the runbook's
max concurrent point count (max_pts), but absolute tag IDs span the full
dataset, so an insert with id >= capacity failed with "Vector id is out
of boundary in the dataset".

Size the provider to data.nrows() + num_start_points instead. Drop the
redundant capacity parameter from build_direct_streamer and derive it
from the loaded dataset in each bftree streaming closure.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Jordan Maples

diskann-benchmark/src/index/bftree/full_precision_streaming.rs

@@ -74,7 +74,7 @@ where
 
         crate::index::streaming::run_streaming::<T, BfTreeFullPrecisionStream<T>, _>(
             input.runbook_params(),
-            |max_points| bftree_streaming::<T>(input, max_points),
+            |_max_points| bftree_streaming::<T>(input),
             output,
         )
     }
@@ -85,41 +85,38 @@ type BfTreeFullPrecisionStream<T> =
 
 fn bftree_streaming<T>(
     input: &BfTreeStreamingRun,
-    max_points: usize,
 ) -> anyhow::Result<bigann::WithData<T, u32, BfTreeFullPrecisionStream<T>>>
 where
     T: bytemuck::Pod + VectorRepr + WithApproximateNorm + SampleableForStart,
 {
     let search = input.search();
 
     let num_start_points = input.build().start_point_strategy().count();
-    let capacity = max_points + num_start_points;
-
-    crate::index::streaming::build_direct_streamer(
-        input.build().data(),
-        search,
-        capacity,
-        |data, capacity| {
-            let config = input.try_as_config()?.build()?;
-            let params = input.bftree_parameters(capacity, data.ncols())?;
-            let start_points = input
-                .build()
-                .start_point_strategy()
-                .compute(data.as_view())?;
-            let provider = BfTreeProvider::new(params, start_points.as_view(), NoStore)?;
-            let index = Arc::new(DiskANNIndex::new(config, provider, None));
-
-            let num_threads_and_tasks = NonZeroUsize::new(input.build().num_threads()).unwrap();
-            Ok(StreamRunner::new(
-                index,
-                FullPrecision,
-                search.clone(),
-                benchmark_core::tokio::runtime(num_threads_and_tasks.get())?,
-                num_threads_and_tasks,
-                input.runbook_params().ip_delete_num_to_replace,
-                input.runbook_params().ip_delete_method.into(),
-                BfTreeMaintainer,
-            ))
-        },
-    )
+
+    crate::index::streaming::build_direct_streamer(input.build().data(), search, |data| {
+        // The direct (non-Managed) path uses absolute runbook tag IDs as slot IDs,
+        // so the provider must span the full dataset ID space rather than the
+        // runbook's max concurrent point count.
+        let capacity = data.nrows() + num_start_points;
+        let config = input.try_as_config()?.build()?;
+        let params = input.bftree_parameters(capacity, data.ncols())?;
+        let start_points = input
+            .build()
+            .start_point_strategy()
+            .compute(data.as_view())?;
+        let provider = BfTreeProvider::new(params, start_points.as_view(), NoStore)?;
+        let index = Arc::new(DiskANNIndex::new(config, provider, None));
+
+        let num_threads_and_tasks = NonZeroUsize::new(input.build().num_threads()).unwrap();
+        Ok(StreamRunner::new(
+            index,
+            FullPrecision,
+            search.clone(),
+            benchmark_core::tokio::runtime(num_threads_and_tasks.get())?,
+            num_threads_and_tasks,
+            input.runbook_params().ip_delete_num_to_replace,
+            input.runbook_params().ip_delete_method.into(),
+            BfTreeMaintainer,
+        ))
+    })
 }

diskann-benchmark/src/index/bftree/spherical_streaming.rs

@@ -68,7 +68,7 @@ impl Benchmark for StreamingSpherical {
 
         crate::index::streaming::run_streaming::<f32, BfTreeSphericalStream, _>(
             input.runbook_params(),
-            |max_points| bftree_sq_streaming_impl(input, max_points),
+            |_max_points| bftree_sq_streaming_impl(input),
             output,
         )
     }
@@ -83,45 +83,42 @@ type BfTreeSphericalStream = StreamRunner<
 
 fn bftree_sq_streaming_impl(
     input: &BfTreeStreamingRun,
-    max_points: usize,
 ) -> anyhow::Result<bigann::WithData<f32, u32, BfTreeSphericalStream>> {
     let search = input.search();
 
     let num_start_points = input.build().start_point_strategy().count();
-    let capacity = max_points + num_start_points;
-
-    crate::index::streaming::build_direct_streamer(
-        input.build().data(),
-        search,
-        capacity,
-        |data, capacity| {
-            let quantizer_poly = super::quantizer_util::build_quantizer(
-                input.quantization(),
-                data.as_view(),
-                input.build().distance(),
-            )?
-            .expect("spherical quantization config guaranteed by try_match");
-
-            let config = input.try_as_config()?.build()?;
-            let params = input.bftree_parameters(capacity, data.ncols())?;
-            let start_points = input
-                .build()
-                .start_point_strategy()
-                .compute(data.as_view())?;
-            let provider = BfTreeProvider::new(params, start_points.as_view(), quantizer_poly)?;
-            let index = Arc::new(DiskANNIndex::new(config, provider, None));
-
-            let num_threads_and_tasks = NonZeroUsize::new(input.build().num_threads()).unwrap();
-            Ok(StreamRunner::new(
-                index,
-                Quantized,
-                search.clone(),
-                benchmark_core::tokio::runtime(num_threads_and_tasks.get())?,
-                num_threads_and_tasks,
-                input.runbook_params().ip_delete_num_to_replace,
-                input.runbook_params().ip_delete_method.into(),
-                BfTreeMaintainer,
-            ))
-        },
-    )
+
+    crate::index::streaming::build_direct_streamer(input.build().data(), search, |data| {
+        // The direct (non-Managed) path uses absolute runbook tag IDs as slot IDs,
+        // so the provider must span the full dataset ID space rather than the
+        // runbook's max concurrent point count.
+        let capacity = data.nrows() + num_start_points;
+        let quantizer_poly = super::quantizer_util::build_quantizer(
+            input.quantization(),
+            data.as_view(),
+            input.build().distance(),
+        )?
+        .expect("spherical quantization config guaranteed by try_match");
+
+        let config = input.try_as_config()?.build()?;
+        let params = input.bftree_parameters(capacity, data.ncols())?;
+        let start_points = input
+            .build()
+            .start_point_strategy()
+            .compute(data.as_view())?;
+        let provider = BfTreeProvider::new(params, start_points.as_view(), quantizer_poly)?;
+        let index = Arc::new(DiskANNIndex::new(config, provider, None));
+
+        let num_threads_and_tasks = NonZeroUsize::new(input.build().num_threads()).unwrap();
+        Ok(StreamRunner::new(
+            index,
+            Quantized,
+            search.clone(),
+            benchmark_core::tokio::runtime(num_threads_and_tasks.get())?,
+            num_threads_and_tasks,
+            input.runbook_params().ip_delete_num_to_replace,
+            input.runbook_params().ip_delete_method.into(),
+            BfTreeMaintainer,
+        ))
+    })
 }

diskann-benchmark/src/index/streaming/mod.rs

@@ -87,25 +87,27 @@ where
 /// For providers where external IDs match internal slots (e.g., bf-tree), the
 /// [`Managed`] layer is unnecessary. This function creates the stack without it.
 ///
-/// The closure receives `(&data, capacity)` so it can use `data.ncols()` for provider params.
+/// Because external runbook tag IDs are used directly as provider slot IDs (no
+/// remapping into a compact slot space), the provider must be sized to span the
+/// full dataset ID range rather than the runbook's max concurrent point count.
+/// The closure receives `&data` so it can size the provider from `data.nrows()`.
 #[cfg(feature = "bftree")]
 pub(crate) fn build_direct_streamer<T, S, F>(
     data_path: &diskann_benchmark_runner::files::InputFile,
     search: &StreamingSearchParams,
-    capacity: usize,
     make_stream: F,
 ) -> anyhow::Result<bigann::WithData<T, u32, S>>
 where
     T: bytemuck::Pod + VectorRepr + 'static,
     S: streaming::Stream<bigann::DataArgs<T, u32>> + 'static,
-    F: FnOnce(&Matrix<T>, usize) -> anyhow::Result<S>,
+    F: FnOnce(&Matrix<T>) -> anyhow::Result<S>,
 {
     let data = datafiles::load_dataset::<T>(datafiles::BinFile(data_path))?;
     let queries = Arc::new(datafiles::load_dataset::<T>(datafiles::BinFile(
         &search.queries,
     ))?);
 
-    let stream = make_stream(&data, capacity)?;
+    let stream = make_stream(&data)?;
 
     let max_k = search.max_k();
     let layered = bigann::WithData::new(stream, data, queries, move |path| {