diff --git a/python/python/tests/test_create_empty_index.py b/python/python/tests/test_create_empty_index.py
index 77d4ab034c9..39a2580b2ff 100644
--- a/python/python/tests/test_create_empty_index.py
+++ b/python/python/tests/test_create_empty_index.py
@@ -31,17 +31,31 @@ def test_create_empty_vector_index():
     data = pa.table({"vector": vectors})
     dataset = lance.write_dataset(data, "memory://")
 
-    # Currently, vector indices with train=False are not supported
-    try:
-        dataset.create_index(
-            "vector", "IVF_PQ", num_partitions=10, num_sub_vectors=8, train=False
-        )
-        # If we get here, the implementation has been added (unexpected for now)
-        assert False, (
-            "Expected NotImplementedError for train=False on vector index, "
-            "but succeeded"
-        )
-    except NotImplementedError as e:
-        # Expected error for unimplemented functionality
-        error_msg = str(e).lower()
-        assert "not yet implemented" in error_msg or "not implemented" in error_msg
+    dataset.create_index(
+        "vector", "IVF_PQ", num_partitions=10, num_sub_vectors=8, train=False
+    )
+
+    # Same shape the scalar case reports: listed, covering nothing yet.
+    indices = dataset.describe_indices()
+    assert len(indices) == 1
+    stats = dataset.stats.index_stats(indices[0].name)
+    assert stats["num_indexed_rows"] == 0
+    assert stats["num_unindexed_rows"] == dataset.count_rows()
+
+
+def test_create_vector_index_below_the_row_floor():
+    """A table too small to train the quantizer takes the index anyway."""
+    dim = 32
+    values = pc.random(100 * dim).cast(pa.float32())
+    vectors = pa.FixedSizeListArray.from_arrays(values, dim)
+    data = pa.table({"vector": vectors})
+    dataset = lance.write_dataset(data, "memory://")
+
+    # 100 vectors cannot train a 256-code codebook, and train defaults to True.
+    dataset.create_index("vector", "IVF_PQ", num_partitions=10, num_sub_vectors=8)
+
+    indices = dataset.describe_indices()
+    assert len(indices) == 1
+    stats = dataset.stats.index_stats(indices[0].name)
+    assert stats["num_indexed_rows"] == 0
+    assert stats["num_unindexed_rows"] == dataset.count_rows()
diff --git a/python/python/tests/test_ivf_centroids.py b/python/python/tests/test_ivf_centroids.py
index 01c2463ab62..34827c330ce 100644
--- a/python/python/tests/test_ivf_centroids.py
+++ b/python/python/tests/test_ivf_centroids.py
@@ -8,7 +8,10 @@
 def test_ivf_centroids_exposed(tmp_path):
     """Verify that centroids for an IVF-based index are exposed via both"""
 
-    dim, rows, parts = 4, 256, 8
+    # A partition is trained only when the data can give it a 256-code
+    # codebook's worth of vectors, so the fixture covers every partition.
+    dim, parts = 4, 8
+    rows = parts * 256
     vecs = pa.array(
         np.random.randn(rows, dim).tolist(),
         type=pa.list_(pa.float32(), dim),
diff --git a/python/python/tests/test_vector_index.py b/python/python/tests/test_vector_index.py
index eb2c036f76b..1dce77ee518 100644
--- a/python/python/tests/test_vector_index.py
+++ b/python/python/tests/test_vector_index.py
@@ -1238,18 +1238,19 @@ def test_create_ivf_rq_index():
     assert stats["indices"][0]["sub_index"]["num_bits"] == 5
     assert stats["indices"][0]["sub_index"]["packed"] is True
 
-    with pytest.raises(
-        NotImplementedError,
-        match="Creating empty vector indices with train=False is not yet implemented",
-    ):
-        ds.delete("id>=0")
-        ds = ds.create_index(
-            "vector",
-            index_type="IVF_RQ",
-            num_partitions=4,
-            num_bits=1,
-            replace=True,
-        )
+    # An emptied table still takes the index; it carries its settings and
+    # covers nothing until there is data to train on.
+    ds.delete("id>=0")
+    ds = ds.create_index(
+        "vector",
+        index_type="IVF_RQ",
+        num_partitions=4,
+        num_bits=1,
+        replace=True,
+    )
+    stats = ds.stats.index_stats("vector_idx")
+    assert stats["num_indexed_rows"] == 0
+    assert stats["num_unindexed_rows"] == 0
 
     zero_vectors = np.zeros((1000, 128)).astype(np.float32).tolist()
     tbl = pa.Table.from_pydict(
@@ -1863,7 +1864,9 @@ def query_index(ds, ntimes, q=None):
                 },
             )
 
-    tbl = create_table(nvec=1024, ndim=16)
+    # Each of the 128 partitions is trained only when the data can give it a
+    # 256-code codebook's worth of vectors.
+    tbl = create_table(nvec=128 * 256, ndim=16)
     dataset = lance.write_dataset(tbl, tmp_path / "test")
 
     dataset.create_index(
@@ -1908,7 +1911,9 @@ def query_index(ds, ntimes, q=None):
                 },
             )
 
-    tbl = create_table(nvec=1024, ndim=16)
+    # Each of the 128 partitions is trained only when the data can give it a
+    # 256-code codebook's worth of vectors.
+    tbl = create_table(nvec=128 * 256, ndim=16)
     dataset = lance.write_dataset(tbl, tmp_path / "test")
 
     dataset.create_index(
diff --git a/rust/lance/src/dataset/mem_wal/api.rs b/rust/lance/src/dataset/mem_wal/api.rs
index ef55d20d5ab..1de4d298bb5 100644
--- a/rust/lance/src/dataset/mem_wal/api.rs
+++ b/rust/lance/src/dataset/mem_wal/api.rs
@@ -823,20 +823,31 @@ async fn load_vector_index_config(
     let column = field.name.clone();
 
     // Inherit the base table's distance type so the in-memory index and the
-    // base index produce comparable distances. Surface the open error
-    // instead of silently defaulting to L2 — flushed `IVF_HNSW_SQ` files
-    // bake this metric into their on-disk metadata, so a wrong default would
-    // be durable corruption.
-    let distance_type = dataset
-        .open_vector_index(&column, &index_meta.uuid, &NoOpMetricsCollector)
-        .await
-        .map_err(|e| {
-            Error::invalid_input(format!(
-                "Failed to open base vector index '{}' to inherit distance type: {}",
-                index_name, e
-            ))
-        })?
-        .metric_type();
+    // base index produce comparable distances. The index's recorded details
+    // state it, and for an index that covers nothing they are the only source:
+    // it carries its settings with no file to open. Opening the index is the
+    // fallback for an entry whose details do not decode. Surface the failure
+    // rather than silently defaulting to L2 — flushed `IVF_HNSW_SQ` files bake this metric
+    // into their on-disk metadata, so a wrong default would be durable
+    // corruption.
+    let recorded = index_meta
+        .index_details
+        .as_deref()
+        .and_then(crate::index::vector::details::vector_params_from_details)
+        .map(|params| params.metric_type);
+    let distance_type = match recorded {
+        Some(distance_type) => distance_type,
+        None => dataset
+            .open_vector_index(&column, &index_meta.uuid, &NoOpMetricsCollector)
+            .await
+            .map_err(|e| {
+                Error::invalid_input(format!(
+                    "Failed to open base vector index '{}' to inherit distance type: {}",
+                    index_name, e
+                ))
+            })?
+            .metric_type(),
+    };
 
     Ok(match hnsw_params {
         Some(params) => MemIndexConfig::hnsw_with_params(
@@ -1050,6 +1061,53 @@ mod tests {
             .expect("a Float32 vector column is maintainable");
     }
 
+    /// An index that covers nothing is maintainable: its recorded details
+    /// state the distance type, so there is no file to open.
+    ///
+    /// A table can register WAL before it holds enough vectors to train, and
+    /// validation refusing the definition would leave the index outside the
+    /// maintained set for the life of the table — the set is a snapshot, so
+    /// training it later does not add it back.
+    #[tokio::test]
+    async fn test_validate_maintained_indexes_accepts_a_definition() {
+        use crate::index::vector::VectorIndexParams;
+        use lance_linalg::distance::DistanceType;
+
+        let tmp = tempfile::tempdir().unwrap();
+        let uri = format!("{}/base", tmp.path().to_str().unwrap());
+        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "vector",
+            DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 4),
+            true,
+        )]));
+        let reader = RecordBatchIterator::new(vec![], schema.clone());
+        let mut dataset = Dataset::write(reader, &uri, Some(WriteParams::default()))
+            .await
+            .unwrap();
+        dataset
+            .create_index(
+                &["vector"],
+                IndexType::Vector,
+                Some("vector_idx".to_string()),
+                &VectorIndexParams::ivf_pq(4, 8, 2, DistanceType::Cosine, 1),
+                true,
+            )
+            .await
+            .unwrap();
+        let indices = dataset.load_indices_by_name("vector_idx").await.unwrap();
+        assert!(
+            indices[0]
+                .fragment_bitmap
+                .as_ref()
+                .is_some_and(roaring::RoaringBitmap::is_empty),
+            "an empty table trains nothing, so the index covers no rows"
+        );
+
+        validate_maintained_indexes(&dataset, &["vector_idx".to_string()])
+            .await
+            .expect("a definition is maintainable");
+    }
+
     #[tokio::test]
     async fn test_validate_maintained_indexes_accepts_btree() {
         // Guards the shard-schema plumbing: validation resolves field ids against
diff --git a/rust/lance/src/dataset/scanner.rs b/rust/lance/src/dataset/scanner.rs
index 8733b167263..99f0aea31f9 100644
--- a/rust/lance/src/dataset/scanner.rs
+++ b/rust/lance/src/dataset/scanner.rs
@@ -7280,7 +7280,7 @@ pub mod test_dataset {
     use uuid::Uuid;
 
     use crate::dataset::WriteParams;
-    use crate::index::vector::VectorIndexParams;
+    use crate::index::vector::{StageParams, VectorIndexParams};
 
     // Creates a dataset with 5 batches where each batch has 80 rows
     //
@@ -7379,7 +7379,13 @@ pub mod test_dataset {
         }
 
         pub async fn make_vector_index_with_metric(&mut self, metric: MetricType) -> Result<()> {
-            let params = VectorIndexParams::ivf_pq(2, 8, 2, metric, 2);
+            let mut params = VectorIndexParams::ivf_pq(2, 8, 2, metric, 2);
+            // Two partitions over 400 vectors only holds at a sample rate this
+            // fixture can cover: the trained count is capped at the vectors
+            // available per centroid.
+            if let Some(StageParams::Ivf(ivf)) = params.stages.first_mut() {
+                ivf.sample_rate = 200;
+            }
             self.dataset
                 .create_index(
                     &["vec"],
@@ -12040,10 +12046,16 @@ mod test {
         data_storage_version: LanceFileVersion,
         #[values(false, true)] stable_row_ids: bool,
     ) {
+        const PARTITIONS: usize = 4;
+        // A partition is trained only when the data can give it a codebook's
+        // worth of vectors, so the fixture has to cover every partition it asks
+        // for; the cached-entry count below counts per partition.
+        const ROWS: usize = PARTITIONS * 256;
+
         let vec_params = vec![
             // TODO: re-enable diskann test when we can tune to get reproducible results.
             // VectorIndexParams::with_diskann_params(MetricType::L2, DiskANNParams::new(10, 1.5, 10)),
-            VectorIndexParams::ivf_pq(4, 8, 2, MetricType::L2, 2),
+            VectorIndexParams::ivf_pq(PARTITIONS, 8, 2, MetricType::L2, 2),
         ];
         for params in vec_params {
             use lance_arrow::FixedSizeListArrayExt;
@@ -12066,14 +12078,14 @@ mod test {
 
             // vectors are [1, 1, 1, ...] [2, 2, 2, ...]
             let vector_values: Float32Array =
-                (0..32 * 512).map(|v| (v / 32) as f32 + 1.0).collect();
+                (0..32 * ROWS).map(|v| (v / 32) as f32 + 1.0).collect();
             let vectors = FixedSizeListArray::try_new_from_values(vector_values, 32).unwrap();
 
             let batches = vec![
                 RecordBatch::try_new(
                     schema.clone(),
                     vec![
-                        Arc::new(Int32Array::from_iter_values(0..512)),
+                        Arc::new(Int32Array::from_iter_values(0..ROWS as i32)),
                         Arc::new(vectors.clone()),
                     ],
                 )
@@ -12175,7 +12187,7 @@ mod test {
                 RecordBatch::try_new(
                     schema.clone(),
                     vec![
-                        Arc::new(Int32Array::from_iter_values(512..1024)),
+                        Arc::new(Int32Array::from_iter_values(ROWS as i32..2 * ROWS as i32)),
                         Arc::new(vectors),
                     ],
                 )
@@ -12205,7 +12217,7 @@ mod test {
                 .await
                 .unwrap();
 
-            dataset.delete("i < 512").await.unwrap();
+            dataset.delete(&format!("i < {ROWS}")).await.unwrap();
 
             let mut scan = dataset.scan();
             scan.nearest("vec", &key, 5).unwrap();
@@ -12224,7 +12236,9 @@ mod test {
             let batch = &results[0];
 
             // It should not pick up any results from the first fragment
-            let expected_i = BTreeSet::from_iter(vec![512, 513, 514, 515, 516]);
+            let first = ROWS as i32;
+            let expected_i =
+                BTreeSet::from_iter(vec![first, first + 1, first + 2, first + 3, first + 4]);
             let column_i = batch.column_by_name("i").unwrap();
             let actual_i: BTreeSet<i32> = as_primitive_array::<Int32Type>(column_i.as_ref())
                 .values()
diff --git a/rust/lance/src/dataset/tests/dataset_index.rs b/rust/lance/src/dataset/tests/dataset_index.rs
index fb59a2f9da8..d0953ffa9d9 100644
--- a/rust/lance/src/dataset/tests/dataset_index.rs
+++ b/rust/lance/src/dataset/tests/dataset_index.rs
@@ -86,7 +86,7 @@ async fn test_create_index(
         false,
     )]));
 
-    let float_arr = generate_random_array(512 * dimension as usize);
+    let float_arr = generate_random_array(2560 * dimension as usize);
     let vectors = Arc::new(
         <arrow_array::FixedSizeListArray as FixedSizeListArrayExt>::try_new_from_values(
             float_arr, dimension,
@@ -667,7 +667,7 @@ async fn test_create_int8_index(
         false,
     )]));
 
-    let int8_arr = generate_random_int8_array(512 * dimension as usize);
+    let int8_arr = generate_random_int8_array(2560 * dimension as usize);
     let vectors = Arc::new(
         <arrow_array::FixedSizeListArray as FixedSizeListArrayExt>::try_new_from_values(
             int8_arr, dimension,
diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs
index 415958a438a..0789f4877c1 100644
--- a/rust/lance/src/index.rs
+++ b/rust/lance/src/index.rs
@@ -838,7 +838,7 @@ fn validate_segment_index_details(index_name: &str, segments: &[IndexMetadata])
 ///
 /// Older vector segments may not have `VectorIndexDetails` in the manifest, so
 /// we also recognize them by the legacy monolithic index file name.
-fn segment_has_vector_details(segment: &IndexMetadata) -> bool {
+pub(crate) fn segment_has_vector_details(segment: &IndexMetadata) -> bool {
     segment.index_details.as_ref().map_or_else(
         || {
             segment
@@ -2649,6 +2649,17 @@ async fn collect_regular_indices_statistics(
     let mut index_uri: Option<String> = None;
 
     for meta in metadatas.iter() {
+        // An index that covers no fragments has no file to load statistics
+        // from: it carries its definition and nothing else until there is
+        // enough data to train it.
+        if meta
+            .fragment_bitmap
+            .as_ref()
+            .is_some_and(roaring::RoaringBitmap::is_empty)
+        {
+            indices_stats.push(serde_json::json!({}));
+            continue;
+        }
         let index_store = Arc::new(LanceIndexStore::from_dataset_for_existing(ds, meta).await?);
         let index_details = scalar::fetch_index_details(ds, field_path, meta).await?;
         if index_uri.is_none() {
@@ -2985,12 +2996,15 @@ impl DatasetIndexInternalExt for Dataset {
             .await?
             .ok_or_else(|| Error::index(format!("Index with id {} does not exist", uuid)))?;
 
-        // Check if this is a vector index by looking at the files list
-        let is_vector_index = if let Some(files) = &index_meta.files {
-            // If we have file metadata, check if INDEX_FILE_NAME is in the list
+        // Declared type first, and the legacy file name only for segments that
+        // predate details: an index awaiting training declares itself a vector
+        // index and has no file, so a file-based answer would send it to the
+        // scalar reader.
+        let is_vector_index = if index_meta.index_details.is_some() {
+            segment_has_vector_details(&index_meta)
+        } else if let Some(files) = &index_meta.files {
             files.iter().any(|f| f.path == INDEX_FILE_NAME)
         } else {
-            // Fall back to file existence check for older indices without file metadata
             let index_dir = self.indice_files_dir(&index_meta)?;
             let index_file = index_dir
                 .clone()
@@ -3799,6 +3813,7 @@ mod tests {
         hnsw::builder::HnswBuildParams,
         ivf::IvfBuildParams,
         kmeans::{KMeansParams, train_kmeans},
+        pq::builder::PQBuildParams,
         sq::builder::SQBuildParams,
     };
     use lance_io::{
@@ -6034,19 +6049,827 @@ mod tests {
         let mut dataset = Dataset::write(reader, test_uri, None).await.unwrap();
 
         let params = VectorIndexParams::ivf_pq(1, 8, 96, DistanceType::L2, 1);
-        let result = dataset
+        dataset
             .create_index(&["vector"], IndexType::Vector, None, &params, false)
-            .await;
+            .await
+            .expect("a table too small to train on still accepts an index");
+
+        // The definition is there and covers nothing: 100 rows cannot train a
+        // 256-code quantizer, so there is nothing to cover yet.
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(segment_covers_nothing(&indices[0]));
+        // Every fragment is still unindexed, which is what optimize will pick
+        // up once the column can train.
+        let unindexed = dataset.unindexed_fragments("vector_idx").await.unwrap();
+        assert_eq!(unindexed.len(), dataset.get_fragments().len());
+    }
+
+    /// More partitions requested than the data supports trains fewer of them.
+    ///
+    /// A partition wants a codebook's worth of vectors, so 300 vectors support
+    /// one partition however many are asked for.
+    #[tokio::test]
+    async fn test_create_index_with_more_partitions_than_rows() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let rows = 300;
+        let mut dataset = small_vector_dataset(test_dir.path(), rows).await;
+
+        let params = VectorIndexParams::ivf_pq(1000, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .expect("more partitions than rows should reduce partitions, not fail");
+
+        // Trained, not degraded: 300 rows clear the 256-code PQ floor.
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(!segment_covers_nothing(&indices[0]));
+
+        // 300 / 256 = 1, not the 1000 requested.
+        assert_eq!(trained_partitions(&dataset).await, vec![1]);
+    }
+
+    /// Partition counts for a trained logical vector index, one per segment.
+    /// Whether a segment covers no rows, which is how a definition reads back
+    /// from the manifest.
+    fn segment_covers_nothing(index: &IndexMetadata) -> bool {
+        index
+            .fragment_bitmap
+            .as_ref()
+            .is_some_and(roaring::RoaringBitmap::is_empty)
+    }
+
+    async fn trained_partitions(dataset: &Dataset) -> Vec<usize> {
+        dataset
+            .open_logical_vector_index("vector", "vector_idx")
+            .await
+            .unwrap()
+            .as_ivf()
+            .unwrap()
+            .num_partitions_per_segment()
+            .into_iter()
+            .map(|(_, partitions)| partitions)
+            .collect()
+    }
+
+    /// A table indexed while empty trains on the first append-mode optimize
+    /// that has data behind it.
+    ///
+    /// Writing rows never touches the index, so the optimize is where the
+    /// definition becomes a real index — and append is the mode scheduled
+    /// maintenance uses, so it has to be the mode that gets there.
+    #[tokio::test]
+    async fn test_append_mode_trains_a_definition_once_data_arrives() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 0).await;
+
+        let params = VectorIndexParams::ivf_pq(2, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        let mut dataset = append_vectors(test_dir.path(), 3000).await;
+        assert_eq!(dataset.count_rows(None).await.unwrap(), 3000);
+        let indices = dataset.load_indices().await.unwrap();
+        assert!(
+            segment_covers_nothing(&indices[0]),
+            "writing rows leaves the index a definition"
+        );
+
+        dataset
+            .optimize_indices(&OptimizeOptions::append())
+            .await
+            .unwrap();
+
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(
+            indices.len(),
+            1,
+            "training supersedes the definition rather than adding a delta to it"
+        );
+        assert!(
+            !segment_covers_nothing(&indices[0]),
+            "3,000 vectors clear the floor, so the append trains the column"
+        );
+        // Sized from the 3,000 vectors present, not the 2 the request named.
+        assert_eq!(trained_partitions(&dataset).await, vec![1]);
+    }
+
+    /// The cap reduces a requested count only while the data cannot support it.
+    #[tokio::test]
+    async fn test_create_index_partition_cap_follows_the_data() {
+        // 8 partitions want 8 * 256 = 2048 vectors: 1000 falls inside the band
+        // where the count is reduced, 3000 clears it.
+        for (rows, expected) in [(1000, 3), (3000, 8)] {
+            let test_dir = tempfile::tempdir().unwrap();
+            let mut dataset = small_vector_dataset(test_dir.path(), rows).await;
+
+            let params = VectorIndexParams::ivf_pq(8, 8, 4, DistanceType::L2, 1);
+            dataset
+                .create_index(&["vector"], IndexType::Vector, None, &params, false)
+                .await
+                .unwrap();
 
-        assert!(matches!(result, Err(Error::Unprocessable { .. })));
-        if let Error::Unprocessable { message, .. } = result.unwrap_err() {
             assert_eq!(
-                message,
-                "Not enough rows to train PQ. Requires 256 rows but only 100 available",
-            )
+                trained_partitions(&dataset).await,
+                vec![expected],
+                "{rows} vectors"
+            );
+        }
+    }
+
+    /// The cap counts the vectors IVF fits a centroid on, not codebook entries.
+    #[tokio::test]
+    async fn test_partition_cap_ignores_the_codebook_size() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 300).await;
+
+        // A 4-bit codebook holds 16 entries, but a partition still wants the
+        // 256 vectors IVF training samples for one centroid.
+        let params = VectorIndexParams::ivf_pq(8, 4, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // 300 / 256 = 1.
+        assert_eq!(trained_partitions(&dataset).await, vec![1]);
+    }
+
+    /// Sampling fewer vectors per centroid makes more partitions supportable.
+    #[tokio::test]
+    async fn test_partition_cap_follows_the_configured_sample_rate() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 300).await;
+
+        let params = VectorIndexParams::with_ivf_pq_params(
+            DistanceType::L2,
+            IvfBuildParams {
+                num_partitions: Some(8),
+                sample_rate: 64,
+                ..Default::default()
+            },
+            PQBuildParams {
+                num_sub_vectors: 4,
+                num_bits: 8,
+                ..Default::default()
+            },
+        );
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // 300 / 64 = 4.
+        assert_eq!(trained_partitions(&dataset).await, vec![4]);
+    }
+
+    /// A definition records no partition count, so the index the table grows
+    /// into is sized from the data present when it finally trains. A count
+    /// asked for on a table too small to train it does not survive the wait.
+    #[tokio::test]
+    async fn test_deferred_index_sizes_from_the_data_it_trains_on() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(8, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+        let indices = dataset.load_indices().await.unwrap();
+        assert!(
+            segment_covers_nothing(&indices[0]),
+            "100 vectors cannot train a 256-code quantizer, so nothing is covered yet"
+        );
+
+        // 3100 vectors would clear 8 * 256, but no count was recorded, so
+        // training derives one from the data: 3100 / 8192 -> 1.
+        let mut dataset = append_vectors(test_dir.path(), 3000).await;
+        dataset
+            .optimize_indices(&OptimizeOptions::default())
+            .await
+            .unwrap();
+
+        assert_eq!(trained_partitions(&dataset).await, vec![1]);
+    }
+
+    /// Appending to an index that is still a definition, on a table that still
+    /// cannot train, is a no-op rather than an error, and stays one when
+    /// repeated.
+    #[tokio::test]
+    async fn test_append_to_a_definition_that_still_cannot_train() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // Twice, so the second call sees whatever the first one left behind.
+        for attempt in 0..2 {
+            dataset
+                .optimize_indices(&OptimizeOptions::append())
+                .await
+                .unwrap_or_else(|error| {
+                    panic!("append {attempt} on a definition must not fail: {error}")
+                });
+            let indices = dataset.load_indices().await.unwrap();
+            assert_eq!(indices.len(), 1);
+            assert!(
+                segment_covers_nothing(&indices[0]),
+                "100 vectors still cannot train a 256-code quantizer"
+            );
+        }
+    }
+
+    /// Compaction leaves an index that covers nothing alone. It has no file to
+    /// remap, and an empty fragment bitmap cannot intersect a rewrite group, so
+    /// the remapper skips it instead of opening a file that is not there.
+    #[tokio::test]
+    async fn test_compaction_skips_a_definition_only_index() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // A second fragment, so compaction has two to rewrite into one.
+        let mut dataset = append_vectors(test_dir.path(), 100).await;
+        assert_eq!(dataset.get_fragments().len(), 2);
+
+        compact_files(
+            &mut dataset,
+            CompactionOptions {
+                target_rows_per_fragment: 1000,
+                ..Default::default()
+            },
+            None,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(dataset.get_fragments().len(), 1);
+        assert_eq!(dataset.count_rows(None).await.unwrap(), 200);
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(
+            segment_covers_nothing(&indices[0]),
+            "the index should still be a definition covering nothing"
+        );
+    }
+
+    /// A reader over one batch of `rows` random 16-dimensional vectors.
+    fn vector_reader(rows: usize) -> impl arrow_array::RecordBatchReader + Send + 'static {
+        let dimensions = 16;
+        let field = Field::new(
+            "vector",
+            DataType::FixedSizeList(
+                Arc::new(Field::new("item", DataType::Float32, true)),
+                dimensions,
+            ),
+            false,
+        );
+        let schema = Arc::new(Schema::new(vec![field]));
+        let values = generate_random_array(rows * dimensions as usize);
+        let vectors =
+            arrow_array::FixedSizeListArray::try_new_from_values(values, dimensions).unwrap();
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(vectors)]).unwrap();
+        RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema)
+    }
+
+    /// Build a dataset of `rows` random vectors, one fragment.
+    async fn small_vector_dataset(dir: &std::path::Path, rows: usize) -> Dataset {
+        Dataset::write(vector_reader(rows), dir.to_str().unwrap(), None)
+            .await
+            .unwrap()
+    }
+
+    /// A dataset of `rows` rows where only the first `non_null` hold a vector.
+    async fn partly_null_vector_dataset(
+        dir: &std::path::Path,
+        rows: usize,
+        non_null: usize,
+    ) -> Dataset {
+        let dimensions = 16;
+        let field = Field::new(
+            "vector",
+            DataType::FixedSizeList(
+                Arc::new(Field::new("item", DataType::Float32, true)),
+                dimensions,
+            ),
+            true,
+        );
+        let schema = Arc::new(Schema::new(vec![field]));
+        let mut builder = arrow_array::builder::FixedSizeListBuilder::new(
+            arrow_array::builder::Float32Builder::new(),
+            dimensions,
+        );
+        for row in 0..rows {
+            if row < non_null {
+                for value in 0..dimensions {
+                    builder.values().append_value(value as f32);
+                }
+                builder.append(true);
+            } else {
+                for _ in 0..dimensions {
+                    builder.values().append_null();
+                }
+                builder.append(false);
+            }
+        }
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(builder.finish())]).unwrap();
+        let reader = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema);
+        Dataset::write(reader, dir.to_str().unwrap(), None)
+            .await
+            .unwrap()
+    }
+
+    /// The floor counts vectors, not rows.
+    ///
+    /// A column that is mostly null has plenty of rows and nothing to train on,
+    /// so it has to degrade rather than hand too few vectors to the quantizer.
+    #[tokio::test]
+    async fn test_create_index_counts_vectors_not_rows() {
+        let test_dir = tempfile::tempdir().unwrap();
+        // 1000 rows clear the floor; the 100 actual vectors do not.
+        let mut dataset = partly_null_vector_dataset(test_dir.path(), 1000, 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .expect("a mostly-null column must not be handed to the quantizer");
+
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(
+            segment_covers_nothing(&indices[0]),
+            "100 vectors cannot train a 256-code quantizer"
+        );
+    }
+
+    /// Append `rows` more random vectors as a new fragment.
+    async fn append_vectors(dir: &std::path::Path, rows: usize) -> Dataset {
+        Dataset::write(
+            vector_reader(rows),
+            dir.to_str().unwrap(),
+            Some(WriteParams {
+                mode: WriteMode::Append,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap()
+    }
+
+    /// The whole point of degrading: the index fills in once the data arrives.
+    ///
+    /// A definition-only index has no model to append to, so optimizing has to
+    /// train it from scratch and then cover every fragment, including the small
+    /// one that existed before the threshold was met.
+    #[tokio::test]
+    async fn test_degraded_index_trains_on_optimize_once_data_arrives() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 8, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // Degraded: 100 rows cannot train a 256-code quantizer.
+        let indices = dataset.load_indices().await.unwrap();
+        assert!(segment_covers_nothing(&indices[0]));
+
+        // 100 -> 500 across two fragments, clearing the floor.
+        let mut dataset = append_vectors(test_dir.path(), 400).await;
+        assert_eq!(dataset.count_rows(None).await.unwrap(), 500);
+        assert_eq!(dataset.get_fragments().len(), 2);
+
+        dataset
+            .optimize_indices(&OptimizeOptions::default())
+            .await
+            .expect("a definition-only index must train once the data is there");
+
+        // Trained and covering: both fragments, nothing left unindexed.
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        let covered = indices[0].fragment_bitmap.as_ref().unwrap();
+        assert_eq!(
+            covered.len() as usize,
+            dataset.get_fragments().len(),
+            "every fragment must be covered, including the pre-threshold one"
+        );
+        assert!(
+            dataset
+                .unindexed_fragments("vector_idx")
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    /// A vector query works while the index is still only a definition.
+    ///
+    /// The index covers no fragments, so every row is unindexed and the search
+    /// has to answer from the table itself rather than opening an index file
+    /// that was never written.
+    #[tokio::test]
+    async fn test_vector_query_against_a_definition_only_index() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let rows = 100;
+        let mut dataset = small_vector_dataset(test_dir.path(), rows).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+        assert!(segment_covers_nothing(
+            &dataset.load_indices().await.unwrap()[0]
+        ));
+
+        let query = vec![0.0_f32; 16];
+        let results = dataset
+            .scan()
+            .nearest("vector", &Float32Array::from(query), 5)
+            .unwrap()
+            .try_into_batch()
+            .await
+            .expect("a query must not open an index file that was never written");
+        assert_eq!(results.num_rows(), 5);
+    }
+
+    /// `fast_search` against an index that is still a definition finds nothing.
+    ///
+    /// It restricts the search to what the index covers, and a definition
+    /// covers no rows, so there is nothing to return rather than a fallback
+    /// scan of the table.
+    #[tokio::test]
+    async fn test_fast_search_against_a_definition_only_index() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        let query = Float32Array::from(vec![0.0_f32; 16]);
+        let results = dataset
+            .scan()
+            .nearest("vector", &query, 5)
+            .unwrap()
+            .fast_search()
+            .try_into_batch()
+            .await
+            .expect("fast_search must not fail on an index that covers nothing");
+        assert_eq!(results.num_rows(), 0);
+    }
+
+    /// Statistics for an index that is still a definition report no coverage.
+    ///
+    /// The same shape the scalar side reports: the index is listed, indexed
+    /// rows are zero, and every row counts as unindexed.
+    #[tokio::test]
+    async fn test_statistics_for_a_definition_only_index() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let rows = 100;
+        let mut dataset = small_vector_dataset(test_dir.path(), rows).await;
+
+        let params = VectorIndexParams::ivf_pq(10, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        let stats: serde_json::Value =
+            serde_json::from_str(&dataset.index_statistics("vector_idx").await.unwrap()).unwrap();
+        assert_eq!(stats["num_indexed_rows"].as_u64(), Some(0));
+        assert_eq!(stats["num_unindexed_rows"].as_u64(), Some(rows as u64));
+    }
+
+    /// An index type without a codebook trains on a table a codebook could not.
+    ///
+    /// Only PQ needs a vector per code; RQ needs a pair of values and flat
+    /// storage needs none, so 100 vectors are enough for these to cover the
+    /// table rather than degrade to a definition.
+    #[rstest]
+    #[case::ivf_flat(VectorIndexParams::ivf_flat(1, DistanceType::L2))]
+    #[case::ivf_rq(VectorIndexParams::ivf_rq(1, 8, DistanceType::L2))]
+    #[tokio::test]
+    async fn test_create_index_without_a_codebook_trains_on_a_small_table(
+        #[case] params: VectorIndexParams,
+    ) {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .expect("an index with no codebook has no row floor to clear");
+
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(
+            !segment_covers_nothing(&indices[0]),
+            "should have trained and covered the table"
+        );
+    }
+
+    /// A multivector row holds a list, so the floor counts the vectors in the
+    /// lists rather than the rows that carry them.
+    #[tokio::test]
+    async fn test_multivector_floor_counts_vectors_not_rows() {
+        let sparse = {
+            let mut lengths = vec![0_usize; 100];
+            lengths[0] = 10;
+            lengths
+        };
+        // Lists per row, whether that trains, and the count that decides it.
+        let cases: [(Vec<usize>, bool, &str); 4] = [
+            (vec![10; 100], true, "100 rows of 10 vectors give 1,000"),
+            (vec![2; 10], false, "10 rows of 2 vectors give 20"),
+            (sparse, false, "one row holding 10 vectors still gives 10"),
+            (vec![0; 100], false, "empty lists give none"),
+        ];
+
+        for (lengths, trains, why) in cases {
+            let test_dir = tempfile::tempdir().unwrap();
+            let mut dataset = multivector_dataset_with(test_dir.path(), &lengths).await;
+
+            // Multivector columns are cosine-only.
+            let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::Cosine, 1);
+            dataset
+                .create_index(&["vector"], IndexType::Vector, None, &params, false)
+                .await
+                .unwrap_or_else(|error| {
+                    panic!("{why}: a table below the floor takes the index rather than failing: {error}")
+                });
+
+            let indices = dataset.load_indices().await.unwrap();
+            assert_eq!(indices.len(), 1, "{why}");
+            let covers_nothing = segment_covers_nothing(&indices[0]);
+            assert_eq!(!covers_nothing, trains, "{why}");
         }
     }
 
+    /// A table of `rows`, each holding `vectors_per_row` vectors.
+    async fn multivector_dataset(
+        dir: &std::path::Path,
+        rows: usize,
+        vectors_per_row: usize,
+    ) -> Dataset {
+        multivector_dataset_with(dir, &vec![vectors_per_row; rows]).await
+    }
+
+    /// A multivector table whose row `i` holds `lengths[i]` vectors, so uneven
+    /// and empty lists can be built as easily as uniform ones.
+    async fn multivector_dataset_with(dir: &std::path::Path, lengths: &[usize]) -> Dataset {
+        use arrow_array::builder::{FixedSizeListBuilder, Float32Builder, ListBuilder};
+
+        let dimensions = 16;
+        let field = Field::new(
+            "vector",
+            DataType::List(Arc::new(Field::new(
+                "item",
+                DataType::FixedSizeList(
+                    Arc::new(Field::new("item", DataType::Float32, true)),
+                    dimensions,
+                ),
+                true,
+            ))),
+            true,
+        );
+        let schema = Arc::new(Schema::new(vec![field]));
+
+        let mut builder =
+            ListBuilder::new(FixedSizeListBuilder::new(Float32Builder::new(), dimensions));
+        for (row, &length) in lengths.iter().enumerate() {
+            for vector in 0..length {
+                for value in 0..dimensions {
+                    builder
+                        .values()
+                        .values()
+                        .append_value((row + vector + value as usize) as f32 + 0.5);
+                }
+                builder.values().append(true);
+            }
+            builder.append(true);
+        }
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(builder.finish())]).unwrap();
+        let reader = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema);
+        Dataset::write(reader, dir.to_str().unwrap(), None)
+            .await
+            .unwrap()
+    }
+
+    /// The partition cap counts a multivector row's whole list too.
+    ///
+    /// 1,000 vectors support three partitions; counting the 100 rows instead
+    /// would allow only one.
+    #[tokio::test]
+    async fn test_partition_cap_counts_multivector_lists() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = multivector_dataset(test_dir.path(), 100, 10).await;
+
+        let params = VectorIndexParams::ivf_pq(8, 8, 4, DistanceType::Cosine, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // 1,000 / 256 = 3, capped from the 8 requested.
+        assert_eq!(trained_partitions(&dataset).await, vec![3]);
+    }
+
+    /// The partition cap counts vectors, so blanks cannot inflate it.
+    ///
+    /// 3,000 rows holding 300 vectors support one partition, not the eight a
+    /// row count would appear to allow.
+    #[tokio::test]
+    async fn test_partition_cap_ignores_null_rows() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = partly_null_vector_dataset(test_dir.path(), 3000, 300).await;
+
+        let params = VectorIndexParams::ivf_pq(8, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // 300 / 256 = 1.
+        assert_eq!(trained_partitions(&dataset).await, vec![1]);
+    }
+
+    /// Deleting every row leaves the index defined.
+    ///
+    /// The definition is the user's declaration, not a property of the data, so
+    /// emptying the table must not withdraw it — otherwise reloading a table
+    /// silently drops the indexes it was created with.
+    #[tokio::test]
+    async fn test_vector_index_survives_deleting_all_rows() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 1024).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 8, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+        assert_eq!(dataset.load_indices().await.unwrap().len(), 1);
+
+        dataset.delete("true").await.unwrap();
+        assert_eq!(dataset.count_rows(None).await.unwrap(), 0);
+
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1, "the index definition must survive");
+
+        // And it is still there after a reopen, so it lives in the manifest
+        // rather than in whatever the session happened to hold.
+        let reopened = Dataset::open(test_dir.path().to_str().unwrap())
+            .await
+            .unwrap();
+        assert_eq!(reopened.load_indices().await.unwrap().len(), 1);
+    }
+
+    /// Retraining an index that is still only a definition does nothing.
+    ///
+    /// A definition has no segment to open, and the retrain path resolves that
+    /// before it reaches for one.
+    #[tokio::test]
+    async fn test_retrain_an_index_that_is_still_a_definition() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 100).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 4, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        dataset
+            .optimize_indices(&OptimizeOptions::retrain())
+            .await
+            .expect("retraining a definition must not look for a segment to open");
+
+        // Still 100 vectors, so still a definition.
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(segment_covers_nothing(&indices[0]));
+    }
+
+    /// A table that shrinks below the training threshold keeps working.
+    ///
+    /// Optimizing was unrunnable in this state: the quantizer cannot train on
+    /// what is left, and erroring there blocks index maintenance outright, with
+    /// dropping and recreating the index as the only way out.
+    #[tokio::test]
+    async fn test_optimize_indices_after_shrinking_below_the_threshold() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 1024).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 8, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        // 1024 -> 224, under a 256-code quantizer's floor.
+        dataset.delete("_rowid < 800").await.unwrap();
+        assert_eq!(dataset.count_rows(None).await.unwrap(), 224);
+
+        // Retrain, not append: append-mode has no new data to index here and so
+        // never reaches the quantizer, which is the path that fails.
+        dataset
+            .optimize_indices(&OptimizeOptions::retrain())
+            .await
+            .expect("optimizing a table that shrank past the threshold must not fail");
+
+        assert_eq!(dataset.load_indices().await.unwrap().len(), 1);
+    }
+
+    /// Updating every row leaves the index defined, and optimizing still runs.
+    #[tokio::test]
+    async fn test_vector_index_survives_updating_all_rows() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let mut dataset = small_vector_dataset(test_dir.path(), 1024).await;
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 8, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .unwrap();
+
+        let update = crate::dataset::UpdateBuilder::new(Arc::new(dataset.clone()))
+            .set("vector", "vector")
+            .unwrap()
+            .build()
+            .unwrap();
+        let updated = update.execute().await.unwrap();
+        let mut dataset = updated.new_dataset.as_ref().clone();
+
+        assert_eq!(dataset.load_indices().await.unwrap().len(), 1);
+        dataset
+            .optimize_indices(&OptimizeOptions::default())
+            .await
+            .expect("optimizing after a full update must not fail");
+        assert_eq!(dataset.load_indices().await.unwrap().len(), 1);
+    }
+
+    /// A table with no rows at all takes an index, which is the case every
+    /// other database allows and the one a fresh or reloaded table is in.
+    #[tokio::test]
+    async fn test_create_vector_index_on_empty_table() {
+        let test_dir = tempfile::tempdir().unwrap();
+        let dimensions = 16;
+        let field = Field::new(
+            "vector",
+            DataType::FixedSizeList(
+                Arc::new(Field::new("item", DataType::Float32, true)),
+                dimensions,
+            ),
+            false,
+        );
+        let schema = Arc::new(Schema::new(vec![field]));
+        let reader = RecordBatchIterator::new(
+            Vec::<std::result::Result<RecordBatch, arrow_schema::ArrowError>>::new(),
+            schema.clone(),
+        );
+        let mut dataset = Dataset::write(reader, test_dir.path().to_str().unwrap(), None)
+            .await
+            .unwrap();
+
+        let params = VectorIndexParams::ivf_pq(1, 8, 96, DistanceType::L2, 1);
+        dataset
+            .create_index(&["vector"], IndexType::Vector, None, &params, false)
+            .await
+            .expect("an empty table still accepts an index");
+
+        let indices = dataset.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert!(segment_covers_nothing(&indices[0]));
+
+        // Reopening has to find the same definition: it is carried by the
+        // manifest, not by a file on disk.
+        let reopened = Dataset::open(test_dir.path().to_str().unwrap())
+            .await
+            .unwrap();
+        let indices = reopened.load_indices().await.unwrap();
+        assert_eq!(indices.len(), 1);
+        assert_eq!(indices[0].name, "vector_idx");
+    }
+
     #[tokio::test]
     async fn test_create_bitmap_index() {
         let test_dir = TempStrDir::default();
diff --git a/rust/lance/src/index/append.rs b/rust/lance/src/index/append.rs
index 245c861de72..81dbcda2df1 100644
--- a/rust/lance/src/index/append.rs
+++ b/rust/lance/src/index/append.rs
@@ -32,7 +32,7 @@ use super::vector::ivf::{
     SteadyStateRebalance, VectorSegmentCompatibility, index_type_for_segmented_optimize,
     optimize_vector_indices, select_steady_state_rebalance, vector_segment_compatibility,
 };
-use super::vector::{LogicalVectorIndex, fresh_vector_segment_params};
+use super::vector::{LogicalVectorIndex, VectorIndexParams, details, fresh_vector_segment_params};
 use super::{CreateIndexBuilder, DatasetIndexInternalExt};
 use crate::dataset::Dataset;
 use crate::dataset::index::LanceIndexStoreExt;
@@ -614,6 +614,12 @@ async fn merge_scalar_indices<'a>(
 }
 
 async fn metadata_is_vector_index(dataset: &Dataset, index: &IndexMetadata) -> Result<bool> {
+    // Declared type first: an index awaiting training declares itself a vector
+    // index and has no file to find it by.
+    if index.index_details.is_some() {
+        return Ok(crate::index::segment_has_vector_details(index));
+    }
+
     if let Some(files) = &index.files {
         return Ok(files.iter().any(|file| file.path == INDEX_FILE_NAME));
     }
@@ -658,33 +664,61 @@ pub async fn merge_indices<'a>(
     .await
 }
 
-async fn build_fresh_vector_segment(
+/// Whether a rebuild of `params` should expect the definition alone, because
+/// the column does not hold the vectors its quantizer needs.
+async fn expects_definition_only(
+    dataset: &Dataset,
+    field_path: &str,
+    params: &VectorIndexParams,
+) -> Result<bool> {
+    let Some(minimum) = super::vector::vector_quantizer_minimum_rows(&params.stages) else {
+        return Ok(false);
+    };
+    Ok(!super::vector::has_vectors_to_train(dataset, field_path, minimum).await?)
+}
+
+/// Retrain a whole segment over `fragment_bitmap` from index parameters alone.
+async fn rebuild_vector_segment(
     dataset: &Dataset,
-    logical_index: &LogicalVectorIndex,
+    name: &str,
+    params: &VectorIndexParams,
     field_path: &str,
     fragment_bitmap: &RoaringBitmap,
     progress: Arc<dyn IndexBuildProgress>,
 ) -> Result<IndexMetadata> {
-    let (reference_metadata, reference_index) = logical_index.iter().last().ok_or_else(|| {
-        Error::index(format!(
-            "Optimize vector index: logical index '{}' has no physical segments",
-            logical_index.name()
-        ))
-    })?;
-    let params = fresh_vector_segment_params(reference_metadata, reference_index.as_ref())?;
     let mut build_dataset = dataset.clone();
-    CreateIndexBuilder::new(
-        &mut build_dataset,
-        &[field_path],
-        IndexType::Vector,
-        &params,
-    )
-    .name(logical_index.name().to_string())
-    .replace(true)
-    .fragments(fragment_bitmap.iter().collect())
-    .progress(progress)
-    .execute_uncommitted()
-    .await
+    CreateIndexBuilder::new(&mut build_dataset, &[field_path], IndexType::Vector, params)
+        .name(name.to_string())
+        .replace(true)
+        .fragments(fragment_bitmap.iter().collect())
+        .progress(progress)
+        .execute_uncommitted()
+        .await
+}
+
+/// True when a segment carries only its definition, so there is nothing to open
+/// and nothing to append to: training is the work it is waiting for.
+///
+/// Both halves carry weight. Coverage alone is not enough: a segment
+/// initialized from another dataset's model holds centroids while covering no
+/// fragments, and retraining it from its parameters would discard them.
+///
+/// The file list is read as "no files recorded", absent included. The manifest
+/// stores it as a repeated field whose empty case means "sizes unknown", so a
+/// segment that recorded an empty list reads back as absent after any commit —
+/// treating absent as "has files" would stop recognising a definition the first
+/// time anything else commits to the table.
+///
+/// A segment whose rows were all deleted answers this the same way, and wants
+/// the same outcome: nothing can be served from it, and a rebuild is what makes
+/// it useful again.
+fn is_definition_only_segment(metadata: &IndexMetadata) -> bool {
+    let no_files_recorded = metadata.files.as_ref().is_none_or(|files| files.is_empty());
+    let covers_nothing = metadata
+        .fragment_bitmap
+        .as_ref()
+        .is_some_and(RoaringBitmap::is_empty);
+    no_files_recorded && covers_nothing
 }
 
 async fn scan_vector_fragments(
@@ -705,16 +739,85 @@ async fn scan_vector_fragments(
     scanner.try_into_stream().await
 }
 
+/// Train a column whose index is still only a definition, from the parameters
+/// that definition carries.
+///
+/// The result supersedes every segment the index had: they hold no data, and
+/// what this writes covers the whole column.
+async fn train_from_definition<'a>(
+    dataset: &Arc<Dataset>,
+    old_indices: &[&'a IndexMetadata],
+    field_path: &str,
+    options: &OptimizeOptions,
+) -> Result<IndexMergeResults<'a>> {
+    let params = old_indices
+        .last()
+        .and_then(|metadata| metadata.index_details.as_deref())
+        .and_then(details::vector_params_from_details)
+        .ok_or_else(|| {
+            Error::index(format!(
+                "Optimize vector index: index '{}' awaits training but carries no parameters",
+                old_indices[0].name
+            ))
+        })?;
+    let fragment_bitmap = dataset.fragment_bitmap.as_ref().clone();
+    let segment = rebuild_vector_segment(
+        dataset.as_ref(),
+        &old_indices[0].name,
+        &params,
+        field_path,
+        &fragment_bitmap,
+        options.progress.clone(),
+    )
+    .await?;
+    // Only a result that looks degraded is worth confirming; a rebuild that
+    // covered rows needs no count to justify it.
+    let definition_expected = is_definition_only_segment(&segment)
+        && expects_definition_only(dataset.as_ref(), field_path, &params).await?;
+    fresh_vector_segment_result(
+        segment,
+        &fragment_bitmap,
+        old_indices.to_vec(),
+        definition_expected,
+    )
+}
+
 fn fresh_vector_segment_result<'a>(
     segment: IndexMetadata,
     expected_fragment_bitmap: &RoaringBitmap,
     removed_indices: Vec<&'a IndexMetadata>,
+    definition_expected: bool,
 ) -> Result<IndexMergeResults<'a>> {
     let fragment_bitmap = segment.fragment_bitmap.ok_or_else(|| {
         Error::index(
             "Optimize vector index: newly built segment has no fragment bitmap".to_string(),
         )
     })?;
+    // A column with too few vectors to train the quantizer yields the
+    // definition alone: its rows return to unindexed, which is the state a
+    // table below the threshold belongs in. Accepted only when the caller
+    // established that beforehand and the segment wrote no files to match, so
+    // coverage lost for any other reason still fails below.
+    let wrote_no_files = segment.files.as_ref().is_some_and(|files| files.is_empty());
+    if definition_expected && wrote_no_files && fragment_bitmap.is_empty() {
+        return Ok(IndexMergeResults {
+            new_uuid: segment.uuid,
+            removed_indices,
+            new_fragment_bitmap: fragment_bitmap,
+            new_dataset_version: segment.dataset_version,
+            new_index_version: segment.index_version,
+            new_index_details: segment
+                .index_details
+                .map(|details| details.as_ref().clone())
+                .ok_or_else(|| {
+                    Error::index(
+                        "Optimize vector index: rebuilt definition has no index details"
+                            .to_string(),
+                    )
+                })?,
+            files: Vec::new(),
+        });
+    }
     if fragment_bitmap != *expected_fragment_bitmap {
         return Err(Error::index(format!(
             "Optimize vector index: newly built segment covers fragments {:?}, expected {:?}",
@@ -839,10 +942,29 @@ pub async fn merge_indices_with_unindexed_frags<'a>(
                 return Ok(merged);
             }
             let rebuild_dormant = live_segments.is_empty() && !dormant_segments.is_empty();
-            if rebuild_dormant && unindexed.is_empty() {
+            // A segment still awaiting training has no file to open and nothing
+            // to append to, so the whole column is trained from the parameters
+            // its definition carries, superseding every old segment. One such
+            // segment is enough to force that: the logical index is opened by
+            // name, so it would be reached whichever segments the caller asks
+            // for, and there is no file behind it.
+            let awaits_training = old_indices
+                .iter()
+                .any(|idx| is_definition_only_segment(idx));
+
+            // Rebuilding and first training both read the whole column, so
+            // neither has anything to do until a fragment sits outside the
+            // index.
+            if unindexed.is_empty() && (rebuild_dormant || awaits_training) {
                 return Ok(None);
             }
 
+            if awaits_training {
+                return train_from_definition(&dataset, old_indices, &field_path, options)
+                    .await
+                    .map(Some);
+            }
+
             let full_logical_index = dataset
                 .open_logical_vector_index(&field_path, &old_indices[0].name)
                 .await?;
@@ -872,19 +994,34 @@ pub async fn merge_indices_with_unindexed_frags<'a>(
             )?;
 
             if options.retrain || rebuild_dormant {
+                let (reference_metadata, reference_index) =
+                    logical_index.iter().last().ok_or_else(|| {
+                        Error::index(format!(
+                            "Optimize vector index: logical index '{}' has no physical segments",
+                            logical_index.name()
+                        ))
+                    })?;
+                let params =
+                    fresh_vector_segment_params(reference_metadata, reference_index.as_ref())?;
                 let fragment_bitmap = dataset.fragment_bitmap.as_ref().clone();
-                let segment = build_fresh_vector_segment(
+                let segment = rebuild_vector_segment(
                     dataset.as_ref(),
-                    &logical_index,
+                    logical_index.name(),
+                    &params,
                     &field_path,
                     &fragment_bitmap,
                     options.progress.clone(),
                 )
                 .await?;
+                // Only a result that looks degraded is worth confirming; a
+                // rebuild that covered rows needs no count to justify it.
+                let definition_expected = is_definition_only_segment(&segment)
+                    && expects_definition_only(dataset.as_ref(), &field_path, &params).await?;
                 return fresh_vector_segment_result(
                     segment,
                     &fragment_bitmap,
                     old_indices.to_vec(),
+                    definition_expected,
                 )
                 .map(Some);
             }
diff --git a/rust/lance/src/index/create.rs b/rust/lance/src/index/create.rs
index b3bd6178932..130ccc7a7c0 100644
--- a/rust/lance/src/index/create.rs
+++ b/rust/lance/src/index/create.rs
@@ -33,7 +33,7 @@ use lance_index::{
 };
 use lance_table::format::{IndexMetadata, list_index_files_with_sizes};
 use std::{collections::HashMap, future::IntoFuture, sync::Arc};
-use tracing::instrument;
+use tracing::{instrument, warn};
 use uuid::Uuid;
 
 use arrow_array::RecordBatchReader;
@@ -248,17 +248,52 @@ impl<'a> CreateIndexBuilder<'a> {
             .map(|resolved| resolved.canonical_path.as_str())
             .unwrap_or(quoted_column.as_str());
 
+        // A fragment list naming the whole table is a whole-table build, not a
+        // subset: `effective_vector_fragments` normalizes it to `None` so a
+        // retrain gets the same row floor as a create.
         let vector_fragments_for_validation =
             is_builtin_vector_index(self.index_type, self.params)
-                .then_some(self.fragments.as_deref())
+                .then(|| effective_vector_fragments(self.dataset, self.fragments.as_deref()))
                 .flatten();
+        let quantizer_minimum_rows = self
+            .params
+            .as_any()
+            .downcast_ref::<VectorIndexParams>()
+            .and_then(|params| super::vector::vector_quantizer_minimum_rows(&params.stages));
         let train = should_train_index(
             self.dataset,
             self.train,
-            vector_fragments_for_validation,
+            vector_fragments_for_validation.as_deref(),
+            quantizer_minimum_rows,
+            column,
         )
         .await?;
 
+        if !train {
+            // A partition count is not among the settings a definition records,
+            // so a caller who asked for one gets a different shape when the
+            // index finally trains.
+            if let Some(requested) = self
+                .params
+                .as_any()
+                .downcast_ref::<VectorIndexParams>()
+                .and_then(|params| {
+                    params.stages.iter().find_map(|stage| match stage {
+                        StageParams::Ivf(ivf) => ivf.num_partitions,
+                        _ => None,
+                    })
+                })
+            {
+                warn!(
+                    column,
+                    requested_num_partitions = requested,
+                    "Recording the index without training it: the partition count \
+                     will be derived from the data it trains on. Use \
+                     target_partition_size to state a shape that survives the wait."
+                );
+            }
+        }
+
         // Load indices from the disk. Names are reserved against every index the
         // manifest carries: one this build cannot read still owns its name, and
         // handing that name out again commits two indices under it.
@@ -571,11 +606,11 @@ impl<'a> CreateIndexBuilder<'a> {
                         "unable to cast index extension to vector".to_string(),
                     ))?;
 
+                // An extension that has not been trained writes nothing: the
+                // definition is the whole index until there is data for it.
                 if train {
                     ext.create_index(self.dataset, column, &index_id, self.params)
                         .await?;
-                } else {
-                    todo!("create empty vector index when train=false");
                 }
                 // Capture file sizes after vector index creation
                 let index_dir = self.dataset.indices_dir().join(index_id.to_string());
@@ -933,10 +968,21 @@ fn is_builtin_vector_index(index_type: IndexType, params: &dyn IndexParams) -> b
         && params.as_any().is::<VectorIndexParams>()
 }
 
+/// Whether there is enough data to train, as opposed to recording the
+/// definition and training later.
+///
+/// A quantizer needs one row per code to train at all. Below that the answer is
+/// the same as `train=false`: keep the definition, cover no rows, and leave
+/// `optimize_indices` to train it once the column fills.
+///
+/// An index type with a quantizer is measured in non-null vectors, since a
+/// column of nulls trains nothing; every other type is satisfied by any row.
 async fn should_train_index(
     dataset: &Dataset,
     train: bool,
     vector_fragments: Option<&[u32]>,
+    minimum_rows: Option<usize>,
+    column: &str,
 ) -> Result<bool> {
     if !train {
         return Ok(false);
@@ -946,12 +992,20 @@ async fn should_train_index(
         return Ok(false);
     }
 
+    // A fragment subset is a caller-driven segment build: the caller chose the
+    // fragments, often supplies the IVF model, and owns the row math. Only
+    // whole-table builds fall back to a definition-only index.
     if let Some(fragment_ids) = vector_fragments {
         dataset.get_fragments_from_ids(fragment_ids)?;
         return Ok(true);
     }
 
-    Ok(dataset.count_rows(None).await? > 0)
+    // Only a quantizer counts vectors; every other index type is satisfied by
+    // any row at all.
+    let Some(minimum) = minimum_rows.map(|minimum| minimum.max(1)) else {
+        return Ok(dataset.count_rows(None).await? > 0);
+    };
+    super::vector::has_vectors_to_train(dataset, column, minimum).await
 }
 
 fn vector_params_have_precomputed_ivf(params: &VectorIndexParams) -> bool {
diff --git a/rust/lance/src/index/vector.rs b/rust/lance/src/index/vector.rs
index 061a2f39d85..3d29547c52e 100644
--- a/rust/lance/src/index/vector.rs
+++ b/rust/lance/src/index/vector.rs
@@ -21,10 +21,12 @@ mod fixture_test;
 
 use self::{ivf::*, pq::PQIndex};
 use arrow_array::Array;
+use arrow_array::cast::AsArray;
 use arrow_schema::{DataType, Schema};
 use builder::{IvfIndexBuilder, VectorIndexBuildSummary};
 use datafusion::physical_plan::SendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
+use futures::TryStreamExt;
 use futures::stream;
 use lance_core::utils::tempfile::TempStdDir;
 use lance_file::versions::v1::reader::FileReader as V1FileReader;
@@ -505,6 +507,166 @@ impl IndexParams for VectorIndexParams {
     }
 }
 
+/// Whether `column` holds at least `minimum` vectors to train a quantizer from.
+///
+/// The metadata-only row total settles anything already short of the floor, so
+/// the validity count runs only where its answer can change the verdict.
+pub async fn has_vectors_to_train(dataset: &Dataset, column: &str, minimum: usize) -> Result<bool> {
+    let total = dataset.count_rows(None).await?;
+    if total < minimum && !is_multivector(dataset, column)? {
+        return Ok(false);
+    }
+    Ok(count_trainable_vectors(dataset, column, total, minimum).await? >= minimum)
+}
+
+/// Vectors in `column` available to train from.
+///
+/// A multivector row holds a list, so its rows bound the vectors from neither
+/// side — ten rows can carry a thousand, and an empty list carries none. Those
+/// are summed from the lists themselves; everything else counts the rows that
+/// hold a vector. Both callers compare the count with a threshold, so `enough`
+/// stops the walk as soon as the answer is settled.
+pub(crate) async fn count_trainable_vectors(
+    dataset: &Dataset,
+    column: &str,
+    total_rows: usize,
+    enough: usize,
+) -> Result<usize> {
+    if !is_multivector(dataset, column)? {
+        return count_non_null_vectors(dataset, column, total_rows).await;
+    }
+
+    let mut scanner = dataset.scan();
+    scanner.project(&[column])?;
+    let mut batches = scanner.try_into_stream().await?;
+    let mut vectors = 0_usize;
+    while let Some(batch) = batches.try_next().await? {
+        let lists = utils::get_column_from_batch(&batch, column)?;
+        let lists = lists.as_list::<i32>();
+        for row in 0..lists.len() {
+            if !lists.is_null(row) {
+                vectors = vectors.saturating_add(lists.value_length(row) as usize);
+            }
+        }
+        if vectors >= enough {
+            return Ok(vectors);
+        }
+    }
+    Ok(vectors)
+}
+
+fn is_multivector(dataset: &Dataset, column: &str) -> Result<bool> {
+    let (vector_type, _) = get_vector_type(dataset.schema(), column)?;
+    Ok(matches!(vector_type, DataType::List(_)))
+}
+
+/// Count the rows of `column` holding a vector.
+///
+/// A column that cannot hold nulls has one per row, so `total_rows` is the
+/// answer and nothing is read.
+async fn count_non_null_vectors(
+    dataset: &Dataset,
+    column: &str,
+    total_rows: usize,
+) -> Result<usize> {
+    let nullable = dataset
+        .schema()
+        .field(column)
+        .is_none_or(|field| field.nullable);
+    if !nullable {
+        return Ok(total_rows);
+    }
+
+    dataset
+        .count_rows(Some(format!("{column} IS NOT NULL")))
+        .await
+}
+
+/// The partition count a whole-table build can train.
+///
+/// A partition's centroid is what search ranks to choose partitions, so one
+/// built from a handful of vectors prunes nothing while still paying the
+/// quantizer's precision loss. The vectors available therefore cap the count,
+/// at `vectors_per_partition` each.
+async fn supported_num_partitions(
+    dataset: &Dataset,
+    column: &str,
+    stages: &[StageParams],
+    requested: usize,
+    mode: &str,
+) -> Result<usize> {
+    let total = dataset.count_rows(None).await?;
+    let enough = vectors_for_partitions(stages, requested);
+    let vectors = count_trainable_vectors(dataset, column, total, enough).await?;
+    if vectors >= enough {
+        return Ok(requested);
+    }
+
+    let supported = match vectors_per_partition(stages) {
+        Some(target) => requested.min(recommended_num_partitions(vectors, target)),
+        None => requested.min(vectors).max(1),
+    };
+    if supported < requested {
+        log::warn!(
+            "{mode}: training {supported} of the {requested} requested IVF partitions; \
+             column '{column}' holds {vectors} vectors"
+        );
+    }
+    Ok(supported)
+}
+
+/// Vectors needed before `partitions` can each train a quantizer of their own.
+///
+/// The upper edge of the reduced-partition band: below this a build trains
+/// fewer partitions than asked for.
+fn vectors_for_partitions(stages: &[StageParams], partitions: usize) -> usize {
+    partitions.saturating_mul(vectors_per_partition(stages).unwrap_or(1))
+}
+
+/// The vectors one partition's centroid should be fitted on, or `None` when the
+/// index stores vectors uncompressed.
+///
+/// A codebook's precision loss is paid on every vector, so a centroid fitted on
+/// a handful prunes nothing and still costs it. Without a codebook there is no
+/// precision to lose, and KMeans needing a vector per centroid is the only
+/// limit.
+fn vectors_per_partition(stages: &[StageParams]) -> Option<usize> {
+    vector_quantizer_minimum_rows(stages)?;
+    // IVF training fits each centroid on `sample_rate` vectors, which is also
+    // where KMeans starts warning (`k * 256` at the default rate). That number
+    // belongs to the IVF stage: an 8-bit codebook holds the same 256 entries by
+    // coincidence, and a 4-bit one asking only 16 vectors per partition would
+    // train centroids that prune nothing.
+    Some(
+        stages
+            .iter()
+            .find_map(|stage| match stage {
+                StageParams::Ivf(ivf) => Some(ivf.sample_rate),
+                _ => None,
+            })
+            .unwrap_or_else(|| IvfBuildParams::default().sample_rate)
+            .max(1),
+    )
+}
+
+/// Rows a PQ codebook needs before it can be trained.
+///
+/// One row per code, which is `2^num_bits` — 256 at the default 8 bits, and not
+/// 256 at any other setting. `None` when the width does not fit a `usize`.
+pub fn pq_quantizer_minimum_rows(num_bits: usize) -> Option<usize> {
+    1_usize.checked_shl(num_bits as u32)
+}
+
+/// Rows a vector index's quantizer needs before it can be trained.
+///
+/// `None` for an index type whose stages name no quantizer with a row floor.
+pub(crate) fn vector_quantizer_minimum_rows(stages: &[StageParams]) -> Option<usize> {
+    stages.iter().find_map(|stage| match stage {
+        StageParams::PQ(pq) => pq_quantizer_minimum_rows(pq.num_bits),
+        _ => None,
+    })
+}
+
 /// Prepare the shared build inputs used by both direct local builds and
 /// staged shard builds.
 ///
@@ -577,7 +739,18 @@ async fn prepare_vector_segment_build(
                 centroids.len()
             )));
         }
-        (Some(num_partitions), _) => num_partitions,
+        (Some(num_partitions), Some(_)) => num_partitions,
+        // A partition's centroid is what search ranks to choose partitions, so
+        // one built from a handful of vectors prunes nothing and still costs
+        // the quantizer's precision. So the vectors available cap how many
+        // partitions a whole-table build trains, at the IVF sample rate each.
+        //
+        // A fragment subset keeps the count it was given: the segments of one
+        // logical index have to agree on it.
+        (Some(requested), None) if fragment_ids.is_none() => {
+            supported_num_partitions(dataset, column, stages, requested, mode).await?
+        }
+        (Some(num_partitions), None) => num_partitions,
         (None, Some(centroids)) => centroids.len(),
         (None, None) => {
             let num_rows = match fragment_ids {
@@ -1563,25 +1736,20 @@ pub(crate) async fn build_vector_index_incremental(
     }
 }
 
-/// Build an empty vector index without training on data
+/// Create a vector index that carries its definition and no data.
+///
+/// The parameters live in the index's `index_details` and the fragment bitmap
+/// is empty, so it covers no rows and has no file to open. `optimize_indices`
+/// trains it once the column holds enough vectors.
 #[instrument(level = "debug", skip_all)]
 pub(crate) async fn build_empty_vector_index(
     _dataset: &Dataset,
-    column: &str,
-    name: &str,
+    _column: &str,
+    _name: &str,
     _uuid: Uuid,
     _params: &VectorIndexParams,
 ) -> Result<Vec<IndexFile>> {
-    // For now, return a NotImplementedError to indicate this functionality
-    // is still being developed
-    Err(Error::not_supported_source(
-        format!(
-            "Creating empty vector indices with train=False is not yet implemented. \
-        Index '{}' for column '{}' cannot be created without training.",
-            name, column
-        )
-        .into(),
-    ))
+    Ok(Vec::new())
 }
 
 #[instrument(level = "debug", skip_all, fields(old_uuid = old_uuid.to_string(), new_uuid = new_uuid.to_string()))]
@@ -2341,11 +2509,11 @@ mod tests {
         let source_uri = format!("{}/source", test_dir.as_str());
         let target_uri = format!("{}/target", test_dir.as_str());
 
-        // Create source dataset with vector column (need at least 256 rows for PQ training)
+        // Ten partitions, each needing a codebook's worth of vectors.
         let source_reader = lance_datagen::gen_batch()
             .col("id", array::step::<Int32Type>())
             .col("vector", array::rand_vec::<Float32Type>(32.into()))
-            .into_reader_rows(RowCount::from(300), BatchCount::from(1));
+            .into_reader_rows(RowCount::from(2560), BatchCount::from(1));
         let mut source_dataset = Dataset::write(source_reader, &source_uri, None)
             .await
             .unwrap();
diff --git a/rust/lance/src/index/vector/ivf/v2.rs b/rust/lance/src/index/vector/ivf/v2.rs
index d4f3134c2d4..151bd6f5675 100644
--- a/rust/lance/src/index/vector/ivf/v2.rs
+++ b/rust/lance/src/index/vector/ivf/v2.rs
@@ -3028,12 +3028,20 @@ mod tests {
     // while 20 neighbors provide a useful recall oracle.
     const PQ_MATRIX_NUM_ROWS: usize = 320;
     const PQ_MATRIX_K: usize = 20;
-    // An 8-bit PQ codebook has 256 centroids, so this is the smallest valid
-    // training fixture shared by the 8-bit and 4-bit runtime cases.
-    const LIGHTWEIGHT_PQ_ROWS: usize = 256;
     const LIGHTWEIGHT_PQ_PARTITIONS: usize = 2;
+    const LIGHTWEIGHT_IVF_SAMPLE_RATE: usize = 16;
+    // An 8-bit PQ codebook has 256 codes and needs a training vector for each,
+    // so this is the smallest valid fixture, shared by the 8-bit and 4-bit
+    // runtime cases.
+    const LIGHTWEIGHT_PQ_ROWS: usize = 256;
     const LIGHTWEIGHT_PQ_SUB_VECTORS: usize = 4;
 
+    /// Partitions a fixture of `rows` supports when each centroid is fitted on
+    /// `sample_rate` vectors, capped by the count the build asked for.
+    fn supported_partitions(rows: usize, sample_rate: usize, requested: usize) -> usize {
+        requested.min(rows / sample_rate).max(1)
+    }
+
     lance_testing::define_stage_event_progress!(RecordingProgress, IndexBuildProgress, Result<()>);
 
     #[test]
@@ -3843,7 +3851,7 @@ mod tests {
 
         let mut ivf_params = IvfBuildParams::new(LIGHTWEIGHT_PQ_PARTITIONS);
         ivf_params.max_iters = 2;
-        ivf_params.sample_rate = 16;
+        ivf_params.sample_rate = LIGHTWEIGHT_IVF_SAMPLE_RATE;
         let pq_params = lightweight_pq_params_with_bits(num_bits);
         let expected_num_sub_vectors = pq_params.num_sub_vectors;
         let params = if use_hnsw {
@@ -3872,10 +3880,12 @@ mod tests {
         let expected_index_type = if use_hnsw { "IVF_HNSW_PQ" } else { "IVF_PQ" };
         let expected_sub_index = if use_hnsw { "HNSW" } else { "PQ" };
         assert_eq!(stats["index_type"], expected_index_type);
-        assert_eq!(
-            stats["indices"][0]["num_partitions"],
-            LIGHTWEIGHT_PQ_PARTITIONS
+        let expected_partitions = supported_partitions(
+            LIGHTWEIGHT_PQ_ROWS,
+            LIGHTWEIGHT_IVF_SAMPLE_RATE,
+            LIGHTWEIGHT_PQ_PARTITIONS,
         );
+        assert_eq!(stats["indices"][0]["num_partitions"], expected_partitions);
         assert_eq!(
             stats["indices"][0]["sub_index"]["index_type"],
             expected_sub_index
@@ -5603,7 +5613,7 @@ mod tests {
         }
     }
 
-    fn pq_matrix_batch<T>() -> RecordBatch
+    fn pq_matrix_batch<T>(rows: usize) -> RecordBatch
     where
         T: ArrowPrimitiveType + 'static,
         T::Native: Copy + 'static,
@@ -5614,7 +5624,7 @@ mod tests {
             .with_seed(Seed(42))
             .col("id", array::step::<UInt64Type>())
             .col("vector", array::rand_vec::<T>(Dimension::from(DIM as u32)))
-            .into_batch_rows(RowCount::from(PQ_MATRIX_NUM_ROWS as u64))
+            .into_batch_rows(RowCount::from(rows as u64))
             .unwrap()
     }
 
@@ -5648,12 +5658,15 @@ mod tests {
 
         let test_dir = TempStrDir::default();
         let test_uri = test_dir.as_str();
-        let batch = pq_matrix_batch::<Float32Type>();
+        let params = pq_matrix_params(nlist, distance_type, version.clone());
+        let batch = pq_matrix_batch::<Float32Type>(PQ_MATRIX_NUM_ROWS);
+        // `pq_matrix_params` fits each centroid on the whole fixture.
+        let expected_partitions =
+            supported_partitions(PQ_MATRIX_NUM_ROWS, PQ_MATRIX_NUM_ROWS, nlist);
         let schema = batch.schema();
         let query = batch["vector"].as_fixed_size_list().value(0);
         let batches = RecordBatchIterator::new(vec![Ok(batch)], schema);
         let mut dataset = Dataset::write(batches, test_uri, None).await.unwrap();
-        let params = pq_matrix_params(nlist, distance_type, version.clone());
         dataset
             .create_index(
                 &["vector"],
@@ -5673,7 +5686,7 @@ mod tests {
         let index = &indices[0];
         assert_eq!(index["index_type"], "IVF_PQ");
         assert_eq!(index["metric_type"], distance_type.to_string());
-        assert_eq!(index["num_partitions"], nlist);
+        assert_eq!(index["num_partitions"], expected_partitions);
         assert_eq!(index["sub_index"]["index_type"], "PQ");
         assert_eq!(
             index["index_file_version"],
@@ -6051,7 +6064,7 @@ mod tests {
     #[tokio::test]
     async fn test_ivf_pq_f64_smoke(#[case] version: IndexFileVersion) {
         let test_dir = TempStrDir::default();
-        let batch = pq_matrix_batch::<Float64Type>();
+        let batch = pq_matrix_batch::<Float64Type>(PQ_MATRIX_NUM_ROWS);
         let schema = batch.schema();
         let vectors = Arc::new(batch["vector"].as_fixed_size_list().clone());
         let batches = RecordBatchIterator::new(vec![Ok(batch)], schema);
@@ -6872,11 +6885,11 @@ mod tests {
     #[tokio::test]
     async fn test_index_stats(
         #[values(
-            (VectorIndexParams::ivf_flat(4, DistanceType::Hamming), IndexType::IvfFlat),
-            (VectorIndexParams::ivf_pq(4, 8, 8, DistanceType::L2, 10), IndexType::IvfPq),
+            (VectorIndexParams::ivf_flat(2, DistanceType::Hamming), IndexType::IvfFlat),
+            (VectorIndexParams::ivf_pq(2, 8, 8, DistanceType::L2, 10), IndexType::IvfPq),
             (VectorIndexParams::with_ivf_hnsw_sq_params(
                 DistanceType::Cosine,
-                IvfBuildParams::new(4),
+                IvfBuildParams::new(2),
                 Default::default(),
                 Default::default()
             ), IndexType::IvfHnswSq),
@@ -6887,7 +6900,7 @@ mod tests {
         let test_dir = TempStrDir::default();
         let test_uri = test_dir.as_str();
 
-        let nlist = 4;
+        let nlist = 2;
         let (mut dataset, _) = match params.metric_type {
             DistanceType::Hamming => generate_test_dataset::<UInt8Type>(test_uri, 0..2).await,
             _ => generate_test_dataset::<Float32Type>(test_uri, 0.0..1.0).await,
diff --git a/rust/lance/src/index/vector/utils.rs b/rust/lance/src/index/vector/utils.rs
index 012d8d3a703..94d93cb74ba 100644
--- a/rust/lance/src/index/vector/utils.rs
+++ b/rust/lance/src/index/vector/utils.rs
@@ -30,7 +30,7 @@ use crate::{Error, Result};
 /// - Simple column names: "column"
 /// - Nested paths: "parent.child" or "parent.child.grandchild"
 /// - Backtick-escaped field names: "parent.`field.with.dots`"
-fn get_column_from_batch(batch: &RecordBatch, column: &str) -> Result<ArrayRef> {
+pub(crate) fn get_column_from_batch(batch: &RecordBatch, column: &str) -> Result<ArrayRef> {
     // Try to get the column directly first (fast path for simple columns)
     if let Some(col) = batch.column_by_name(column) {
         return Ok(col.clone());
@@ -85,7 +85,7 @@ fn get_column_from_batch(batch: &RecordBatch, column: &str) -> Result<ArrayRef>
     Ok(current_array)
 }
 
-async fn estimate_multivector_vectors_per_row(
+pub(crate) async fn estimate_multivector_vectors_per_row(
     dataset: &Dataset,
     column: &str,
     num_rows: usize,