feat: allow training results to create multiple writers

c/test_vindex.c

@@ -536,6 +536,40 @@ static void test_supported_index_roundtrips(void) {
         4);
 }
 
+static void test_training_opens_multiple_writers_before_consumption(void) {
+    const char *keys[] = {"index.type", "dimension", "nlist", "metric"};
+    const char *values[] = {"ivf_flat", "1", "1", "l2"};
+    const float data[] = {0.0f, 1.0f};
+
+    PaimonVindexTrainerHandle *trainer = paimon_vindex_trainer_open(keys, values, 4);
+    ASSERT_TRUE(trainer != NULL);
+    ASSERT_EQ_I64(
+        paimon_vindex_trainer_add_training_vectors(trainer, data, 2),
+        0);
+    PaimonVindexTrainingHandle *training = paimon_vindex_trainer_finish(trainer);
+    ASSERT_TRUE(training != NULL);
+    paimon_vindex_trainer_free(trainer);
+
+    PaimonVindexWriterHandle *first =
+        paimon_vindex_writer_open_from_training(training);
+    PaimonVindexWriterHandle *second =
+        paimon_vindex_writer_open_from_training(training);
+    PaimonVindexWriterHandle *consuming = paimon_vindex_writer_open(training);
+    ASSERT_TRUE(first != NULL);
+    ASSERT_TRUE(second != NULL);
+    ASSERT_TRUE(consuming != NULL);
+
+    uintptr_t dimension = 0;
+    ASSERT_EQ_I64(paimon_vindex_writer_dimension(first, &dimension), 0);
+    ASSERT_EQ_I64(dimension, 1);
+
+    paimon_vindex_writer_free(first);
+    paimon_vindex_writer_free(second);
+    paimon_vindex_writer_free(consuming);
+    paimon_vindex_training_free(training);
+    printf("PASS training_opens_multiple_writers_before_consumption\n");
+}
+
 static void test_extensible_search_params_defaults(void) {
     PaimonVindexSearchParamsEx params = paimon_vindex_search_params_ex_default();
 
@@ -554,6 +588,7 @@ static void test_extensible_search_params_defaults(void) {
 int main(void) {
     test_extensible_search_params_defaults();
     test_supported_index_roundtrips();
+    test_training_opens_multiple_writers_before_consumption();
     test_output_write_callback_error_propagates();
     test_output_flush_callback_error_propagates();
     test_input_read_ranges_callback_error_propagates();

core/src/diskann.rs

@@ -281,6 +281,27 @@ impl DiskAnnIndex {
         }
     }
 
+    /// Creates an empty index that reuses the trained product quantizer.
+    pub(crate) fn from_trained(trained: &DiskAnnIndex) -> Self {
+        Self {
+            d: trained.d,
+            metric: trained.metric,
+            pq: ProductQuantizer {
+                d: trained.pq.d,
+                m: trained.pq.m,
+                nbits: trained.pq.nbits,
+                dsub: trained.pq.dsub,
+                ksub: trained.pq.ksub,
+                chunk_offsets: trained.pq.chunk_offsets.clone(),
+                centroids: trained.pq.centroids.clone(),
+                centroid_norms_cache: trained.pq.centroid_norms_cache.clone(),
+            },
+            build_params: trained.build_params,
+            ids: Vec::new(),
+            vectors: Vec::new(),
+        }
+    }
+
     pub fn train(&mut self, data: &[f32], n: usize) -> io::Result<()> {
         if n == 0 {
             return Err(invalid_input(

core/src/index.rs

@@ -1294,6 +1294,31 @@ impl VectorIndexTraining {
     pub fn dimension(&self) -> usize {
         self.inner.dimension()
     }
+
+    /// Creates an empty writer that reuses this training result.
+    ///
+    /// The training result remains available, so callers can create independent writers for
+    /// multiple segments without repeating the training step. Use [`VectorIndexWriter::new`] when
+    /// the training result is only needed once and may be consumed.
+    pub fn create_writer(&self) -> VectorIndexWriter {
+        match &self.inner {
+            VectorIndexWriter::IvfFlat(index) => {
+                VectorIndexWriter::IvfFlat(IVFFlatIndex::from_trained(index))
+            }
+            VectorIndexWriter::IvfSq(index) => {
+                VectorIndexWriter::IvfSq(IVFSQIndex::from_trained(index))
+            }
+            VectorIndexWriter::IvfPq(index) => {
+                VectorIndexWriter::IvfPq(IVFPQIndex::from_trained(index))
+            }
+            VectorIndexWriter::IvfRq(index) => {
+                VectorIndexWriter::IvfRq(IVFRQIndex::from_trained(index))
+            }
+            VectorIndexWriter::DiskAnn(index) => {
+                VectorIndexWriter::DiskAnn(DiskAnnIndex::from_trained(index))
+            }
+        }
+    }
 }
 
 pub enum VectorIndexWriter {
@@ -2632,6 +2657,60 @@ mod tests {
         }
     }
 
+    fn assert_reusable_training_creates_independent_writers(config: VectorIndexConfig) {
+        let dimension = config.dimension();
+        let nlist = config.nlist();
+        let training_count = 256;
+        let segment_count = 48;
+        let data = generate_clustered_data(training_count, dimension, nlist);
+        let training = VectorIndexTrainer::train(config, &data, training_count).unwrap();
+
+        let mut first = training.create_writer();
+        let first_ids = (1_000..1_000 + segment_count as i64).collect::<Vec<_>>();
+        first
+            .add_vectors(
+                &first_ids,
+                &data[..segment_count * dimension],
+                segment_count,
+            )
+            .unwrap();
+
+        let mut second = training.create_writer();
+        let second_ids = (2_000..2_000 + segment_count as i64).collect::<Vec<_>>();
+        second
+            .add_vectors(
+                &second_ids,
+                &data[segment_count * dimension..2 * segment_count * dimension],
+                segment_count,
+            )
+            .unwrap();
+
+        for (mut writer, expected_ids, query) in [
+            (
+                first,
+                1_000..1_000 + segment_count as i64,
+                &data[..dimension],
+            ),
+            (
+                second,
+                2_000..2_000 + segment_count as i64,
+                &data[segment_count * dimension..(segment_count + 1) * dimension],
+            ),
+        ] {
+            let mut bytes = Vec::new();
+            writer.write(&mut PosWriter::new(&mut bytes)).unwrap();
+            let mut reader = VectorIndexReader::open(Cursor::new(bytes)).unwrap();
+            assert_eq!(reader.metadata().total_vectors, segment_count as i64);
+            let params = if reader.metadata().index_type == IndexType::DiskAnn {
+                VectorSearchParams::with_l_search(5, 32)
+            } else {
+                VectorSearchParams::new(5, nlist)
+            };
+            let (ids, _) = reader.search(query, params).unwrap();
+            assert!(ids.into_iter().all(|id| expected_ids.contains(&id)));
+        }
+    }
+
     fn build_reader(config: VectorIndexConfig) -> (VectorIndexReader<Cursor<Vec<u8>>>, Vec<f32>) {
         let d = config.dimension();
         let nlist = config.nlist();
@@ -3037,6 +3116,43 @@ mod tests {
         );
     }
 
+    #[test]
+    fn reusable_training_creates_independent_writers_for_all_index_types() {
+        assert_reusable_training_creates_independent_writers(VectorIndexConfig::IvfFlat {
+            dimension: 8,
+            nlist: 4,
+            metric: MetricType::L2,
+            use_approximate_coarse_assignment: true,
+        });
+        assert_reusable_training_creates_independent_writers(
+            VectorIndexConfig::ivf_pq(16, 4, MetricType::L2, false).unwrap(),
+        );
+        assert_reusable_training_creates_independent_writers(VectorIndexConfig::IvfRq {
+            dimension: 8,
+            nlist: 4,
+            bits: DEFAULT_RQ_BITS,
+            metric: MetricType::L2,
+            use_approximate_coarse_assignment: true,
+        });
+        assert_reusable_training_creates_independent_writers(VectorIndexConfig::IvfSq {
+            dimension: 8,
+            nlist: 4,
+            metric: MetricType::L2,
+            use_approximate_coarse_assignment: true,
+        });
+        assert_reusable_training_creates_independent_writers(VectorIndexConfig::DiskAnn {
+            dimension: 8,
+            metric: MetricType::L2,
+            pq_m: 4,
+            pq_bits: 4,
+            build: DiskAnnBuildParams {
+                max_degree: 8,
+                build_search_list_size: 16,
+                ..DiskAnnBuildParams::default()
+            },
+        });
+    }
+
     #[test]
     fn unified_config_rejects_invalid_pq_m() {
         let err = match VectorIndexTrainer::new(VectorIndexConfig::IvfPq {

core/src/ivfflat.rs

@@ -44,6 +44,21 @@ impl IVFFlatIndex {
         }
     }
 
+    /// Creates an empty index that reuses the trained coarse quantizer.
+    pub(crate) fn from_trained(trained: &IVFFlatIndex) -> Self {
+        let mut index = Self {
+            d: trained.d,
+            nlist: trained.nlist,
+            metric: trained.metric,
+            quantizer_centroids: trained.quantizer_centroids.clone(),
+            ids: vec![Vec::new(); trained.nlist],
+            vectors: vec![Vec::new(); trained.nlist],
+            coarse_assignment: CoarseAssignment::default(),
+        };
+        index.set_approximate_coarse_assignment(trained.coarse_assignment.approximate_enabled());
+        index
+    }
+
     pub fn quantizer_centroids(&self) -> &[f32] {
         &self.quantizer_centroids
     }

core/src/ivfrq.rs

@@ -121,6 +121,30 @@ impl IVFRQIndex {
         }
     }
 
+    /// Creates an empty index that reuses the trained coarse quantizer and rotation.
+    pub(crate) fn from_trained(trained: &IVFRQIndex) -> Self {
+        let mut index = Self {
+            d: trained.d,
+            padded_d: trained.padded_d,
+            nlist: trained.nlist,
+            bits: trained.bits,
+            metric: trained.metric,
+            quantizer_centroids: trained.quantizer_centroids.clone(),
+            quantizer_centroid_norms: trained.quantizer_centroid_norms.clone(),
+            rotated_centroids: trained.rotated_centroids.clone(),
+            rotation_seed: trained.rotation_seed,
+            rotation_rounds: trained.rotation_rounds,
+            ids: vec![Vec::new(); trained.nlist],
+            codes: vec![Vec::new(); trained.nlist],
+            factors: vec![Vec::new(); trained.nlist],
+            quantizer: trained.quantizer.clone(),
+            rotation: trained.rotation.clone(),
+            coarse_assignment: CoarseAssignment::default(),
+        };
+        index.set_approximate_coarse_assignment(trained.coarse_assignment.approximate_enabled());
+        index
+    }
+
     pub fn quantizer_centroids(&self) -> &[f32] {
         &self.quantizer_centroids
     }

core/src/ivfsq.rs

@@ -53,6 +53,23 @@ impl IVFSQIndex {
         }
     }
 
+    /// Creates an empty index that reuses the trained coarse and scalar quantizers.
+    pub(crate) fn from_trained(trained: &IVFSQIndex) -> Self {
+        let mut index = Self {
+            d: trained.d,
+            nlist: trained.nlist,
+            metric: trained.metric,
+            quantizer_centroids: trained.quantizer_centroids.clone(),
+            sq: trained.sq.clone(),
+            list_sqs: trained.list_sqs.clone(),
+            ids: vec![Vec::new(); trained.nlist],
+            codes: vec![Vec::new(); trained.nlist],
+            coarse_assignment: CoarseAssignment::default(),
+        };
+        index.set_approximate_coarse_assignment(trained.coarse_assignment.approximate_enabled());
+        index
+    }
+
     pub fn quantizer_centroids(&self) -> &[f32] {
         &self.quantizer_centroids
     }

cpp/test_vindex.cpp

@@ -302,6 +302,23 @@ static void test_supported_index_roundtrips() {
         4);
 }
 
+static void test_training_opens_multiple_writers_before_consumption() {
+    const float data[] = {0.0f, 1.0f};
+    paimon::vindex::Trainer trainer({
+        {"index.type", "ivf_flat"},
+        {"dimension", "1"},
+        {"nlist", "1"},
+        {"metric", "l2"},
+    });
+    auto training = trainer.add_training_vectors(data, 2).finish_training();
+
+    paimon::vindex::Writer first(training);
+    paimon::vindex::Writer second(training);
+    paimon::vindex::Writer consuming(std::move(training));
+    ASSERT_EQ(first.dimension(), 1);
+    printf("PASS training_opens_multiple_writers_before_consumption\n");
+}
+
 static void test_worker_callback_reentry_is_rejected() {
     int callback_context = 0;
     paimon::vindex::detail::NativeHandleMutex mutex;
@@ -341,6 +358,7 @@ static void test_extensible_search_params_forward_query_tuning() {
 
 int main() {
     test_supported_index_roundtrips();
+    test_training_opens_multiple_writers_before_consumption();
     test_worker_callback_reentry_is_rejected();
     test_extensible_search_params_forward_query_tuning();
     return 0;

docs/api.html

@@ -47,8 +47,16 @@ <h2>Public integration layers</h2>
 
         <section class="article-section" id="lifecycle">
           <h2>Shared lifecycle</h2>
-          <div class="flow" aria-label="Unified API lifecycle"><div class="flow-step"><small>01</small><strong>Create a Trainer<br>Parse and validate options</strong></div><div class="flow-step"><small>02</small><strong>Submit one or more<br>training batches</strong></div><div class="flow-step"><small>03</small><strong>Finish training and<br>create a one-shot Writer</strong></div><div class="flow-step"><small>04</small><strong>Add row IDs / vectors<br>and write the file</strong></div><div class="flow-step"><small>05</small><strong>Detect file magic<br>and execute searches</strong></div></div>
-          <ul><li>Vectors are contiguous <code>f32</code> values; length must equal <code>vector_count × dimension</code>.</li><li>Training data may arrive in batches. Every IVF trainer keeps a deterministic reservoir of at most <code>max(65,536, 64 × resolved nlist)</code> vectors. DiskANN starts from a 50,000-row cap and lowers it when necessary so the retained sample, optional cosine-normalized copy, codebook, and parallel PQ-training scratch fit <code>diskann.memory-budget-bytes</code>. Sampling is independent of batch boundaries.</li><li>The Python and Java one-shot <code>train</code> helpers infer <code>dimension</code> from the matrix and use its row count for automatic <code>nlist</code>. When the matrix is only a sample, pass the final corpus size as <code>expected-vector-count</code>. Streaming Trainer APIs require a concrete dimension before their first batch.</li><li>A Writer may receive production vectors in multiple batches. Row-ID count must equal vector count.</li><li>Readers expose metadata, single-query search, batch search, and Roaring64-filtered variants.</li><li>Files carry their type and resolved model sections. Callers do not pass index options again when opening a Reader.</li></ul>
+          <div class="flow" aria-label="Unified API lifecycle"><div class="flow-step"><small>01</small><strong>Create a Trainer<br>Parse and validate options</strong></div><div class="flow-step"><small>02</small><strong>Submit one or more<br>training batches</strong></div><div class="flow-step"><small>03</small><strong>Finish training and<br>create one or more Writers</strong></div><div class="flow-step"><small>04</small><strong>Add row IDs / vectors<br>and write each file</strong></div><div class="flow-step"><small>05</small><strong>Detect file magic<br>and execute searches</strong></div></div>
+          <ul><li>Vectors are contiguous <code>f32</code> values; length must equal <code>vector_count × dimension</code>.</li><li>Training data may arrive in batches. Every IVF trainer keeps a deterministic reservoir of at most <code>max(65,536, 64 × resolved nlist)</code> vectors. DiskANN starts from a 50,000-row cap and lowers it when necessary so the retained sample, optional cosine-normalized copy, codebook, and parallel PQ-training scratch fit <code>diskann.memory-budget-bytes</code>. Sampling is independent of batch boundaries.</li><li>The Python and Java one-shot <code>train</code> helpers infer <code>dimension</code> from the matrix and use its row count for automatic <code>nlist</code>. When the matrix is only a sample, pass the final corpus size as <code>expected-vector-count</code>. Streaming Trainer APIs require a concrete dimension before their first batch.</li><li>A completed training result can create multiple independent Writers. Each Writer starts with the same trained centroids, quantizers, and rotations but empty vector payloads, so separate segments can share one training pass.</li><li>A Writer may receive production vectors in multiple batches. Row-ID count must equal vector count.</li><li>Readers expose metadata, single-query search, batch search, and Roaring64-filtered variants.</li><li>Files carry their type and resolved model sections. Callers do not pass index options again when opening a Reader.</li></ul>
+          <div class="table-wrap"><table><thead><tr><th>Language</th><th>Reusable training</th><th>Existing consuming form</th></tr></thead><tbody>
+            <tr><td>Rust</td><td><code>training.create_writer()</code></td><td><code>VectorIndexWriter::new(training)</code></td></tr>
+            <tr><td>C</td><td><code>paimon_vindex_writer_open_from_training(training)</code></td><td><code>paimon_vindex_writer_open(training)</code></td></tr>
+            <tr><td>C++</td><td><code>Writer writer(training)</code></td><td><code>Writer writer(std::move(training))</code></td></tr>
+            <tr><td>Java</td><td><code>training.createWriter()</code></td><td><code>new VectorIndexWriter(training)</code></td></tr>
+            <tr><td>Python</td><td><code>training.create_writer()</code></td><td><code>VectorIndexWriter(training)</code></td></tr>
+          </tbody></table></div>
+          <p>Reusable Writer creation is opt-in; the existing consuming forms and their behavior remain unchanged. Do not consume or close a training result until every reusable Writer has been created.</p>
           <div class="callout warning"><strong>IVF coarse assignment is approximate by default for large centroid matrices</strong>When <code>dimension × nlist ≥ 1,000,000</code>, <code>ivf.coarse-assignment=auto</code> uses a Vamana graph while training and adding vectors. Search still selects lists by exact centroid distance, so graph assignment can lower recall at small <code>nprobe</code> and does not guarantee that a vector is found by a self-query with <code>nprobe=1</code>. Set <code>ivf.coarse-assignment=exact</code> to disable the graph, preserve exact nearest-centroid assignment, and avoid graph startup cost for small non-empty batches. Empty batches never build the graph.</div>
         </section>
 

ffi/src/lib.rs

@@ -463,6 +463,16 @@ unsafe fn training_mut<'a>(
     }
 }
 
+unsafe fn training_ref<'a>(
+    handle: *const PaimonVindexTrainingHandle,
+) -> Result<&'a PaimonVindexTrainingHandle, String> {
+    if handle.is_null() {
+        Err("null training handle".to_string())
+    } else {
+        Ok(unsafe { &*handle })
+    }
+}
+
 unsafe fn reader_mut<'a>(
     handle: *mut PaimonVindexReaderHandle,
 ) -> Result<&'a mut PaimonVindexReaderHandle, String> {
@@ -828,6 +838,26 @@ pub unsafe extern "C" fn paimon_vindex_writer_open(
     })
 }
 
+/// Opens an independent writer without consuming `training`.
+///
+/// The same training handle may be used to open additional writers until it is consumed by
+/// `paimon_vindex_writer_open` or freed by `paimon_vindex_training_free`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vindex_writer_open_from_training(
+    training: *const PaimonVindexTrainingHandle,
+) -> *mut PaimonVindexWriterHandle {
+    ffi_ptr(|| {
+        let training = unsafe { training_ref(training) }?;
+        let training = training
+            .inner
+            .as_ref()
+            .ok_or_else(|| "training has already been consumed".to_string())?;
+        Ok(Box::into_raw(Box::new(PaimonVindexWriterHandle {
+            inner: training.create_writer(),
+        })))
+    })
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn paimon_vindex_writer_free(handle: *mut PaimonVindexWriterHandle) {
     if !handle.is_null() {