From 8fc74437d7e0a0b98ab1c085cdf67a89faa268d5 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 12:52:33 +0000
Subject: [PATCH 01/21] benchmark: configure PiPNN graph builds

---
 diskann-benchmark/Cargo.toml                |  4 ++
 diskann-benchmark/src/disk_index/build.rs   | 23 ++++--
 diskann-benchmark/src/inputs/disk.rs        | 78 +++++++++++++++------
 diskann-benchmark/src/inputs/graph_index.rs | 23 +++++-
 4 files changed, 99 insertions(+), 29 deletions(-)

diff --git a/diskann-benchmark/Cargo.toml b/diskann-benchmark/Cargo.toml
index 31739244c9..2005d639ed 100644
--- a/diskann-benchmark/Cargo.toml
+++ b/diskann-benchmark/Cargo.toml
@@ -32,6 +32,7 @@ diskann-label-filter.workspace = true
 diskann-tools = { workspace = true }
 diskann-disk = { workspace = true, optional = true }
 diskann-bftree = { workspace = true, optional = true }
+diskann-pipnn = { workspace = true, optional = true }
 cfg-if.workspace = true
 diskann-benchmark-runner = { workspace = true }
 opentelemetry = { workspace = true, optional = true }
@@ -82,3 +83,6 @@ disk-index = [
     "dep:opentelemetry_sdk",
     "dep:scopeguard",
 ]
+
+# Enable PiPNN graph construction.
+pipnn = ["dep:diskann-pipnn", "diskann-disk/pipnn"]
diff --git a/diskann-benchmark/src/disk_index/build.rs b/diskann-benchmark/src/disk_index/build.rs
index 9ab095cdfa..0ab1f689a0 100644
--- a/diskann-benchmark/src/disk_index/build.rs
+++ b/diskann-benchmark/src/disk_index/build.rs
@@ -84,6 +84,7 @@ where
         params.l_build,
         metric.into(),
         |b| {
+            b.alpha(params.alpha);
             b.saturate_after_prune(true);
         },
     )
@@ -93,11 +94,23 @@ where
 
     let metadata = load_metadata_from_file(storage_provider, &data_path)?;
 
-    let build_parameters = DiskIndexBuildParameters::new(
-        MemoryBudget::try_from_gb(params.build_ram_limit_gb)?,
-        params.quantization_type,
-        NumPQChunks::new_with(params.num_pq_chunks.get(), metadata.ndims())?,
-    );
+    let search_pq_chunks = NumPQChunks::new_with(params.num_pq_chunks.get(), metadata.ndims())?;
+    let build_parameters = match &params.build_algorithm {
+        diskann_disk::BuildAlgorithm::Vamana => DiskIndexBuildParameters::new(
+            MemoryBudget::try_from_gb(params.build_ram_limit_gb)?,
+            params.quantization_type.ok_or_else(|| {
+                anyhow::anyhow!("quantization_type is required for Vamana builds")
+            })?,
+            search_pq_chunks,
+        ),
+        #[cfg(feature = "pipnn")]
+        diskann_disk::BuildAlgorithm::PiPNN(config) => DiskIndexBuildParameters::new_pipnn(
+            MemoryBudget::try_from_gb(params.build_ram_limit_gb)?,
+            search_pq_chunks,
+            config.clone(),
+        ),
+        _ => anyhow::bail!("unsupported disk graph-build algorithm"),
+    };
 
     let index_configuration = IndexConfiguration::new(
         metric,
diff --git a/diskann-benchmark/src/inputs/disk.rs b/diskann-benchmark/src/inputs/disk.rs
index 032665cd7a..d1a815969d 100644
--- a/diskann-benchmark/src/inputs/disk.rs
+++ b/diskann-benchmark/src/inputs/disk.rs
@@ -15,6 +15,8 @@ use diskann_benchmark_runner::{files::InputFile, utils::datatype::DataType, Chec
 #[cfg(feature = "disk-index")]
 use diskann_disk::search::search_mode::SearchMode;
 #[cfg(feature = "disk-index")]
+use diskann_disk::BuildAlgorithm;
+#[cfg(feature = "disk-index")]
 use diskann_disk::QuantizationType;
 use diskann_providers::storage::{get_compressed_pq_file, get_disk_index_file, get_pq_pivot_file};
 use serde::{Deserialize, Serialize};
@@ -64,14 +66,27 @@ pub(crate) struct DiskIndexBuild {
     pub(crate) dim: usize,
     pub(crate) max_degree: usize,
     pub(crate) l_build: usize,
+    #[serde(default = "default_alpha")]
+    pub(crate) alpha: f32,
     pub(crate) num_threads: usize,
+    /// Graph-build RAM limit. PiPNN falls back to Vamana when its estimated
+    /// one-shot peak exceeds this value.
     pub(crate) build_ram_limit_gb: f64,
     pub(crate) num_pq_chunks: NonZeroUsize,
     #[cfg(feature = "disk-index")]
-    pub(crate) quantization_type: QuantizationType,
+    #[serde(default)]
+    pub(crate) quantization_type: Option<QuantizationType>,
+    /// Build algorithm: "Vamana" (default) or "PiPNN" with config params.
+    #[cfg(feature = "disk-index")]
+    #[serde(default)]
+    pub(crate) build_algorithm: BuildAlgorithm,
     pub(crate) save_path: String,
 }
 
+fn default_alpha() -> f32 {
+    diskann::graph::config::defaults::ALPHA
+}
+
 #[cfg(feature = "disk-index")]
 #[derive(Debug, Serialize, Deserialize, Default)]
 pub(crate) struct DiskSearchMode {
@@ -230,10 +245,23 @@ impl DiskIndexBuild {
         if self.num_threads == 0 {
             anyhow::bail!("num_threads must be positive");
         }
-        if self.build_ram_limit_gb <= 0.0 {
+        if !self.alpha.is_finite() || self.alpha < 1.0 {
+            anyhow::bail!("alpha must be finite and at least 1.0");
+        }
+        if !self.build_ram_limit_gb.is_finite() || self.build_ram_limit_gb <= 0.0 {
             anyhow::bail!("build_ram_limit_gb must be strictly positive");
         }
-
+        #[cfg(feature = "disk-index")]
+        match &self.build_algorithm {
+            BuildAlgorithm::Vamana if self.quantization_type.is_none() => {
+                anyhow::bail!("quantization_type is required for Vamana builds");
+            }
+            #[cfg(feature = "pipnn")]
+            BuildAlgorithm::PiPNN(_) if self.quantization_type.is_some() => {
+                anyhow::bail!("PiPNN graph construction is full precision; omit quantization_type");
+            }
+            _ => {}
+        }
         // Relative save path with respect to output directory is not supported.
         if checker.output_directory().is_some() {
             anyhow::bail!("relative save_path with respect to output_directory is not supported");
@@ -338,11 +366,14 @@ impl Example for DiskIndexOperation {
             dim: 128,
             max_degree: 32,
             l_build: 50,
+            alpha: default_alpha(),
             num_threads: 8,
             build_ram_limit_gb: 16.0,
             num_pq_chunks: NonZeroUsize::new(16).unwrap(),
             #[cfg(feature = "disk-index")]
-            quantization_type: QuantizationType::PQ { num_chunks: 16 },
+            quantization_type: Some(QuantizationType::PQ { num_chunks: 16 }),
+            #[cfg(feature = "disk-index")]
+            build_algorithm: BuildAlgorithm::default(),
             save_path: "sample_index_l50_r32".to_string(),
         };
 
@@ -422,27 +453,30 @@ impl DiskIndexBuild {
         write_field!(f, "Dim", self.dim)?;
         write_field!(f, "Max Degree", self.max_degree)?;
         write_field!(f, "L Build", self.l_build)?;
+        write_field!(f, "Alpha", self.alpha)?;
         write_field!(f, "Build Threads", self.num_threads)?;
         write_field!(f, "Build RAM Limit GB", self.build_ram_limit_gb)?;
         write_field!(f, "PQ Chunks", self.num_pq_chunks)?;
         #[cfg(feature = "disk-index")]
-        match &self.quantization_type {
-            QuantizationType::FP => write_field!(f, "Quantization", "full precision")?,
-            QuantizationType::PQ { num_chunks } => {
-                write_field!(f, "Quantization", format!("pq, chunks {num_chunks}"))?
-            }
-            QuantizationType::SQ {
-                nbits,
-                standard_deviation,
-            } => {
-                if let Some(sd) = standard_deviation {
-                    write_field!(
-                        f,
-                        "Quantization",
-                        format!("sq, nbits {nbits}, stdev {}", sd.into_inner())
-                    )?
-                } else {
-                    write_field!(f, "Quantization", format!("sq, nbits {nbits}"))?
+        if let Some(quantization_type) = &self.quantization_type {
+            match quantization_type {
+                QuantizationType::FP => write_field!(f, "Quantization", "full precision")?,
+                QuantizationType::PQ { num_chunks } => {
+                    write_field!(f, "Quantization", format!("pq, chunks {num_chunks}"))?
+                }
+                QuantizationType::SQ {
+                    nbits,
+                    standard_deviation,
+                } => {
+                    if let Some(sd) = standard_deviation {
+                        write_field!(
+                            f,
+                            "Quantization",
+                            format!("sq, nbits {nbits}, stdev {}", sd.into_inner())
+                        )?
+                    } else {
+                        write_field!(f, "Quantization", format!("sq, nbits {nbits}"))?
+                    }
                 }
             }
             QuantizationType::Spherical(bits) => write_field!(
@@ -451,6 +485,8 @@ impl DiskIndexBuild {
                 format!("spherical, nbits {}", bits.as_usize())
             )?,
         }
+        #[cfg(feature = "disk-index")]
+        write_field!(f, "Build Algorithm", self.build_algorithm)?;
         write_field!(f, "Save Path", self.save_path)?;
         Ok(())
     }
diff --git a/diskann-benchmark/src/inputs/graph_index.rs b/diskann-benchmark/src/inputs/graph_index.rs
index c7ed6c1410..1397eecbfe 100644
--- a/diskann-benchmark/src/inputs/graph_index.rs
+++ b/diskann-benchmark/src/inputs/graph_index.rs
@@ -3,7 +3,7 @@
  * Licensed under the MIT license.
  */
 
-use std::num::{NonZero, NonZeroU32, NonZeroUsize};
+use std::num::{NonZeroU32, NonZeroUsize};
 
 use anyhow::{anyhow, Context};
 use diskann::{
@@ -722,6 +722,9 @@ pub(crate) struct IndexBuild {
     num_threads: usize,
     multi_insert: Option<MultiInsert>,
     save_path: Option<String>,
+    #[cfg(feature = "pipnn")]
+    #[serde(default)]
+    build_algorithm: diskann_disk::BuildAlgorithm,
 }
 
 impl IndexBuild {
@@ -787,9 +790,12 @@ impl IndexBuild {
         num_points: usize,
         dim: usize,
     ) -> DefaultProviderParameters {
+        let frozen_points =
+            NonZeroUsize::new(self.start_point_strategy.count()).unwrap_or(NonZeroUsize::MIN);
+
         DefaultProviderParameters {
             max_points: num_points,
-            frozen_points: NonZero::new(self.start_point_strategy.count()).unwrap(),
+            frozen_points,
             metric: self.distance.into(),
             dim,
             max_degree: self.exact_max_degree() as u32,
@@ -855,7 +861,11 @@ impl IndexBuild {
         self.num_threads
     }
 
-    #[cfg(any(feature = "spherical-quantization", feature = "bftree"))]
+    #[cfg(any(
+        feature = "spherical-quantization",
+        feature = "bftree",
+        feature = "pipnn"
+    ))]
     pub(crate) fn distance(&self) -> SimilarityMeasure {
         self.distance
     }
@@ -875,6 +885,11 @@ impl IndexBuild {
     pub(crate) fn save_path(&self) -> Option<&str> {
         self.save_path.as_deref()
     }
+
+    #[cfg(feature = "pipnn")]
+    pub(crate) fn build_algorithm(&self) -> &diskann_disk::BuildAlgorithm {
+        &self.build_algorithm
+    }
 }
 
 impl Example for IndexBuild {
@@ -892,6 +907,8 @@ impl Example for IndexBuild {
             insert_retry: None,
             start_point_strategy: StartPointStrategy::Medoid,
             save_path: None,
+            #[cfg(feature = "pipnn")]
+            build_algorithm: diskann_disk::BuildAlgorithm::Vamana,
         }
     }
 }

From ffd730b13b3203892b67eba0511d9ea2afb3b969 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 12:52:33 +0000
Subject: [PATCH 02/21] benchmark: add dedicated PiPNN batch pipeline

---
 diskann-benchmark/src/index/benchmarks.rs |  54 ++++---
 diskann-benchmark/src/index/build.rs      | 171 ++++++++++++++++++++++
 2 files changed, 205 insertions(+), 20 deletions(-)

diff --git a/diskann-benchmark/src/index/benchmarks.rs b/diskann-benchmark/src/index/benchmarks.rs
index 34ae5ff5f5..31f965f04a 100644
--- a/diskann-benchmark/src/index/benchmarks.rs
+++ b/diskann-benchmark/src/index/benchmarks.rs
@@ -211,26 +211,40 @@ where
         writeln!(output, "{}", input)?;
         let (index, build_stats) = match &input.source {
             IndexSource::Build(build) => {
-                let (index, build_stats) = run_build(
-                    build,
-                    common::FullPrecision,
-                    None,
-                    output,
-                    |data| {
-                        let index = diskann_async::new_index::<T, _>(
-                            build.try_as_config()?.build()?,
-                            build.inmem_parameters(data.nrows(), data.ncols()),
-                            common::NoDeletes,
-                        )?;
-                        build::set_start_points(
-                            index.provider(),
-                            data.as_view(),
-                            *build.start_point_strategy(),
-                        )?;
-                        Ok(index)
-                    },
-                    single_or_multi_insert,
-                )?;
+                let mut incremental = || {
+                    run_build(
+                        build,
+                        common::FullPrecision,
+                        None,
+                        output,
+                        |data| {
+                            let index = diskann_async::new_index::<T, _>(
+                                build.try_as_config()?.build()?,
+                                build.inmem_parameters(data.nrows(), data.ncols()),
+                                common::NoDeletes,
+                            )?;
+                            build::set_start_points(
+                                index.provider(),
+                                data.as_view(),
+                                *build.start_point_strategy(),
+                            )?;
+                            Ok(index)
+                        },
+                        single_or_multi_insert,
+                    )
+                };
+                #[cfg(feature = "pipnn")]
+                let result = match build.build_algorithm() {
+                    diskann_disk::BuildAlgorithm::PiPNN(parameters) => {
+                        let data =
+                            Arc::new(datafiles::load_dataset(datafiles::BinFile(build.data()))?);
+                        build::pipnn_build(data, build, parameters)
+                    }
+                    _ => incremental(),
+                };
+                #[cfg(not(feature = "pipnn"))]
+                let result = incremental();
+                let (index, build_stats) = result?;
 
                 // save the index if requested
                 if let Some(save_path) = build.save_path() {
diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index b4bf186e68..b696e64a93 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -20,10 +20,17 @@ use diskann_providers::{
     model::{configuration::IndexConfiguration, graph::provider::async_::inmem::SetStartPoints},
     storage::{AsyncIndexMetadata, LoadWith, SaveWith},
 };
+#[cfg(feature = "pipnn")]
+use diskann_providers::{
+    index::diskann_async,
+    model::graph::provider::async_::common::{self, SetElementHelper},
+};
 use diskann_utils::{
     future::AsyncFriendly,
     views::{Matrix, MatrixView},
 };
+#[cfg(feature = "pipnn")]
+use diskann_vector::DistanceFunction;
 use indicatif::{ProgressBar, ProgressStyle};
 use serde::Serialize;
 
@@ -105,6 +112,114 @@ where
     }
 }
 
+#[cfg(feature = "pipnn")]
+pub(crate) fn pipnn_build<T>(
+    data: Arc<Matrix<T>>,
+    input: &IndexBuild,
+    parameters: &diskann_disk::PiPNNParameters,
+) -> anyhow::Result<(diskann_async::MemoryIndex<T>, BuildStats)>
+where
+    T: diskann::graph::SampleableForStart + diskann::utils::VectorRepr,
+{
+    use anyhow::Context;
+
+    let npoints = data.nrows();
+    let dimensions = data.ncols();
+    let metric = input.distance().into();
+    let graph = input.try_as_config()?.build()?;
+    let pool = rayon::ThreadPoolBuilder::new()
+        .num_threads(input.num_threads())
+        .build()
+        .context("failed to create PiPNN build thread pool")?;
+
+    let started = std::time::Instant::now();
+    let adjacency = {
+        let context =
+            diskann_pipnn::PiPNNBuildContext::new(parameters.into(), &graph, metric, &pool)?;
+        diskann_pipnn::build_graph(data.as_view(), &context)?
+    };
+    let start_points = input
+        .start_point_strategy()
+        .compute(data.as_view())
+        .map_err(|error| {
+            ANNError::new(
+                diskann::ANNErrorKind::DiskANN(StartPointComputeError),
+                error,
+            )
+        })?;
+    let distance = T::distance(metric, Some(dimensions));
+    let start_sources = start_points
+        .row_iter()
+        .map(|start| {
+            let bytes: &[u8] = bytemuck::cast_slice(start);
+            data.row_iter()
+                .position(|row| bytemuck::cast_slice::<T, u8>(row) == bytes)
+                .or_else(|| {
+                    data.row_iter()
+                        .enumerate()
+                        .min_by(|(_, left), (_, right)| {
+                            distance
+                                .evaluate_similarity(start, left)
+                                .total_cmp(&distance.evaluate_similarity(start, right))
+                        })
+                        .map(|(index, _)| index)
+                })
+                .context("PiPNN cannot connect a start point to an empty dataset")
+        })
+        .collect::<anyhow::Result<Vec<_>>>()?;
+    let start_neighbors: Vec<_> = start_sources
+        .into_iter()
+        .map(|source| adjacency[source].clone())
+        .collect();
+
+    // Unlike incremental insertion, PiPNN finishes all batch-build scratch
+    // before allocating the provider that owns the searchable index.
+    let index = diskann_async::new_index::<T, _>(
+        graph,
+        input.inmem_parameters(npoints, dimensions),
+        common::NoDeletes,
+    )?;
+    for (id, vector) in data.row_iter().enumerate() {
+        let id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
+        index.data_provider.base_vectors.set_element(&id, vector)?;
+    }
+    for (id, neighbors) in adjacency.into_iter().enumerate() {
+        index
+            .provider()
+            .neighbors()
+            .set_neighbors_sync(id, &neighbors)?;
+    }
+    index.provider().set_start_points(start_points.row_iter())?;
+    let start_ids = index.provider().starting_points()?;
+    anyhow::ensure!(
+        start_ids.len() == start_neighbors.len(),
+        "PiPNN provider created {} start slots for {} start vectors",
+        start_ids.len(),
+        start_neighbors.len()
+    );
+    for (start_id, neighbors) in start_ids.into_iter().zip(&start_neighbors) {
+        index
+            .provider()
+            .neighbors()
+            .set_neighbors_sync(start_id as usize, neighbors)?;
+    }
+
+    let total_time = MicroSeconds::from(started.elapsed());
+    let per_vector = MicroSeconds::new(
+        total_time
+            .as_micros()
+            .checked_div(u64::try_from(npoints)?)
+            .context("PiPNN build received an empty dataset")?,
+    );
+    let stats = BuildStats {
+        kind: BuildKind::PiPNN,
+        total_time,
+        vectors_inserted: npoints,
+        insert_latencies: percentiles::compute_percentiles(&mut [per_vector])?,
+    };
+    Ok((index, stats))
+}
+
 #[cfg(any(feature = "scalar-quantization", feature = "spherical-quantization"))]
 pub(crate) fn only_single_insert<DP, T, S>(
     index: Arc<DiskANNIndex<DP>>,
@@ -153,6 +268,8 @@ where
 pub(crate) enum BuildKind {
     SingleInsert,
     MultiInsert,
+    #[cfg(feature = "pipnn")]
+    PiPNN,
 }
 
 impl std::fmt::Display for BuildKind {
@@ -160,6 +277,8 @@ impl std::fmt::Display for BuildKind {
         match self {
             Self::SingleInsert => write!(f, "single insert"),
             Self::MultiInsert => write!(f, "multi insert"),
+            #[cfg(feature = "pipnn")]
+            Self::PiPNN => write!(f, "PiPNN"),
         }
     }
 }
@@ -285,3 +404,55 @@ where
 
     Ok(index)
 }
+
+#[cfg(all(test, feature = "pipnn"))]
+mod pipnn_tests {
+    use super::*;
+
+    #[test]
+    fn dedicated_pipeline_respects_the_requested_start_strategy() {
+        let input: IndexBuild = serde_json::from_value(serde_json::json!({
+            "data_type": "float32",
+            "data": "unused.fbin",
+            "distance": "squared_l2",
+            "max_degree": 4,
+            "l_build": 8,
+            "start_point_strategy": "first_vector",
+            "alpha": 1.2,
+            "backedge_ratio": 1.0,
+            "num_threads": 2,
+            "multi_insert": null,
+            "save_path": null,
+            "build_algorithm": {
+                "algorithm": "PiPNN",
+                "c_max": 8,
+                "c_min": 2,
+                "p_samp": 0.5,
+                "fanout": [2],
+                "k": 2,
+                "replicas": 1
+            }
+        }))
+        .unwrap();
+        let diskann_disk::BuildAlgorithm::PiPNN(parameters) = input.build_algorithm() else {
+            panic!("expected PiPNN parameters");
+        };
+        let mut data = Matrix::new(0.0_f32, 16, 2);
+        for (index, row) in data.row_iter_mut().enumerate() {
+            row.copy_from_slice(&[index as f32, (index % 3) as f32]);
+        }
+
+        let (index, stats) = pipnn_build(Arc::new(data), &input, parameters).unwrap();
+        assert_eq!(stats.vectors_inserted, 16);
+        let starts = index.provider().starting_points().unwrap();
+        assert_eq!(starts.len(), 1);
+        // SAFETY: the completed build has no concurrent vector writers.
+        let start = unsafe {
+            index
+                .data_provider
+                .base_vectors
+                .get_vector_sync(starts[0] as usize)
+        };
+        assert_eq!(start, [0.0, 0.0]);
+    }
+}

From fc865c520dbf17b1abaa76e84b6781aae4d75cfb Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 12:52:33 +0000
Subject: [PATCH 03/21] test(benchmark): cover PiPNN build pipelines

---
 .../example/pipnn-disk-index.json             | 46 ++++++++++++++++++
 .../example/pipnn-graph-index.json            | 47 +++++++++++++++++++
 diskann-benchmark/src/main.rs                 | 19 ++++++++
 3 files changed, 112 insertions(+)
 create mode 100644 diskann-benchmark/example/pipnn-disk-index.json
 create mode 100644 diskann-benchmark/example/pipnn-graph-index.json

diff --git a/diskann-benchmark/example/pipnn-disk-index.json b/diskann-benchmark/example/pipnn-disk-index.json
new file mode 100644
index 0000000000..8602049e50
--- /dev/null
+++ b/diskann-benchmark/example/pipnn-disk-index.json
@@ -0,0 +1,46 @@
+{
+  "search_directories": [
+    "test_data/disk_index_search"
+  ],
+  "jobs": [
+    {
+      "type": "disk-index",
+      "content": {
+        "source": {
+          "disk-index-source": "Build",
+          "data_type": "float32",
+          "data": "disk_index_siftsmall_learn_256pts_data.fbin",
+          "distance": "squared_l2",
+          "dim": 128,
+          "max_degree": 32,
+          "l_build": 50,
+          "alpha": 1.2,
+          "num_threads": 1,
+          "build_ram_limit_gb": 2.0,
+          "num_pq_chunks": 128,
+          "build_algorithm": {
+            "algorithm": "PiPNN",
+            "c_max": 64,
+            "c_min": 16,
+            "p_samp": 0.05,
+            "fanout": [4, 2],
+            "k": 2,
+            "replicas": 1
+          },
+          "save_path": "pipnn_disk_index"
+        },
+        "search_phase": {
+          "queries": "disk_index_sample_query_10pts.fbin",
+          "groundtruth": "disk_index_10pts_idx_uint32_exact_ground_truth.bin",
+          "search_list": [10, 20, 40],
+          "beam_width": 4,
+          "recall_at": 10,
+          "num_threads": 1,
+          "is_flat_search": false,
+          "distance": "squared_l2",
+          "vector_filters_file": null
+        }
+      }
+    }
+  ]
+}
diff --git a/diskann-benchmark/example/pipnn-graph-index.json b/diskann-benchmark/example/pipnn-graph-index.json
new file mode 100644
index 0000000000..3ed4d62dbb
--- /dev/null
+++ b/diskann-benchmark/example/pipnn-graph-index.json
@@ -0,0 +1,47 @@
+{
+  "search_directories": [
+    "test_data/disk_index_search"
+  ],
+  "jobs": [
+    {
+      "type": "graph-index-build",
+      "content": {
+        "source": {
+          "index-source": "Build",
+          "data_type": "float32",
+          "data": "disk_index_siftsmall_learn_256pts_data.fbin",
+          "distance": "squared_l2",
+          "max_degree": 32,
+          "l_build": 50,
+          "alpha": 1.2,
+          "backedge_ratio": 1.0,
+          "num_threads": 1,
+          "start_point_strategy": "first_vector",
+          "build_algorithm": {
+            "algorithm": "PiPNN",
+            "c_max": 64,
+            "c_min": 16,
+            "p_samp": 0.05,
+            "fanout": [4, 2],
+            "k": 2,
+            "replicas": 1
+          }
+        },
+        "search_phase": {
+          "search-type": "topk",
+          "queries": "disk_index_sample_query_10pts.fbin",
+          "groundtruth": "disk_index_10pts_idx_uint32_exact_ground_truth.bin",
+          "reps": 1,
+          "num_threads": [1],
+          "runs": [
+            {
+              "search_n": 10,
+              "search_l": [10, 20, 40],
+              "recall_k": 10
+            }
+          ]
+        }
+      }
+    }
+  ]
+}
diff --git a/diskann-benchmark/src/main.rs b/diskann-benchmark/src/main.rs
index 13cf1ec31f..1ed1b2a150 100644
--- a/diskann-benchmark/src/main.rs
+++ b/diskann-benchmark/src/main.rs
@@ -312,6 +312,25 @@ mod tests {
         run_integration_test(raw);
     }
 
+    #[test]
+    #[cfg(feature = "pipnn")]
+    fn pipnn_graph_index_integration() {
+        let raw = value_from_file(&example_directory().join("pipnn-graph-index.json"));
+        run_integration_test(raw);
+    }
+
+    #[test]
+    #[cfg(all(feature = "disk-index", feature = "pipnn"))]
+    fn pipnn_disk_index_integration() {
+        let mut raw = value_from_file(&example_directory().join("pipnn-disk-index.json"));
+        let directory = tempfile::tempdir().unwrap();
+        let save_path = directory.path().join("pipnn_disk_index");
+        *raw.pointer_mut("/jobs/0/content/source/save_path")
+            .expect("PiPNN disk example must declare save_path") =
+            Value::String(save_path.to_string_lossy().into_owned());
+        run_integration_test(raw);
+    }
+
     /////////////////////////
     //    Flat Search      //
     /////////////////////////

From bb98da295866a3ef2e5f9f6dd07007add8b23fec Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 13:14:23 +0000
Subject: [PATCH 04/21] benchmark: exclude provider allocation from build
 timing

---
 diskann-benchmark/src/index/build.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index b696e64a93..ed4f1d2865 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -171,6 +171,7 @@ where
         .into_iter()
         .map(|source| adjacency[source].clone())
         .collect();
+    let batch_elapsed = started.elapsed();
 
     // Unlike incremental insertion, PiPNN finishes all batch-build scratch
     // before allocating the provider that owns the searchable index.
@@ -179,6 +180,9 @@ where
         input.inmem_parameters(npoints, dimensions),
         common::NoDeletes,
     )?;
+    // Provider allocation is setup, just as it is for the incremental path;
+    // BuildStats covers graph construction plus provider installation.
+    let install_started = std::time::Instant::now();
     for (id, vector) in data.row_iter().enumerate() {
         let id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
         index.data_provider.base_vectors.set_element(&id, vector)?;
@@ -204,7 +208,7 @@ where
             .set_neighbors_sync(start_id as usize, neighbors)?;
     }
 
-    let total_time = MicroSeconds::from(started.elapsed());
+    let total_time = MicroSeconds::from(batch_elapsed + install_started.elapsed());
     let per_vector = MicroSeconds::new(
         total_time
             .as_micros()

From aeb65e2be54a5f3f58441aeaa0a8d16b701dff9b Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 13:48:26 +0000
Subject: [PATCH 05/21] docs(benchmark): clarify PiPNN memory-budget scope

---
 diskann-benchmark/src/inputs/disk.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/diskann-benchmark/src/inputs/disk.rs b/diskann-benchmark/src/inputs/disk.rs
index d1a815969d..2ac456ed9e 100644
--- a/diskann-benchmark/src/inputs/disk.rs
+++ b/diskann-benchmark/src/inputs/disk.rs
@@ -69,8 +69,8 @@ pub(crate) struct DiskIndexBuild {
     #[serde(default = "default_alpha")]
     pub(crate) alpha: f32,
     pub(crate) num_threads: usize,
-    /// Graph-build RAM limit. PiPNN falls back to Vamana when its estimated
-    /// one-shot peak exceeds this value.
+    /// Memory budget for bounded disk-index pipeline stages. Explicit PiPNN
+    /// graph construction remains one-shot and is not silently replaced.
     pub(crate) build_ram_limit_gb: f64,
     pub(crate) num_pq_chunks: NonZeroUsize,
     #[cfg(feature = "disk-index")]

From ba3759a51cf7c055003a62a6e91f206f89d96ed6 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 13:49:09 +0000
Subject: [PATCH 06/21] build(benchmark): update PiPNN feature lockfile

---
 Cargo.lock | 1 +
 1 file changed, 1 insertion(+)

diff --git a/Cargo.lock b/Cargo.lock
index 447fa9a287..0c20207c6b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -476,6 +476,7 @@ dependencies = [
  "diskann-disk",
  "diskann-inmem",
  "diskann-label-filter",
+ "diskann-pipnn",
  "diskann-providers",
  "diskann-quantization",
  "diskann-tools",

From 0c93372f70b71a0d7aec2f5a5da48dff8b2bbd4f Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 13:53:03 +0000
Subject: [PATCH 07/21] test(benchmark): assert PiPNN registered results

---
 diskann-benchmark/src/main.rs | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/diskann-benchmark/src/main.rs b/diskann-benchmark/src/main.rs
index 1ed1b2a150..90ca28de02 100644
--- a/diskann-benchmark/src/main.rs
+++ b/diskann-benchmark/src/main.rs
@@ -266,7 +266,7 @@ mod tests {
         }
     }
 
-    fn run_integration_test(mut raw: serde_json::Value) {
+    fn run_integration_test(mut raw: serde_json::Value) -> Vec<Value> {
         // First, parse and modify the input file to establish paths relative to the
         // directory building the dispatcher.
         // let mut raw = serde_json::from_str(json_string).unwrap();
@@ -301,6 +301,7 @@ mod tests {
 
         let results: Vec<Value> = load_from_file(&output_path);
         assert_eq!(results.len(), num_jobs(&raw));
+        results
     }
 
     ////////////////////////////////
@@ -316,7 +317,11 @@ mod tests {
     #[cfg(feature = "pipnn")]
     fn pipnn_graph_index_integration() {
         let raw = value_from_file(&example_directory().join("pipnn-graph-index.json"));
-        run_integration_test(raw);
+        let results = run_integration_test(raw);
+        let result = &results[0]["results"];
+        assert_eq!(result["build"]["kind"], "PiPNN");
+        assert_eq!(result["build"]["vectors_inserted"], 256);
+        assert_eq!(result["search"]["Topk"].as_array().unwrap().len(), 3);
     }
 
     #[test]

From a81f7711996d59c4ffb7b9a43ee48e315c1b0e71 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 29 Jul 2026 16:33:57 +0000
Subject: [PATCH 08/21] fix(benchmark): connect start slots to real graph
 sources

---
 diskann-benchmark/src/index/build.rs | 26 +++++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index ed4f1d2865..17d77561e8 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -5,6 +5,8 @@
 
 use std::{num::NonZeroUsize, sync::Arc};
 
+#[cfg(feature = "pipnn")]
+use diskann::graph::AdjacencyList;
 use diskann::{
     graph::{DiskANNIndex, StartPointStrategy},
     provider::{self, DataProvider, DefaultContext},
@@ -167,10 +169,18 @@ where
                 .context("PiPNN cannot connect a start point to an empty dataset")
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
-    let start_neighbors: Vec<_> = start_sources
+    let degree = graph.pruned_degree().get();
+    let start_neighbors = start_sources
         .into_iter()
-        .map(|source| adjacency[source].clone())
-        .collect();
+        .map(|source| {
+            let source_id = u32::try_from(source).context("PiPNN start source exceeds u32::MAX")?;
+            let mut neighbors = AdjacencyList::with_capacity(degree);
+            neighbors.push(source_id);
+            neighbors.extend_from_slice(&adjacency[source]);
+            neighbors.truncate(degree);
+            Ok(neighbors)
+        })
+        .collect::<anyhow::Result<Vec<_>>>()?;
     let batch_elapsed = started.elapsed();
 
     // Unlike incremental insertion, PiPNN finishes all batch-build scratch
@@ -458,5 +468,15 @@ mod pipnn_tests {
                 .get_vector_sync(starts[0] as usize)
         };
         assert_eq!(start, [0.0, 0.0]);
+        let mut neighbors = AdjacencyList::new();
+        index
+            .provider()
+            .neighbors()
+            .get_neighbors_sync(starts[0] as usize, &mut neighbors)
+            .unwrap();
+        assert!(
+            neighbors.contains(0),
+            "start slot must enter the real graph"
+        );
     }
 }

From de207fe0e306c8866e1a8a42a7f2c630e358c502 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Fri, 31 Jul 2026 03:23:45 +0000
Subject: [PATCH 09/21] docs(benchmark): map PiPNN batch lifecycle

---
 diskann-benchmark/src/index/benchmarks.rs |  3 +++
 diskann-benchmark/src/index/build.rs      | 31 +++++++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/diskann-benchmark/src/index/benchmarks.rs b/diskann-benchmark/src/index/benchmarks.rs
index 31f965f04a..5555af8ed0 100644
--- a/diskann-benchmark/src/index/benchmarks.rs
+++ b/diskann-benchmark/src/index/benchmarks.rs
@@ -233,6 +233,9 @@ where
                         single_or_multi_insert,
                     )
                 };
+                // PiPNN is selected before constructing the incremental provider:
+                // explicit batch selection must never allocate an incremental
+                // index and then silently fall through to Vamana insertion.
                 #[cfg(feature = "pipnn")]
                 let result = match build.build_algorithm() {
                     diskann_disk::BuildAlgorithm::PiPNN(parameters) => {
diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 17d77561e8..79f5b8c663 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -115,6 +115,28 @@ where
 }
 
 #[cfg(feature = "pipnn")]
+/// Run the dedicated batch lifecycle and install its result in a searchable provider.
+///
+/// Incremental Vamana interleaves provider allocation with per-point insertion.
+/// PiPNN must not use that path: its partitions and candidate graph are complete
+/// before a provider exists. The benchmark lifecycle is therefore:
+///
+/// ```text
+/// Arc<Matrix<T>> ──> PiPNN core ──> Vec<AdjacencyList<u32>>
+///       │                                      │
+///       ├──> start vectors ──> source IDs      │
+///       │                                      v
+///       └──────────────────────────────> provider installation ──> MemoryIndex
+/// ```
+///
+/// | Interval | Included in build statistics | Large owner |
+/// | --- | --- | --- |
+/// | batch graph + start computation | yes | input matrix, core stage outputs |
+/// | provider allocation | no (setup, matching incremental path) | empty provider |
+/// | vectors, edges, and start-slot installation | yes | searchable provider |
+///
+/// The returned provider is a benchmark/search adapter; provider concerns do not
+/// leak back into `diskann-pipnn`.
 pub(crate) fn pipnn_build<T>(
     data: Arc<Matrix<T>>,
     input: &IndexBuild,
@@ -150,6 +172,10 @@ where
             )
         })?;
     let distance = T::distance(metric, Some(dimensions));
+    // Provider start vectors occupy frozen slots outside the real point-ID
+    // range. Connect each slot to an identical source row when one exists;
+    // synthetic strategies fall back to the nearest real row under the build
+    // metric. `total_cmp` gives deterministic ordering for all finite/NaN bits.
     let start_sources = start_points
         .row_iter()
         .map(|start| {
@@ -170,6 +196,9 @@ where
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
     let degree = graph.pruned_degree().get();
+    // Seed every frozen start slot with its source point followed by that
+    // point's outgoing row. Truncation applies the same graph degree bound as
+    // real rows, and including the source guarantees entry into the real graph.
     let start_neighbors = start_sources
         .into_iter()
         .map(|source| {
@@ -193,6 +222,8 @@ where
     // Provider allocation is setup, just as it is for the incremental path;
     // BuildStats covers graph construction plus provider installation.
     let install_started = std::time::Instant::now();
+    // Install vectors before edges so a successfully returned MemoryIndex never
+    // exposes graph IDs whose source vectors are absent.
     for (id, vector) in data.row_iter().enumerate() {
         let id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
         index.data_provider.base_vectors.set_element(&id, vector)?;

From 856c70256f09f6c0ba5df8e1ee38f9783184c97c Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Mon, 3 Aug 2026 02:13:40 +0000
Subject: [PATCH 10/21] fix(benchmark): preserve PiPNN start adjacency

---
 diskann-benchmark/src/index/build.rs | 35 ++++++++++++----------------
 1 file changed, 15 insertions(+), 20 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 79f5b8c663..8dcbe434ef 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -5,8 +5,6 @@
 
 use std::{num::NonZeroUsize, sync::Arc};
 
-#[cfg(feature = "pipnn")]
-use diskann::graph::AdjacencyList;
 use diskann::{
     graph::{DiskANNIndex, StartPointStrategy},
     provider::{self, DataProvider, DefaultContext},
@@ -195,21 +193,14 @@ where
                 .context("PiPNN cannot connect a start point to an empty dataset")
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
-    let degree = graph.pruned_degree().get();
-    // Seed every frozen start slot with its source point followed by that
-    // point's outgoing row. Truncation applies the same graph degree bound as
-    // real rows, and including the source guarantees entry into the real graph.
-    let start_neighbors = start_sources
+    // A frozen start slot carries the chosen source vector, so expanding it
+    // must expose exactly that real source's outgoing row. Prepending the source
+    // ID would consume one degree slot and discard a graph edge, changing every
+    // search from the graph produced by the core builder.
+    let start_neighbors: Vec<_> = start_sources
         .into_iter()
-        .map(|source| {
-            let source_id = u32::try_from(source).context("PiPNN start source exceeds u32::MAX")?;
-            let mut neighbors = AdjacencyList::with_capacity(degree);
-            neighbors.push(source_id);
-            neighbors.extend_from_slice(&adjacency[source]);
-            neighbors.truncate(degree);
-            Ok(neighbors)
-        })
-        .collect::<anyhow::Result<Vec<_>>>()?;
+        .map(|source| adjacency[source].clone())
+        .collect();
     let batch_elapsed = started.elapsed();
 
     // Unlike incremental insertion, PiPNN finishes all batch-build scratch
@@ -453,6 +444,7 @@ where
 #[cfg(all(test, feature = "pipnn"))]
 mod pipnn_tests {
     use super::*;
+    use diskann::graph::AdjacencyList;
 
     #[test]
     fn dedicated_pipeline_respects_the_requested_start_strategy() {
@@ -505,9 +497,12 @@ mod pipnn_tests {
             .neighbors()
             .get_neighbors_sync(starts[0] as usize, &mut neighbors)
             .unwrap();
-        assert!(
-            neighbors.contains(0),
-            "start slot must enter the real graph"
-        );
+        let mut source_neighbors = AdjacencyList::new();
+        index
+            .provider()
+            .neighbors()
+            .get_neighbors_sync(0, &mut source_neighbors)
+            .unwrap();
+        assert_eq!(neighbors, source_neighbors);
     }
 }

From 63de4a7b37026d3e4c9f93d951d1bd31cc64fcee Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Wed, 5 Aug 2026 07:34:21 +0000
Subject: [PATCH 11/21] refactor(benchmark): use diskann pipnn module

Forward the benchmark feature through diskann and keep the dedicated batch lifecycle without a direct implementation-crate dependency.
---
 Cargo.lock                           |  1 -
 diskann-benchmark/Cargo.toml         |  3 +--
 diskann-benchmark/src/index/build.rs | 12 ++++++++----
 3 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 0c20207c6b..447fa9a287 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -476,7 +476,6 @@ dependencies = [
  "diskann-disk",
  "diskann-inmem",
  "diskann-label-filter",
- "diskann-pipnn",
  "diskann-providers",
  "diskann-quantization",
  "diskann-tools",
diff --git a/diskann-benchmark/Cargo.toml b/diskann-benchmark/Cargo.toml
index 2005d639ed..8d7e5eaec5 100644
--- a/diskann-benchmark/Cargo.toml
+++ b/diskann-benchmark/Cargo.toml
@@ -32,7 +32,6 @@ diskann-label-filter.workspace = true
 diskann-tools = { workspace = true }
 diskann-disk = { workspace = true, optional = true }
 diskann-bftree = { workspace = true, optional = true }
-diskann-pipnn = { workspace = true, optional = true }
 cfg-if.workspace = true
 diskann-benchmark-runner = { workspace = true }
 opentelemetry = { workspace = true, optional = true }
@@ -85,4 +84,4 @@ disk-index = [
 ]
 
 # Enable PiPNN graph construction.
-pipnn = ["dep:diskann-pipnn", "diskann-disk/pipnn"]
+pipnn = ["diskann/pipnn", "diskann-disk/pipnn"]
diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 8dcbe434ef..7fefe86aa4 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -134,7 +134,7 @@ where
 /// | vectors, edges, and start-slot installation | yes | searchable provider |
 ///
 /// The returned provider is a benchmark/search adapter; provider concerns do not
-/// leak back into `diskann-pipnn`.
+/// leak back into `diskann::graph::pipnn`.
 pub(crate) fn pipnn_build<T>(
     data: Arc<Matrix<T>>,
     input: &IndexBuild,
@@ -156,9 +156,13 @@ where
 
     let started = std::time::Instant::now();
     let adjacency = {
-        let context =
-            diskann_pipnn::PiPNNBuildContext::new(parameters.into(), &graph, metric, &pool)?;
-        diskann_pipnn::build_graph(data.as_view(), &context)?
+        let context = diskann::graph::pipnn::PiPNNBuildContext::new(
+            parameters.into(),
+            &graph,
+            metric,
+            &pool,
+        )?;
+        diskann::graph::pipnn::build_graph(data.as_view(), &context)?
     };
     let start_points = input
         .start_point_strategy()

From 9a3be65d540102fbab4d3349b1a92c4c52201378 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Thu, 6 Aug 2026 03:29:59 +0000
Subject: [PATCH 12/21] fix(benchmark): fail closed on unavailable pipnn

Reject PiPNN graph inputs when the feature is disabled and stop reporting synthetic per-insert percentiles for batch construction.
---
 diskann-benchmark/src/index/build.rs          | 28 +++++------
 .../src/index/streaming/stats.rs              | 12 +++--
 diskann-benchmark/src/inputs/graph_index.rs   | 49 +++++++++++++++++++
 3 files changed, 72 insertions(+), 17 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 7fefe86aa4..0cf7516026 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -245,17 +245,11 @@ where
     }
 
     let total_time = MicroSeconds::from(batch_elapsed + install_started.elapsed());
-    let per_vector = MicroSeconds::new(
-        total_time
-            .as_micros()
-            .checked_div(u64::try_from(npoints)?)
-            .context("PiPNN build received an empty dataset")?,
-    );
     let stats = BuildStats {
         kind: BuildKind::PiPNN,
         total_time,
         vectors_inserted: npoints,
-        insert_latencies: percentiles::compute_percentiles(&mut [per_vector])?,
+        insert_latencies: None,
     };
     Ok((index, stats))
 }
@@ -328,7 +322,8 @@ pub(crate) struct BuildStats {
     pub(crate) kind: BuildKind,
     pub(crate) total_time: MicroSeconds,
     pub(crate) vectors_inserted: usize,
-    pub(crate) insert_latencies: percentiles::Percentiles<MicroSeconds>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) insert_latencies: Option<percentiles::Percentiles<MicroSeconds>>,
 }
 
 impl BuildStats {
@@ -349,7 +344,7 @@ impl BuildStats {
             kind,
             total_time,
             vectors_inserted,
-            insert_latencies: percentiles::compute_percentiles(&mut latencies)?,
+            insert_latencies: Some(percentiles::compute_percentiles(&mut latencies)?),
         })
     }
 }
@@ -359,11 +354,15 @@ impl std::fmt::Display for BuildStats {
         writeln!(f, "Index Build Time: {}s", self.total_time.as_seconds())?;
         writeln!(f, "Vectors Inserted: {}", self.vectors_inserted)?;
         writeln!(f, "Kind: {}", self.kind)?;
-        write!(
-            f,
-            "Insert Latencies:\n  average: {}us\n      p90: {}\n      p99: {}\n\n",
-            self.insert_latencies.mean, self.insert_latencies.p90, self.insert_latencies.p99,
-        )
+        if let Some(latencies) = &self.insert_latencies {
+            write!(
+                f,
+                "Insert Latencies:\n  average: {}us\n      p90: {}\n      p99: {}\n\n",
+                latencies.mean, latencies.p90, latencies.p99,
+            )
+        } else {
+            writeln!(f, "Insert Latencies: not measured for batch construction\n")
+        }
     }
 }
 
@@ -485,6 +484,7 @@ mod pipnn_tests {
 
         let (index, stats) = pipnn_build(Arc::new(data), &input, parameters).unwrap();
         assert_eq!(stats.vectors_inserted, 16);
+        assert!(stats.insert_latencies.is_none());
         let starts = index.provider().starting_points().unwrap();
         assert_eq!(starts.len(), 1);
         // SAFETY: the completed build has no concurrent vector writers.
diff --git a/diskann-benchmark/src/index/streaming/stats.rs b/diskann-benchmark/src/index/streaming/stats.rs
index 4ab5e3ed47..1e431f8af1 100644
--- a/diskann-benchmark/src/index/streaming/stats.rs
+++ b/diskann-benchmark/src/index/streaming/stats.rs
@@ -179,9 +179,15 @@ where
                         let mut r = table.row(row);
                         r.insert(stage, 0);
                         r.insert(stats.kind, 1);
-                        r.insert(format!("{:.1}us", stats.insert_latencies.mean), 2);
-                        r.insert(stats.insert_latencies.p90, 3);
-                        r.insert(stats.insert_latencies.p99, 4);
+                        if let Some(latencies) = &stats.insert_latencies {
+                            r.insert(format!("{:.1}us", latencies.mean), 2);
+                            r.insert(latencies.p90, 3);
+                            r.insert(latencies.p99, 4);
+                        } else {
+                            r.insert("not measured", 2);
+                            r.insert("-", 3);
+                            r.insert("-", 4);
+                        }
 
                         row += 1;
                         stage += 1;
diff --git a/diskann-benchmark/src/inputs/graph_index.rs b/diskann-benchmark/src/inputs/graph_index.rs
index 1397eecbfe..3fd991461b 100644
--- a/diskann-benchmark/src/inputs/graph_index.rs
+++ b/diskann-benchmark/src/inputs/graph_index.rs
@@ -725,6 +725,30 @@ pub(crate) struct IndexBuild {
     #[cfg(feature = "pipnn")]
     #[serde(default)]
     build_algorithm: diskann_disk::BuildAlgorithm,
+    #[cfg(not(feature = "pipnn"))]
+    #[serde(default)]
+    build_algorithm: Option<serde_json::Value>,
+}
+
+#[cfg(not(feature = "pipnn"))]
+fn validate_disabled_build_algorithm(
+    build_algorithm: Option<&serde_json::Value>,
+) -> Result<(), anyhow::Error> {
+    let Some(build_algorithm) = build_algorithm else {
+        return Ok(());
+    };
+    let algorithm = build_algorithm
+        .as_object()
+        .and_then(|object| object.get("algorithm"))
+        .and_then(serde_json::Value::as_str)
+        .ok_or_else(|| anyhow!("build_algorithm must be an object containing an algorithm tag"))?;
+    match algorithm {
+        "Vamana" => Ok(()),
+        "PiPNN" => Err(anyhow!(
+            "PiPNN graph construction requires the `pipnn` feature"
+        )),
+        other => Err(anyhow!("unrecognized graph build algorithm `{other}`")),
+    }
 }
 
 impl IndexBuild {
@@ -830,6 +854,9 @@ impl IndexBuild {
     }
 
     pub(crate) fn validate(&mut self, checker: &mut Checker) -> Result<(), anyhow::Error> {
+        #[cfg(not(feature = "pipnn"))]
+        validate_disabled_build_algorithm(self.build_algorithm.as_ref())?;
+
         self.data.resolve(checker)?;
 
         // We allow overwriting of already existing save paths, since users like to do this
@@ -909,6 +936,8 @@ impl Example for IndexBuild {
             save_path: None,
             #[cfg(feature = "pipnn")]
             build_algorithm: diskann_disk::BuildAlgorithm::Vamana,
+            #[cfg(not(feature = "pipnn"))]
+            build_algorithm: None,
         }
     }
 }
@@ -1499,3 +1528,23 @@ impl std::fmt::Display for DynamicIndexRun {
         Ok(())
     }
 }
+
+#[cfg(all(test, not(feature = "pipnn")))]
+mod disabled_pipnn_tests {
+    use super::*;
+
+    #[test]
+    fn pipnn_request_fails_closed_without_the_feature() {
+        let pipnn = serde_json::json!({
+            "algorithm": "PiPNN",
+            "c_max": 512,
+            "fanout": [8, 3]
+        });
+        let error = validate_disabled_build_algorithm(Some(&pipnn)).unwrap_err();
+        assert!(error.to_string().contains("requires the `pipnn` feature"));
+
+        let vamana = serde_json::json!({ "algorithm": "Vamana" });
+        validate_disabled_build_algorithm(Some(&vamana)).unwrap();
+        validate_disabled_build_algorithm(None).unwrap();
+    }
+}

From 9000c187b452bba62ec6ad92b0781e6b075c96dd Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Thu, 6 Aug 2026 04:06:53 +0000
Subject: [PATCH 13/21] fix(benchmark): reject pipnn dynamic runs

PiPNN is a one-shot batch builder; dynamic graph jobs now fail instead of silently using incremental Vamana.
---
 diskann-benchmark/src/inputs/graph_index.rs | 33 +++++++++++++++++++++
 1 file changed, 33 insertions(+)

diff --git a/diskann-benchmark/src/inputs/graph_index.rs b/diskann-benchmark/src/inputs/graph_index.rs
index 3fd991461b..b2254bda10 100644
--- a/diskann-benchmark/src/inputs/graph_index.rs
+++ b/diskann-benchmark/src/inputs/graph_index.rs
@@ -730,6 +730,21 @@ pub(crate) struct IndexBuild {
     build_algorithm: Option<serde_json::Value>,
 }
 
+#[cfg(feature = "pipnn")]
+fn validate_dynamic_build_algorithm(
+    build_algorithm: &diskann_disk::BuildAlgorithm,
+) -> Result<(), anyhow::Error> {
+    match build_algorithm {
+        diskann_disk::BuildAlgorithm::Vamana => Ok(()),
+        diskann_disk::BuildAlgorithm::PiPNN(_) => Err(anyhow!(
+            "PiPNN is a batch builder and is not supported by graph-index-dynamic-run"
+        )),
+        _ => Err(anyhow!(
+            "the selected graph build algorithm is not supported by graph-index-dynamic-run"
+        )),
+    }
+}
+
 #[cfg(not(feature = "pipnn"))]
 fn validate_disabled_build_algorithm(
     build_algorithm: Option<&serde_json::Value>,
@@ -1481,6 +1496,9 @@ impl DynamicIndexRun {
     }
 
     pub(crate) fn validate(&mut self, checker: &mut Checker) -> anyhow::Result<()> {
+        #[cfg(feature = "pipnn")]
+        validate_dynamic_build_algorithm(self.build.build_algorithm())?;
+
         self.build.validate(checker)?;
         self.runbook_params.validate(checker)?;
         self.search_phase.validate(checker)?;
@@ -1529,6 +1547,21 @@ impl std::fmt::Display for DynamicIndexRun {
     }
 }
 
+#[cfg(all(test, feature = "pipnn"))]
+mod dynamic_pipnn_tests {
+    use super::*;
+
+    #[test]
+    fn dynamic_run_rejects_the_batch_only_pipnn_algorithm() {
+        let error = validate_dynamic_build_algorithm(&diskann_disk::BuildAlgorithm::PiPNN(
+            diskann_disk::PiPNNParameters::default(),
+        ))
+        .unwrap_err();
+        assert!(error.to_string().contains("not supported"));
+        validate_dynamic_build_algorithm(&diskann_disk::BuildAlgorithm::Vamana).unwrap();
+    }
+}
+
 #[cfg(all(test, not(feature = "pipnn")))]
 mod disabled_pipnn_tests {
     use super::*;

From ea200b37ae087904a6ba914a07e4018072c8194e Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Thu, 6 Aug 2026 08:57:32 +0000
Subject: [PATCH 14/21] fix(benchmark): adapt pipnn errors to main

---
 diskann-benchmark/src/index/build.rs | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 0cf7516026..abe36b4df7 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -167,12 +167,7 @@ where
     let start_points = input
         .start_point_strategy()
         .compute(data.as_view())
-        .map_err(|error| {
-            ANNError::new(
-                diskann::ANNErrorKind::DiskANN(StartPointComputeError),
-                error,
-            )
-        })?;
+        .map_err(ANNError::new)?;
     let distance = T::distance(metric, Some(dimensions));
     // Provider start vectors occupy frozen slots outside the real point-ID
     // range. Connect each slot to an identical source row when one exists;

From 4f2429e06048ba7330bc1b07838c858f609812b5 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Thu, 6 Aug 2026 09:43:07 +0000
Subject: [PATCH 15/21] fix(benchmark): allow platform-sized input enum

---
 diskann-benchmark/src/inputs/graph_index.rs | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/diskann-benchmark/src/inputs/graph_index.rs b/diskann-benchmark/src/inputs/graph_index.rs
index b2254bda10..a10e664074 100644
--- a/diskann-benchmark/src/inputs/graph_index.rs
+++ b/diskann-benchmark/src/inputs/graph_index.rs
@@ -969,6 +969,10 @@ impl std::fmt::Display for IndexBuild {
 
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(tag = "index-source")] // Use tagged enums for JSON
+#[allow(
+    clippy::large_enum_variant,
+    reason = "build configuration is parsed once; boxing would add indirection to every source access"
+)]
 pub enum IndexSource {
     Load(IndexLoad),
     Build(IndexBuild),

From f401eebbdbf4812b625acba1980c0aaca2f611ea Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Fri, 7 Aug 2026 03:14:50 +0000
Subject: [PATCH 16/21] docs(benchmark): complete vector safety proof

---
 diskann-benchmark/src/index/build.rs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index abe36b4df7..ccccaa10e8 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -482,7 +482,9 @@ mod pipnn_tests {
         assert!(stats.insert_latencies.is_none());
         let starts = index.provider().starting_points().unwrap();
         assert_eq!(starts.len(), 1);
-        // SAFETY: the completed build has no concurrent vector writers.
+        // SAFETY: `starting_points` returns provider-installed frozen IDs, so
+        // `starts[0] < base_vectors.total()`. The completed build has no concurrent
+        // vector writers, so no mutable alias exists for the returned shared slice.
         let start = unsafe {
             index
                 .data_provider

From c50648cf8e269b078cb687bc013991a1115dd294 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Fri, 7 Aug 2026 12:12:09 +0000
Subject: [PATCH 17/21] docs(pipnn): simplify benchmark lifecycle comments

---
 diskann-benchmark/src/index/benchmarks.rs |  5 +--
 diskann-benchmark/src/index/build.rs      | 55 ++++++++---------------
 diskann-benchmark/src/inputs/disk.rs      |  6 +--
 3 files changed, 23 insertions(+), 43 deletions(-)

diff --git a/diskann-benchmark/src/index/benchmarks.rs b/diskann-benchmark/src/index/benchmarks.rs
index 5555af8ed0..88ef21673c 100644
--- a/diskann-benchmark/src/index/benchmarks.rs
+++ b/diskann-benchmark/src/index/benchmarks.rs
@@ -233,9 +233,8 @@ where
                         single_or_multi_insert,
                     )
                 };
-                // PiPNN is selected before constructing the incremental provider:
-                // explicit batch selection must never allocate an incremental
-                // index and then silently fall through to Vamana insertion.
+                // PiPNN uses the batch build path. Do not create an incremental
+                // provider or run Vamana insertion.
                 #[cfg(feature = "pipnn")]
                 let result = match build.build_algorithm() {
                     diskann_disk::BuildAlgorithm::PiPNN(parameters) => {
diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index ccccaa10e8..c2bfaccc8a 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -113,28 +113,14 @@ where
 }
 
 #[cfg(feature = "pipnn")]
-/// Run the dedicated batch lifecycle and install its result in a searchable provider.
+/// Build a PiPNN graph and install it in a searchable provider.
 ///
-/// Incremental Vamana interleaves provider allocation with per-point insertion.
-/// PiPNN must not use that path: its partitions and candidate graph are complete
-/// before a provider exists. The benchmark lifecycle is therefore:
+/// PiPNN completes its graph before provider creation. The function also computes
+/// start vectors and their source IDs. It then installs vectors, edges, and start
+/// slots in a new `MemoryIndex`.
 ///
-/// ```text
-/// Arc<Matrix<T>> ──> PiPNN core ──> Vec<AdjacencyList<u32>>
-///       │                                      │
-///       ├──> start vectors ──> source IDs      │
-///       │                                      v
-///       └──────────────────────────────> provider installation ──> MemoryIndex
-/// ```
-///
-/// | Interval | Included in build statistics | Large owner |
-/// | --- | --- | --- |
-/// | batch graph + start computation | yes | input matrix, core stage outputs |
-/// | provider allocation | no (setup, matching incremental path) | empty provider |
-/// | vectors, edges, and start-slot installation | yes | searchable provider |
-///
-/// The returned provider is a benchmark/search adapter; provider concerns do not
-/// leak back into `diskann::graph::pipnn`.
+/// `BuildStats` includes graph construction and provider installation. It does
+/// not include provider allocation.
 pub(crate) fn pipnn_build<T>(
     data: Arc<Matrix<T>>,
     input: &IndexBuild,
@@ -169,10 +155,9 @@ where
         .compute(data.as_view())
         .map_err(ANNError::new)?;
     let distance = T::distance(metric, Some(dimensions));
-    // Provider start vectors occupy frozen slots outside the real point-ID
-    // range. Connect each slot to an identical source row when one exists;
-    // synthetic strategies fall back to the nearest real row under the build
-    // metric. `total_cmp` gives deterministic ordering for all finite/NaN bits.
+    // Start vectors use frozen IDs outside the real point-ID range. Connect each
+    // frozen ID to its source row. For a synthetic start vector, connect it to
+    // the nearest real row. `total_cmp` gives a deterministic total order.
     let start_sources = start_points
         .row_iter()
         .map(|start| {
@@ -192,28 +177,24 @@ where
                 .context("PiPNN cannot connect a start point to an empty dataset")
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
-    // A frozen start slot carries the chosen source vector, so expanding it
-    // must expose exactly that real source's outgoing row. Prepending the source
-    // ID would consume one degree slot and discard a graph edge, changing every
-    // search from the graph produced by the core builder.
+    // A frozen slot stores the selected source vector. Its adjacency must equal
+    // that source row. Adding the source ID would remove one graph edge because
+    // the row has a fixed degree.
     let start_neighbors: Vec<_> = start_sources
         .into_iter()
         .map(|source| adjacency[source].clone())
         .collect();
     let batch_elapsed = started.elapsed();
 
-    // Unlike incremental insertion, PiPNN finishes all batch-build scratch
-    // before allocating the provider that owns the searchable index.
     let index = diskann_async::new_index::<T, _>(
         graph,
         input.inmem_parameters(npoints, dimensions),
         common::NoDeletes,
     )?;
-    // Provider allocation is setup, just as it is for the incremental path;
-    // BuildStats covers graph construction plus provider installation.
+    // `BuildStats` excludes provider allocation and includes provider installation.
     let install_started = std::time::Instant::now();
-    // Install vectors before edges so a successfully returned MemoryIndex never
-    // exposes graph IDs whose source vectors are absent.
+    // Install vectors before edges. A returned index must contain a vector for
+    // every graph ID.
     for (id, vector) in data.row_iter().enumerate() {
         let id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
         index.data_provider.base_vectors.set_element(&id, vector)?;
@@ -482,9 +463,9 @@ mod pipnn_tests {
         assert!(stats.insert_latencies.is_none());
         let starts = index.provider().starting_points().unwrap();
         assert_eq!(starts.len(), 1);
-        // SAFETY: `starting_points` returns provider-installed frozen IDs, so
-        // `starts[0] < base_vectors.total()`. The completed build has no concurrent
-        // vector writers, so no mutable alias exists for the returned shared slice.
+        // SAFETY: `starting_points` returns installed frozen IDs. Therefore,
+        // `starts[0] < base_vectors.total()`. The completed build has no vector
+        // writer, so the returned shared slice has no mutable alias.
         let start = unsafe {
             index
                 .data_provider
diff --git a/diskann-benchmark/src/inputs/disk.rs b/diskann-benchmark/src/inputs/disk.rs
index 2ac456ed9e..0a57dbd442 100644
--- a/diskann-benchmark/src/inputs/disk.rs
+++ b/diskann-benchmark/src/inputs/disk.rs
@@ -69,14 +69,14 @@ pub(crate) struct DiskIndexBuild {
     #[serde(default = "default_alpha")]
     pub(crate) alpha: f32,
     pub(crate) num_threads: usize,
-    /// Memory budget for bounded disk-index pipeline stages. Explicit PiPNN
-    /// graph construction remains one-shot and is not silently replaced.
+    /// Memory budget for disk-index stages that support bounded work.
+    /// PiPNN always uses its one-shot graph build.
     pub(crate) build_ram_limit_gb: f64,
     pub(crate) num_pq_chunks: NonZeroUsize,
     #[cfg(feature = "disk-index")]
     #[serde(default)]
     pub(crate) quantization_type: Option<QuantizationType>,
-    /// Build algorithm: "Vamana" (default) or "PiPNN" with config params.
+    /// Build algorithm: `Vamana` by default, or `PiPNN` with its parameters.
     #[cfg(feature = "disk-index")]
     #[serde(default)]
     pub(crate) build_algorithm: BuildAlgorithm,

From 411d00ba05e0c8848c7fdee29d3b879fa2174bba Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Fri, 7 Aug 2026 12:29:47 +0000
Subject: [PATCH 18/21] docs(pipnn): use direct benchmark wording

---
 diskann-benchmark/src/index/build.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index c2bfaccc8a..39d63f67c2 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -178,8 +178,8 @@ where
         })
         .collect::<anyhow::Result<Vec<_>>>()?;
     // A frozen slot stores the selected source vector. Its adjacency must equal
-    // that source row. Adding the source ID would remove one graph edge because
-    // the row has a fixed degree.
+    // that source row. Adding the source ID removes one graph edge because the
+    // row has a fixed degree.
     let start_neighbors: Vec<_> = start_sources
         .into_iter()
         .map(|source| adjacency[source].clone())

From 75f4a3b8d82dbcea101a78f5223638338fbb1dff Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Mon, 10 Aug 2026 07:00:04 +0000
Subject: [PATCH 19/21] perf(pipnn): score each start candidate once

---
 diskann-benchmark/src/index/build.rs | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index 39d63f67c2..f8b2de462b 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -167,11 +167,8 @@ where
                 .or_else(|| {
                     data.row_iter()
                         .enumerate()
-                        .min_by(|(_, left), (_, right)| {
-                            distance
-                                .evaluate_similarity(start, left)
-                                .total_cmp(&distance.evaluate_similarity(start, right))
-                        })
+                        .map(|(index, row)| (index, distance.evaluate_similarity(start, row)))
+                        .min_by(|(_, left), (_, right)| left.total_cmp(right))
                         .map(|(index, _)| index)
                 })
                 .context("PiPNN cannot connect a start point to an empty dataset")

From 8556fa879dea1af49f32f1bd68c4cfcfac793c1b Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Fri, 4 Sep 2026 06:07:32 +0000
Subject: [PATCH 20/21] fix(benchmark): retain spherical summary arm

---
 diskann-benchmark/src/inputs/disk.rs | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/diskann-benchmark/src/inputs/disk.rs b/diskann-benchmark/src/inputs/disk.rs
index 0a57dbd442..2451586852 100644
--- a/diskann-benchmark/src/inputs/disk.rs
+++ b/diskann-benchmark/src/inputs/disk.rs
@@ -478,12 +478,12 @@ impl DiskIndexBuild {
                         write_field!(f, "Quantization", format!("sq, nbits {nbits}"))?
                     }
                 }
+                QuantizationType::Spherical(bits) => write_field!(
+                    f,
+                    "Quantization",
+                    format!("spherical, nbits {}", bits.as_usize())
+                )?,
             }
-            QuantizationType::Spherical(bits) => write_field!(
-                f,
-                "Quantization",
-                format!("spherical, nbits {}", bits.as_usize())
-            )?,
         }
         #[cfg(feature = "disk-index")]
         write_field!(f, "Build Algorithm", self.build_algorithm)?;

From c6d1670f499c7c72610fd759687213c54bfc5757 Mon Sep 17 00:00:00 2001
From: Weiyao Luo <9347182+SeliMeli@users.noreply.github.com>
Date: Mon, 14 Sep 2026 03:39:44 +0000
Subject: [PATCH 21/21] perf(pipnn): install graph rows in parallel

---
 diskann-benchmark/src/index/build.rs | 115 +++++++++++++++++++--------
 1 file changed, 80 insertions(+), 35 deletions(-)

diff --git a/diskann-benchmark/src/index/build.rs b/diskann-benchmark/src/index/build.rs
index f8b2de462b..ed11512b18 100644
--- a/diskann-benchmark/src/index/build.rs
+++ b/diskann-benchmark/src/index/build.rs
@@ -130,6 +130,7 @@ where
     T: diskann::graph::SampleableForStart + diskann::utils::VectorRepr,
 {
     use anyhow::Context;
+    use rayon::prelude::*;
 
     let npoints = data.nrows();
     let dimensions = data.ncols();
@@ -192,16 +193,32 @@ where
     let install_started = std::time::Instant::now();
     // Install vectors before edges. A returned index must contain a vector for
     // every graph ID.
-    for (id, vector) in data.row_iter().enumerate() {
-        let id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
-        index.data_provider.base_vectors.set_element(&id, vector)?;
-    }
-    for (id, neighbors) in adjacency.into_iter().enumerate() {
-        index
-            .provider()
-            .neighbors()
-            .set_neighbors_sync(id, &neighbors)?;
-    }
+    #[expect(
+        clippy::disallowed_methods,
+        reason = "the supplied pool owns both terminal operations"
+    )]
+    pool.install(|| -> anyhow::Result<()> {
+        (0..npoints)
+            .into_par_iter()
+            .try_for_each(|id| -> anyhow::Result<()> {
+                let point_id = u32::try_from(id).context("PiPNN point ID exceeds u32::MAX")?;
+                index
+                    .data_provider
+                    .base_vectors
+                    .set_element(&point_id, data.row(id))?;
+                Ok(())
+            })?;
+        adjacency
+            .into_par_iter()
+            .enumerate()
+            .try_for_each(|(id, neighbors)| {
+                index
+                    .provider()
+                    .neighbors()
+                    .set_neighbors_sync(id, &neighbors)
+            })?;
+        Ok(())
+    })?;
     index.provider().set_start_points(start_points.row_iter())?;
     let start_ids = index.provider().starting_points()?;
     anyhow::ensure!(
@@ -422,27 +439,36 @@ mod pipnn_tests {
     use super::*;
     use diskann::graph::AdjacencyList;
 
-    #[test]
-    fn dedicated_pipeline_respects_the_requested_start_strategy() {
+    fn assert_vectors_edges_and_start_points_are_installed(threads: usize) {
+        // Given: coordinates i² have increasing gaps, so k=1 selects the previous
+        // point for every i>0; point 0 selects 1. Reverse edges form a chain.
+        // Seventeen rows cross the provider's sixteen-ID write-lock group.
+        let coordinates: [f32; 17] = std::array::from_fn(|id| (id * id) as f32);
+        let expected_neighbors: [Vec<u32>; 17] = std::array::from_fn(|id| {
+            (0..coordinates.len())
+                .filter(|&other| id.abs_diff(other) == 1)
+                .map(|other| other as u32)
+                .collect()
+        });
         let input: IndexBuild = serde_json::from_value(serde_json::json!({
             "data_type": "float32",
             "data": "unused.fbin",
             "distance": "squared_l2",
-            "max_degree": 4,
+            "max_degree": 2,
             "l_build": 8,
             "start_point_strategy": "first_vector",
             "alpha": 1.2,
             "backedge_ratio": 1.0,
-            "num_threads": 2,
+            "num_threads": threads,
             "multi_insert": null,
             "save_path": null,
             "build_algorithm": {
                 "algorithm": "PiPNN",
-                "c_max": 8,
-                "c_min": 2,
+                "c_max": 17,
+                "c_min": 1,
                 "p_samp": 0.5,
                 "fanout": [2],
-                "k": 2,
+                "k": 1,
                 "replicas": 1
             }
         }))
@@ -450,38 +476,57 @@ mod pipnn_tests {
         let diskann_disk::BuildAlgorithm::PiPNN(parameters) = input.build_algorithm() else {
             panic!("expected PiPNN parameters");
         };
-        let mut data = Matrix::new(0.0_f32, 16, 2);
-        for (index, row) in data.row_iter_mut().enumerate() {
-            row.copy_from_slice(&[index as f32, (index % 3) as f32]);
-        }
+        let mut data = Matrix::new(0.0_f32, coordinates.len(), 1);
+        data.as_mut_slice().copy_from_slice(&coordinates);
 
+        // When
         let (index, stats) = pipnn_build(Arc::new(data), &input, parameters).unwrap();
-        assert_eq!(stats.vectors_inserted, 16);
+
+        // Then: every real row contains its own vector and adjacency.
+        assert_eq!(stats.vectors_inserted, coordinates.len());
         assert!(stats.insert_latencies.is_none());
+        for (id, (&coordinate, expected)) in coordinates.iter().zip(&expected_neighbors).enumerate()
+        {
+            // SAFETY: id is a real dataset row; installation finished before return.
+            let vector = unsafe { index.data_provider.base_vectors.get_vector_sync(id) };
+            assert_eq!(vector, &[coordinate], "vector id={id}, threads={threads}");
+            let mut neighbors = AdjacencyList::new();
+            index
+                .provider()
+                .neighbors()
+                .get_neighbors_sync(id, &mut neighbors)
+                .unwrap();
+            assert_eq!(
+                &*neighbors, expected,
+                "adjacency id={id}, threads={threads}"
+            );
+        }
         let starts = index.provider().starting_points().unwrap();
         assert_eq!(starts.len(), 1);
-        // SAFETY: `starting_points` returns installed frozen IDs. Therefore,
-        // `starts[0] < base_vectors.total()`. The completed build has no vector
-        // writer, so the returned shared slice has no mutable alias.
+        // SAFETY: starting_points returns installed frozen IDs and no writer remains.
         let start = unsafe {
             index
                 .data_provider
                 .base_vectors
                 .get_vector_sync(starts[0] as usize)
         };
-        assert_eq!(start, [0.0, 0.0]);
-        let mut neighbors = AdjacencyList::new();
-        index
-            .provider()
-            .neighbors()
-            .get_neighbors_sync(starts[0] as usize, &mut neighbors)
-            .unwrap();
-        let mut source_neighbors = AdjacencyList::new();
+        assert_eq!(start, &[coordinates[0]]);
+        let mut start_neighbors = AdjacencyList::new();
         index
             .provider()
             .neighbors()
-            .get_neighbors_sync(0, &mut source_neighbors)
+            .get_neighbors_sync(starts[0] as usize, &mut start_neighbors)
             .unwrap();
-        assert_eq!(neighbors, source_neighbors);
+        assert_eq!(&*start_neighbors, &expected_neighbors[0]);
+    }
+
+    #[test]
+    fn one_thread_installs_vectors_edges_and_requested_start_point() {
+        assert_vectors_edges_and_start_points_are_installed(1);
+    }
+
+    #[test]
+    fn four_threads_install_vectors_edges_and_requested_start_point() {
+        assert_vectors_edges_and_start_points_are_installed(4);
     }
 }