Make Managed opt-in: bftree streams bypass ID management layer

Introduce StreamingOutput trait and make run_streaming generic over
the stream type. This allows:

- bftree: implements Stream<DataArgs> directly on StreamRunner, using
  tags as slot IDs without translation (no Managed wrapper)
- inmem: continues using Managed for tag-to-slot ID translation

When inmem 2.0 lands with native ID management, Managed can be
deleted entirely with no infrastructure changes needed.

Also:
- Remove SlotReclaim::Immediate and recycle_tags (bftree-only paths)
- Simplify PhantomData<fn() -> T> to PhantomData<T> on StreamRunner

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

Author: Jordan Maples

diskann-benchmark/src/index/benchmarks.rs

@@ -329,7 +329,7 @@ where
     ) -> anyhow::Result<Vec<managed::Stats<StreamStats>>> {
         writeln!(output, "{}", input)?;
 
-        streaming::run_streaming::<T, _>(
+        streaming::run_streaming::<T, Managed<T, StreamStats>, _>(
             input.runbook_params(),
             |max_points| full_precision_streaming::<T>(input, max_points),
             output,
@@ -742,7 +742,7 @@ where
         .ok_or_else(|| anyhow::anyhow!("consolidate_threshold is required for inmem streaming"))?;
     let capacity = ((max_points as f32) * (1.0 + 2.0 * consolidate_threshold)).ceil() as usize;
 
-    streaming::build_streamer(
+    streaming::build_managed_streamer(
         input.build().data(),
         search,
         streaming::managed::SlotReclaim::Deferred(consolidate_threshold),

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

@@ -20,12 +20,7 @@ use diskann_providers::model::graph::provider::async_::common::FullPrecision;
 use diskann_utils::sampling::WithApproximateNorm;
 
 use crate::{
-    index::streaming::{
-        managed::{self, Managed},
-        runner::BfTreeMaintainer,
-        stats::StreamStats,
-        StreamRunner,
-    },
+    index::streaming::{runner::BfTreeMaintainer, stats::StreamStats, StreamRunner},
     inputs::bftree::{BfTreeStreamingRun, QuantConfig},
     utils,
 };
@@ -52,7 +47,7 @@ where
     T: VectorRepr + WithApproximateNorm + SampleableForStart + AsDataType + bytemuck::Pod,
 {
     type Input = BfTreeStreamingRun;
-    type Output = Vec<managed::Stats<StreamStats>>;
+    type Output = Vec<StreamStats>;
 
     fn try_match(&self, input: &Self::Input) -> Result<MatchScore, FailureScore> {
         let mut failure_score: Option<u32> = None;
@@ -91,18 +86,21 @@ where
     ) -> anyhow::Result<Self::Output> {
         writeln!(output, "{}", input)?;
 
-        crate::index::streaming::run_streaming::<T, _>(
+        crate::index::streaming::run_streaming::<T, BfTreeFullPrecisionStream<T>, _>(
             input.runbook_params(),
             |max_points| bftree_streaming::<T>(input, max_points),
             output,
         )
     }
 }
 
+type BfTreeFullPrecisionStream<T> =
+    StreamRunner<BfTreeProvider<T, NoStore>, T, FullPrecision, BfTreeMaintainer>;
+
 fn bftree_streaming<T>(
     input: &BfTreeStreamingRun,
     max_points: usize,
-) -> anyhow::Result<bigann::WithData<T, u32, Managed<T, StreamStats>>>
+) -> anyhow::Result<bigann::WithData<T, u32, BfTreeFullPrecisionStream<T>>>
 where
     T: bytemuck::Pod + VectorRepr + WithApproximateNorm + SampleableForStart,
 {
@@ -111,10 +109,9 @@ where
     let num_start_points = input.build().start_point_strategy().count();
     let capacity = max_points + num_start_points;
 
-    crate::index::streaming::build_streamer(
+    crate::index::streaming::build_direct_streamer(
         input.build().data(),
         search,
-        crate::index::streaming::managed::SlotReclaim::Immediate,
         capacity,
         |data, capacity| {
             let config = input.try_as_config()?.build()?;

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

@@ -21,11 +21,7 @@ use diskann_bftree::BfTreeProvider;
 use diskann_providers::model::graph::provider::async_::common::Quantized;
 
 use crate::{
-    index::streaming::{
-        managed::{self, Managed},
-        stats::StreamStats,
-        BfTreeMaintainer, StreamRunner,
-    },
+    index::streaming::{stats::StreamStats, BfTreeMaintainer, StreamRunner},
     inputs::bftree::{BfTreeStreamingRun, QuantConfig},
     utils,
 };
@@ -43,7 +39,7 @@ impl StreamingSpherical {
 
 impl Benchmark for StreamingSpherical {
     type Input = BfTreeStreamingRun;
-    type Output = Vec<managed::Stats<StreamStats>>;
+    type Output = Vec<StreamStats>;
 
     fn try_match(&self, input: &Self::Input) -> Result<MatchScore, FailureScore> {
         let mut failure_score: Option<u32> = None;
@@ -87,27 +83,33 @@ impl Benchmark for StreamingSpherical {
     ) -> anyhow::Result<Self::Output> {
         writeln!(output, "{}", input)?;
 
-        crate::index::streaming::run_streaming::<f32, _>(
+        crate::index::streaming::run_streaming::<f32, BfTreeSphericalStream, _>(
             input.runbook_params(),
             |max_points| bftree_sq_streaming_impl(input, max_points),
             output,
         )
     }
 }
 
+type BfTreeSphericalStream = StreamRunner<
+    BfTreeProvider<f32, diskann_bftree::quant::QuantVectorProvider>,
+    f32,
+    Quantized,
+    BfTreeMaintainer,
+>;
+
 fn bftree_sq_streaming_impl(
     input: &BfTreeStreamingRun,
     max_points: usize,
-) -> anyhow::Result<bigann::WithData<f32, u32, Managed<f32, StreamStats>>> {
+) -> 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_streamer(
+    crate::index::streaming::build_direct_streamer(
         input.build().data(),
         search,
-        crate::index::streaming::managed::SlotReclaim::Immediate,
         capacity,
         |data, capacity| {
             let quantizer_poly = super::quantizer_util::build_quantizer(

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

@@ -21,11 +21,6 @@ use crate::utils::streaming::TagSlotManager;
 /// while soft-delete providers need a deferred consolidation pass.
 #[derive(Debug, Clone)]
 pub(crate) enum SlotReclaim {
-    /// Slots are recycled to `empty_slots` immediately during delete.
-    /// No maintenance pass is needed.
-    #[cfg(feature = "bftree")]
-    Immediate,
-
     /// Slots are held in `deleted_slots` until maintenance fires.
     /// Maintenance triggers when `num_deleted > num_active * threshold`.
     Deferred(f32),
@@ -148,8 +143,6 @@ where
         let (overhead_slots, slots) = timed!(self.book_keeping.find_slots_by_tags(tags.clone())?);
         let output = self.stream.delete(&slots)?;
         let (overhead_reclaim, _) = timed!(match &self.reclaim {
-            #[cfg(feature = "bftree")]
-            SlotReclaim::Immediate => self.book_keeping.recycle_tags(tags)?,
             SlotReclaim::Deferred(_) => self.book_keeping.mark_tags_deleted(tags)?,
         });
         Ok(Stats::new(overhead_slots + overhead_reclaim, output))
@@ -163,8 +156,6 @@ where
 
     fn needs_maintenance(&mut self) -> bool {
         match &self.reclaim {
-            #[cfg(feature = "bftree")]
-            SlotReclaim::Immediate => false,
             SlotReclaim::Deferred(threshold) => {
                 let num_active = self.book_keeping.num_active();
                 let limit = (num_active as f32 * threshold) as usize;

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

@@ -6,7 +6,7 @@
 use std::{io::Write, sync::Arc};
 
 use diskann::utils::VectorRepr;
-use diskann_benchmark_core::streaming::{executors::bigann, Executor};
+use diskann_benchmark_core::streaming::{self, executors::bigann, Executor};
 use diskann_benchmark_runner::output::Output;
 use diskann_utils::views::Matrix;
 
@@ -24,14 +24,34 @@ use crate::{
     utils::datafiles,
 };
 
+/// Trait for streaming benchmark outputs that wrap or produce [`stats::StreamStats`].
+///
+/// This allows [`run_streaming`] to work with both direct streams (which produce
+/// `StreamStats` directly) and managed streams (which produce `managed::Stats<StreamStats>`).
+pub(crate) trait StreamingOutput: std::fmt::Display + 'static {
+    fn stream_stats(&self) -> &stats::StreamStats;
+}
+
+impl StreamingOutput for stats::StreamStats {
+    fn stream_stats(&self) -> &stats::StreamStats {
+        self
+    }
+}
+
+impl StreamingOutput for managed::Stats<stats::StreamStats> {
+    fn stream_stats(&self) -> &stats::StreamStats {
+        self.inner()
+    }
+}
+
 /// Construct the streaming stack: load data/queries, create the managed stream via the
 /// closure, then wrap in [`Managed`] and [`bigann::WithData`].
 ///
 /// `capacity` is the pre-computed slot count passed to [`Managed::new`]. Each backend
 /// computes this differently (inmem applies headroom, bf_tree adds start points).
 ///
 /// The closure receives `(&data, capacity)` so it can use `data.ncols()` for provider params.
-pub(crate) fn build_streamer<T, M, F>(
+pub(crate) fn build_managed_streamer<T, M, F>(
     data_path: &diskann_benchmark_runner::files::InputFile,
     search: &StreamingSearchParams,
     reclaim: managed::SlotReclaim,
@@ -62,19 +82,57 @@ where
     Ok(layered)
 }
 
+/// Construct a direct streaming stack (no ID management layer).
+///
+/// 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.
+#[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>,
+{
+    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 max_k = search.max_k();
+    let layered = bigann::WithData::new(stream, data, queries, move |path| {
+        Ok(Box::new(datafiles::load_groundtruth(
+            datafiles::BinFile(path),
+            Some(max_k),
+        )?))
+    });
+
+    Ok(layered)
+}
+
 /// Run a streaming benchmark using the given runbook parameters.
 ///
 /// `make_streamer` receives `max_points` from the loaded runbook and returns the
 /// constructed streamer. This is shared across all streaming benchmarks (inmem, bftree)
 /// to avoid duplicating the runbook load → run_with → stage banner → summary logic.
-pub(crate) fn run_streaming<T, F>(
+pub(crate) fn run_streaming<T, S, F>(
     runbook_params: &StreamingRunbookParams,
     make_streamer: F,
     mut output: &mut dyn Output,
-) -> anyhow::Result<Vec<managed::Stats<stats::StreamStats>>>
+) -> anyhow::Result<Vec<S::Output>>
 where
     T: 'static,
-    F: FnOnce(usize) -> anyhow::Result<bigann::WithData<T, u32, Managed<T, stats::StreamStats>>>,
+    S: streaming::Stream<bigann::DataArgs<T, u32>>,
+    S::Output: StreamingOutput,
+    F: FnOnce(usize) -> anyhow::Result<bigann::WithData<T, u32, S>>,
 {
     let groundtruth_directory = runbook_params
         .resolved_gt_directory
@@ -95,22 +153,24 @@ where
     let stages = runbook.len();
     let mut i = 1;
 
-    runbook.run_with(
-        &mut streamer,
-        |o: managed::Stats<stats::StreamStats>| -> anyhow::Result<()> {
-            if o.inner().is_maintain() {
-                let message = format!("Ran maintenance before stage {}", i);
-                write!(output, "{}", crate::utils::SmallBanner(&message))?;
-            } else {
-                let message = format!("Finished stage {} of {}: {}", i, stages, o.inner().kind());
-                write!(output, "{}", crate::utils::SmallBanner(&message))?;
-                i += 1;
-            }
-            writeln!(output, "{}", o)?;
-            results.push(o);
-            Ok(())
-        },
-    )?;
+    runbook.run_with(&mut streamer, |o: S::Output| -> anyhow::Result<()> {
+        if o.stream_stats().is_maintain() {
+            let message = format!("Ran maintenance before stage {}", i);
+            write!(output, "{}", crate::utils::SmallBanner(&message))?;
+        } else {
+            let message = format!(
+                "Finished stage {} of {}: {}",
+                i,
+                stages,
+                o.stream_stats().kind()
+            );
+            write!(output, "{}", crate::utils::SmallBanner(&message))?;
+            i += 1;
+        }
+        writeln!(output, "{}", o)?;
+        results.push(o);
+        Ok(())
+    })?;
 
     write!(
         output,
@@ -121,7 +181,7 @@ where
     writeln!(
         output,
         "{}",
-        stats::Summary::new(results.iter().map(|r| r.inner()))
+        stats::Summary::new(results.iter().map(|r| r.stream_stats()))
     )?;
 
     Ok(results)