From d90747b9349ee2a7aa48b8055086e41ee876ea6b Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Thu, 11 Jun 2026 00:28:01 +0200
Subject: [PATCH 1/9] fix(python): added cleanup_frag_reuse_index

---
 python/python/lance/dataset.py         |  4 +++
 python/python/lance/lance/__init__.pyi |  1 +
 python/python/tests/test_optimize.py   | 43 ++++++++++++++++++++++++++
 python/src/dataset.rs                  | 11 +++++++
 4 files changed, 59 insertions(+)

diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py
index 831e194f9f3..3f6b553bf5f 100644
--- a/python/python/lance/dataset.py
+++ b/python/python/lance/dataset.py
@@ -4622,6 +4622,10 @@ def migrate_manifest_paths_v2(self):
         """
         self._ds.migrate_manifest_paths_v2()
 
+    def cleanup_frag_reuse_index(self) -> None:
+        """Prune obsolete generations from the ``__lance_frag_reuse`` system index."""
+        self._ds.cleanup_frag_reuse_index()
+
     def delete_config_keys(self, keys: list[str]) -> None:
         """Delete specified configuration keys from the dataset.
 
diff --git a/python/python/lance/lance/__init__.pyi b/python/python/lance/lance/__init__.pyi
index 38d82738063..9af7167940e 100644
--- a/python/python/lance/lance/__init__.pyi
+++ b/python/python/lance/lance/__init__.pyi
@@ -446,6 +446,7 @@ class _Dataset:
     ) -> Tuple[_Dataset, Transaction]: ...
     def validate(self): ...
     def migrate_manifest_paths_v2(self): ...
+    def cleanup_frag_reuse_index(self) -> None: ...
     def drop_columns(self, columns: List[str]): ...
     def add_columns_from_reader(
         self, reader: pa.RecordBatchReader, batch_size: Optional[int] = None
diff --git a/python/python/tests/test_optimize.py b/python/python/tests/test_optimize.py
index ccd889db116..ec3fbe53431 100644
--- a/python/python/tests/test_optimize.py
+++ b/python/python/tests/test_optimize.py
@@ -324,6 +324,49 @@ def test_defer_index_remap(tmp_path: Path):
     assert any(idx.name == "__lance_frag_reuse" for idx in indices)
 
 
+def test_cleanup_frag_reuse_index(tmp_path: Path):
+    """cleanup_frag_reuse_index prunes all reuse generations that every user
+    index has caught up to.
+
+    Setup: 6 fragments, one BTREE scalar index.  Compact with
+    defer_index_remap=True so the frag-reuse index is populated.  Rebuild the
+    scalar index (create_scalar_index with replace=True) so it is newer than the
+    reuse generation, meaning the generation can be pruned.  Then call
+    cleanup_frag_reuse_index and verify that num_versions drops to 0.
+    """
+    base_dir = tmp_path / "dataset"
+    data = pa.table({"i": range(6_000), "val": range(6_000)})
+    dataset = lance.write_dataset(data, base_dir, max_rows_per_file=1_000)
+    dataset.create_scalar_index("i", "BTREE")
+
+    dataset.delete("i < 500")
+    dataset.optimize.compact_files(
+        target_rows_per_fragment=2_000, defer_index_remap=True, num_threads=1
+    )
+
+    dataset = lance.dataset(base_dir)
+    assert any(
+        idx.name == "__lance_frag_reuse" for idx in dataset.describe_indices()
+    ), "precondition: defer_index_remap must have created the frag-reuse index"
+
+    before_stats = dataset.stats.index_stats("__lance_frag_reuse")
+    assert before_stats["num_versions"] >= 1, (
+        "precondition: frag-reuse index must have at least one version before cleanup"
+    )
+
+    dataset.create_scalar_index("i", "BTREE", replace=True)
+    dataset = lance.dataset(base_dir)
+
+    dataset.cleanup_frag_reuse_index()
+
+    dataset = lance.dataset(base_dir)
+    after_stats = dataset.stats.index_stats("__lance_frag_reuse")
+    assert after_stats["num_versions"] == 0, (
+        f"cleanup_frag_reuse_index should have pruned all reuse generations "
+        f"but num_versions={after_stats['num_versions']}"
+    )
+
+
 @pytest.mark.filterwarnings("ignore::DeprecationWarning")
 def test_describe_indices_matches_list_indices_for_frag_reuse(tmp_path: Path):
     """describe_indices() and list_indices() must agree on the index_type
diff --git a/python/src/dataset.rs b/python/src/dataset.rs
index 8bfa81aeae4..e8e864d300a 100644
--- a/python/src/dataset.rs
+++ b/python/src/dataset.rs
@@ -2788,6 +2788,17 @@ impl Dataset {
         Ok(())
     }
 
+    fn cleanup_frag_reuse_index(&mut self) -> PyResult<()> {
+        let mut new_self = self.ds.as_ref().clone();
+        rt().block_on(
+            None,
+            lance::dataset::index::frag_reuse::cleanup_frag_reuse_index(&mut new_self),
+        )?
+        .map_err(|err: lance::Error| PyIOError::new_err(err.to_string()))?;
+        self.ds = Arc::new(new_self);
+        Ok(())
+    }
+
     fn drop_columns(&mut self, columns: Vec<String>) -> PyResult<()> {
         let mut new_self = self.ds.as_ref().clone();
         let columns: Vec<_> = columns.iter().map(|s| s.as_str()).collect();

From 8e65b016d45a3d9ec23ef37db63c9dfabb66099b Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Sat, 20 Jun 2026 14:15:38 +0200
Subject: [PATCH 2/9] fix(frag-reuse): surface cleanup metadata load failure
 instead of panicking

---
 python/python/tests/test_optimize.py       | 27 ++++++++++++++++------
 rust/lance/src/dataset/index/frag_reuse.rs | 21 ++++++++++++++---
 2 files changed, 38 insertions(+), 10 deletions(-)

diff --git a/python/python/tests/test_optimize.py b/python/python/tests/test_optimize.py
index ec3fbe53431..18534c5837d 100644
--- a/python/python/tests/test_optimize.py
+++ b/python/python/tests/test_optimize.py
@@ -325,14 +325,15 @@ def test_defer_index_remap(tmp_path: Path):
 
 
 def test_cleanup_frag_reuse_index(tmp_path: Path):
-    """cleanup_frag_reuse_index prunes all reuse generations that every user
-    index has caught up to.
+    """cleanup_frag_reuse_index prunes only the reuse generations that every user
+    index has caught up to, and never the ones still in use.
 
     Setup: 6 fragments, one BTREE scalar index.  Compact with
-    defer_index_remap=True so the frag-reuse index is populated.  Rebuild the
-    scalar index (create_scalar_index with replace=True) so it is newer than the
-    reuse generation, meaning the generation can be pruned.  Then call
-    cleanup_frag_reuse_index and verify that num_versions drops to 0.
+    defer_index_remap=True so the frag-reuse index is populated.  While the
+    scalar index has not caught up, cleanup must retain the generation.  After
+    rebuilding the scalar index (create_scalar_index with replace=True) it is
+    newer than the reuse generation, so cleanup may prune it and num_versions
+    drops to 0.
     """
     base_dir = tmp_path / "dataset"
     data = pa.table({"i": range(6_000), "val": range(6_000)})
@@ -350,10 +351,22 @@ def test_cleanup_frag_reuse_index(tmp_path: Path):
     ), "precondition: defer_index_remap must have created the frag-reuse index"
 
     before_stats = dataset.stats.index_stats("__lance_frag_reuse")
-    assert before_stats["num_versions"] >= 1, (
+    versions_before = before_stats["num_versions"]
+    assert versions_before >= 1, (
         "precondition: frag-reuse index must have at least one version before cleanup"
     )
 
+    # Negative case: index not yet remapped, so cleanup must retain the generation.
+    dataset.cleanup_frag_reuse_index()
+    dataset = lance.dataset(base_dir)
+    retained_stats = dataset.stats.index_stats("__lance_frag_reuse")
+    assert retained_stats["num_versions"] == versions_before, (
+        f"cleanup_frag_reuse_index must retain generations that an index has not "
+        f"caught up to, but num_versions went from {versions_before} to "
+        f"{retained_stats['num_versions']}"
+    )
+
+    # Positive case: rebuilding catches the index up, so cleanup can now prune all.
     dataset.create_scalar_index("i", "BTREE", replace=True)
     dataset = lance.dataset(base_dir)
 
diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index 4fbefcd4725..969087177bd 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -38,9 +38,8 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
         return Ok(());
     };
 
-    let frag_reuse_details = load_frag_reuse_index_details(dataset, frag_reuse_index_meta)
-        .await
-        .unwrap();
+    // Surface corrupt/stale metadata as a normal error instead of panicking on unwrap.
+    let frag_reuse_details = load_frag_reuse_index_details(dataset, frag_reuse_index_meta).await?;
 
     let mut retained_versions = Vec::new();
     let mut fragment_bitmaps = RoaringBitmap::new();
@@ -212,6 +211,22 @@ mod tests {
             is_index_remap_caught_up(&frag_reuse_details.versions[0], scalar_index).unwrap()
         );
 
+        // Cleanup must not prune a reuse version while an index has not caught up to it.
+        cleanup_frag_reuse_index(&mut dataset).await.unwrap();
+        let frag_reuse_index_meta = dataset
+            .load_index_by_name(FRAG_REUSE_INDEX_NAME)
+            .await
+            .unwrap()
+            .expect("Fragment reuse index must be available");
+        let frag_reuse_details = load_frag_reuse_index_details(&dataset, &frag_reuse_index_meta)
+            .await
+            .unwrap();
+        assert_eq!(
+            frag_reuse_details.versions.len(),
+            1,
+            "reuse version must be retained while an index is not caught up"
+        );
+
         // Remap and check index is caught up
         remapping::remap_column_index(&mut dataset, &["i"], index_name.clone())
             .await

From a26d17c2900051f334b193956e4ba096f371462a Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Sat, 20 Jun 2026 15:24:34 +0200
Subject: [PATCH 3/9] fix(frag-reuse): address review feedback on
 cleanup_frag_reuse_index

---
 python/python/lance/dataset.py             | 26 ++++++-
 python/python/tests/test_optimize.py       |  4 +-
 python/src/dataset.rs                      |  2 +-
 rust/lance/src/dataset/index/frag_reuse.rs | 90 ++++++++++++++++++----
 4 files changed, 102 insertions(+), 20 deletions(-)

diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py
index ec9406b9fe4..5bcf3edc037 100644
--- a/python/python/lance/dataset.py
+++ b/python/python/lance/dataset.py
@@ -4660,10 +4660,6 @@ def migrate_manifest_paths_v2(self):
         """
         self._ds.migrate_manifest_paths_v2()
 
-    def cleanup_frag_reuse_index(self) -> None:
-        """Prune obsolete generations from the ``__lance_frag_reuse`` system index."""
-        self._ds.cleanup_frag_reuse_index()
-
     def delete_config_keys(self, keys: list[str]) -> None:
         """Delete specified configuration keys from the dataset.
 
@@ -6845,6 +6841,28 @@ def optimize_indices(self, **kwargs):
         """
         self._dataset._ds.optimize_indices(**kwargs)
 
+    def cleanup_frag_reuse_index(self) -> None:
+        """Prune obsolete generations from the ``__lance_frag_reuse`` system index.
+
+        The fragment-reuse index is created by :meth:`compact_files` when
+        ``defer_index_remap=True``. It retains one generation per deferred
+        compaction so indices can be remapped lazily. Once every index has caught
+        up to a generation (for example after rebuilding it or calling
+        :meth:`optimize_indices`), that generation is obsolete and can be pruned.
+
+        This is safe to call at any time: a generation that an index has not yet
+        caught up to is always retained.
+
+        Examples
+        --------
+        >>> import lance
+        >>> import pyarrow as pa
+        >>> data = pa.table({"id": [1, 2, 3]})
+        >>> dataset = lance.write_dataset(data, "memory://frag_reuse_example")
+        >>> dataset.optimize.cleanup_frag_reuse_index()
+        """
+        self._dataset._ds.cleanup_frag_reuse_index()
+
     def enable_auto_cleanup(self, auto_cleanup_config: AutoCleanupConfig, **kwargs):
         """Enable autocleaning for an existing dataset.
 
diff --git a/python/python/tests/test_optimize.py b/python/python/tests/test_optimize.py
index ad4f6b6e34a..0bb7ecda587 100644
--- a/python/python/tests/test_optimize.py
+++ b/python/python/tests/test_optimize.py
@@ -357,7 +357,7 @@ def test_cleanup_frag_reuse_index(tmp_path: Path):
     )
 
     # Negative case: index not yet remapped, so cleanup must retain the generation.
-    dataset.cleanup_frag_reuse_index()
+    dataset.optimize.cleanup_frag_reuse_index()
     dataset = lance.dataset(base_dir)
     retained_stats = dataset.stats.index_stats("__lance_frag_reuse")
     assert retained_stats["num_versions"] == versions_before, (
@@ -370,7 +370,7 @@ def test_cleanup_frag_reuse_index(tmp_path: Path):
     dataset.create_scalar_index("i", "BTREE", replace=True)
     dataset = lance.dataset(base_dir)
 
-    dataset.cleanup_frag_reuse_index()
+    dataset.optimize.cleanup_frag_reuse_index()
 
     dataset = lance.dataset(base_dir)
     after_stats = dataset.stats.index_stats("__lance_frag_reuse")
diff --git a/python/src/dataset.rs b/python/src/dataset.rs
index a917719a2f6..c44bb19e2a9 100644
--- a/python/src/dataset.rs
+++ b/python/src/dataset.rs
@@ -2794,7 +2794,7 @@ impl Dataset {
             None,
             lance::dataset::index::frag_reuse::cleanup_frag_reuse_index(&mut new_self),
         )?
-        .map_err(|err: lance::Error| PyIOError::new_err(err.to_string()))?;
+        .infer_error()?;
         self.ds = Arc::new(new_self);
         Ok(())
     }
diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index 191c9bb3b2d..df5fe568e41 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -43,21 +43,20 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
 
     let mut retained_versions = Vec::new();
     let mut fragment_bitmaps = RoaringBitmap::new();
-    for version in frag_reuse_details.versions.iter() {
-        let check_results = indices
-            .iter()
-            .map(|idx| is_index_remap_caught_up(version, idx))
-            .collect::<Vec<_>>();
-
-        if check_results
-            .iter()
-            .any(|r| matches!(r, Err(Error::InvalidInput { .. })))
-        {
-            // If the check fails, the reuse version is likely corrupted, do not retain it.
-            continue;
+    'versions: for version in frag_reuse_details.versions.iter() {
+        let mut all_caught_up = true;
+        for idx in indices.iter() {
+            match is_index_remap_caught_up(version, idx) {
+                Ok(true) => {}
+                Ok(false) => all_caught_up = false,
+                // An InvalidInput error means the reuse version is likely corrupted; drop it.
+                Err(Error::InvalidInput { .. }) => continue 'versions,
+                // Any other error is unexpected; surface it instead of panicking.
+                Err(e) => return Err(e),
+            }
         }
 
-        if !check_results.into_iter().all(|r| r.unwrap()) {
+        if !all_caught_up {
             fragment_bitmaps.extend(version.new_frag_bitmap());
             retained_versions.push(version.clone());
         }
@@ -259,6 +258,71 @@ mod tests {
         ));
     }
 
+    /// Corrupt or stale fragment-reuse metadata must surface as a normal error,
+    /// not a panic. Regression for the `load_frag_reuse_index_details().unwrap()`
+    /// that used to crash the (Python) caller on unloadable metadata.
+    #[tokio::test]
+    async fn test_cleanup_frag_reuse_index_corrupt_metadata_errors() {
+        let mut dataset = lance_datagen::gen_batch()
+            .col("i", lance_datagen::array::step::<Int32Type>())
+            .into_ram_dataset(FragmentCount::from(6), FragmentRowCount::from(1000))
+            .await
+            .unwrap();
+
+        dataset
+            .create_index(
+                &["i"],
+                IndexType::Scalar,
+                Some("scalar".into()),
+                &ScalarIndexParams::default(),
+                false,
+            )
+            .await
+            .unwrap();
+
+        compact_files(
+            &mut dataset,
+            CompactionOptions {
+                target_rows_per_fragment: 2_000,
+                defer_index_remap: true,
+                ..Default::default()
+            },
+            None,
+        )
+        .await
+        .unwrap();
+
+        let frag_reuse_index_meta = dataset
+            .load_index_by_name(FRAG_REUSE_INDEX_NAME)
+            .await
+            .unwrap()
+            .expect("Fragment reuse index must be available");
+
+        // Replace the reuse index with one whose details cannot be loaded.
+        let corrupt_meta = IndexMetadata {
+            uuid: uuid::Uuid::new_v4(),
+            index_details: None,
+            ..frag_reuse_index_meta.clone()
+        };
+        let transaction = Transaction::new(
+            dataset.manifest.version,
+            Operation::CreateIndex {
+                new_indices: vec![corrupt_meta],
+                removed_indices: vec![frag_reuse_index_meta],
+            },
+            None,
+        );
+        dataset
+            .apply_commit(transaction, &Default::default(), &Default::default())
+            .await
+            .unwrap();
+
+        assert!(
+            cleanup_frag_reuse_index(&mut dataset).await.is_err(),
+            "corrupt frag-reuse metadata must surface as an error, not a panic"
+        );
+    }
+
     /// With more than one index on the table, remapping every index must catch
     /// all of them up so the reuse index can be trimmed.
     ///

From 01f809887af9ae9fd7191fd4a90b1e3a6abbef10 Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Sat, 27 Jun 2026 14:36:24 +0200
Subject: [PATCH 4/9] fix(frag-reuse): align cleanup generation retention with
 load path and bound metadata reads

---
 rust/lance/src/dataset/index/frag_reuse.rs | 131 ++++++++++++++++++---
 rust/lance/src/index/frag_reuse.rs         | 124 +++++++++++++++++--
 2 files changed, 233 insertions(+), 22 deletions(-)

diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index df5fe568e41..1c0e31a6a32 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -43,20 +43,8 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
 
     let mut retained_versions = Vec::new();
     let mut fragment_bitmaps = RoaringBitmap::new();
-    'versions: for version in frag_reuse_details.versions.iter() {
-        let mut all_caught_up = true;
-        for idx in indices.iter() {
-            match is_index_remap_caught_up(version, idx) {
-                Ok(true) => {}
-                Ok(false) => all_caught_up = false,
-                // An InvalidInput error means the reuse version is likely corrupted; drop it.
-                Err(Error::InvalidInput { .. }) => continue 'versions,
-                // Any other error is unexpected; surface it instead of panicking.
-                Err(e) => return Err(e),
-            }
-        }
-
-        if !all_caught_up {
+    for version in frag_reuse_details.versions.iter() {
+        if !is_reuse_version_caught_up(version, &indices)? {
             fragment_bitmaps.extend(version.new_frag_bitmap());
             retained_versions.push(version.clone());
         }
@@ -95,6 +83,33 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
     Ok(())
 }
 
+/// Decide whether a reuse version can be safely cleaned up given the current indices.
+///
+/// A reuse version is only droppable once *every* index has caught up to it. This
+/// mirrors the load/remap path: `recalculate_fragment_bitmap` treats a rewrite group
+/// that straddles indexed and non-indexed fragments as a recoverable state that aborts
+/// the remap (preserving the reuse generation) rather than as corruption to discard.
+/// We therefore treat the same `InvalidInput` straddling state as "not caught up" so the
+/// reuse generation is retained — dropping it would leave the affected indexes unable to
+/// remap their stale fragments. Any other (unexpected) error is surfaced to the caller.
+fn is_reuse_version_caught_up(
+    frag_reuse_version: &FragReuseVersion,
+    indices: &[IndexMetadata],
+) -> lance_core::Result<bool> {
+    for idx in indices.iter() {
+        match is_index_remap_caught_up(frag_reuse_version, idx) {
+            Ok(true) => {}
+            // The index is behind (or straddles indexed/non-indexed data); the reuse
+            // version is still needed and must be retained, exactly as the load path
+            // would preserve it.
+            Ok(false) | Err(Error::InvalidInput { .. }) => return Ok(false),
+            // Any other error is unexpected; surface it instead of panicking.
+            Err(e) => return Err(e),
+        }
+    }
+    Ok(true)
+}
+
 fn is_index_remap_caught_up(
     frag_reuse_version: &FragReuseVersion,
     index_meta: &IndexMetadata,
@@ -153,6 +168,7 @@ mod tests {
     use arrow_array::types::{Float32Type, Int32Type};
     use lance_datagen::Dimension;
     use lance_index::IndexType;
+    use lance_index::frag_reuse::{FragDigest, FragReuseGroup};
     use lance_index::scalar::ScalarIndexParams;
 
     #[tokio::test]
@@ -323,6 +339,93 @@ mod tests {
         );
     }
 
+    fn frag_digest(id: u64) -> FragDigest {
+        FragDigest {
+            id,
+            physical_rows: 1000,
+            num_deleted_rows: 0,
+        }
+    }
+
+    /// Build a reuse version with a single rewrite group rewriting `old_ids` into
+    /// `new_ids` at the given dataset version.
+    fn reuse_version(dataset_version: u64, old_ids: &[u64], new_ids: &[u64]) -> FragReuseVersion {
+        FragReuseVersion {
+            dataset_version,
+            groups: vec![FragReuseGroup {
+                changed_row_addrs: Vec::new(),
+                old_frags: old_ids.iter().copied().map(frag_digest).collect(),
+                new_frags: new_ids.iter().copied().map(frag_digest).collect(),
+            }],
+        }
+    }
+
+    fn index_meta_with_bitmap(dataset_version: u64, frag_ids: &[u32]) -> IndexMetadata {
+        IndexMetadata {
+            uuid: uuid::Uuid::new_v4(),
+            fields: vec![0],
+            name: "straddle_idx".into(),
+            dataset_version,
+            fragment_bitmap: Some(RoaringBitmap::from_iter(frag_ids.iter().copied())),
+            index_details: None,
+            index_version: 0,
+            created_at: None,
+            base_id: None,
+            files: None,
+        }
+    }
+
+    /// A rewrite group that straddles indexed and non-indexed fragments must NOT be
+    /// treated as corrupt-and-droppable by cleanup. The load/remap path
+    /// (`recalculate_fragment_bitmap`) aborts on this state and thereby *preserves*
+    /// the reuse generation, so cleanup must report the version as "not caught up"
+    /// (retain it). A buggy "drop on straddle" implementation would report it as
+    /// caught up and silently prune the still-needed generation.
+    #[test]
+    fn test_straddling_group_is_not_caught_up() {
+        // Reuse version at dataset version 5 rewrites old fragments [1, 2] -> [10].
+        let version = reuse_version(5, &[1, 2], &[10]);
+
+        // Index is at a *newer* dataset version (so the version short-circuit does NOT
+        // fire) but its bitmap only covers fragment 1 of the group — i.e. it straddles
+        // indexed (1) and non-indexed (2) data, which makes
+        // `is_index_remap_caught_up` return `Err(InvalidInput)`.
+        let index = index_meta_with_bitmap(7, &[1]);
+        assert!(
+            matches!(
+                is_index_remap_caught_up(&version, &index),
+                Err(Error::InvalidInput { .. })
+            ),
+            "a straddling group must produce InvalidInput at the per-index check"
+        );
+
+        // The retention decision must treat that straddle as "not caught up" so the
+        // reuse generation is retained, mirroring the load path.
+        assert!(
+            !is_reuse_version_caught_up(&version, std::slice::from_ref(&index)).unwrap(),
+            "cleanup must retain (not drop) a reuse version whose group straddles \
+             indexed and non-indexed fragments"
+        );
+    }
+
+    /// Sanity check the opposite direction: an index is "caught up" once it has
+    /// already remapped the group's old fragments away, i.e. its bitmap no longer
+    /// contains any of them (and it is at/after the reuse version). Such a reuse
+    /// version is droppable.
+    #[test]
+    fn test_remapped_group_is_caught_up() {
+        let version = reuse_version(5, &[1, 2], &[10]);
+        // Bitmap points at the new fragment only; the old fragments [1, 2] have
+        // already been remapped away, so this index is caught up.
+        let index = index_meta_with_bitmap(7, &[10]);
+        assert!(is_reuse_version_caught_up(&version, std::slice::from_ref(&index)).unwrap());
+
+        // An index still covering all old fragments has NOT remapped yet, so the
+        // version is still needed (not caught up).
+        let stale_index = index_meta_with_bitmap(7, &[1, 2]);
+        assert!(!is_reuse_version_caught_up(&version, std::slice::from_ref(&stale_index)).unwrap());
+    }
+
     /// With more than one index on the table, remapping every index must catch
     /// all of them up so the reuse index can be trimmed.
     ///
diff --git a/rust/lance/src/index/frag_reuse.rs b/rust/lance/src/index/frag_reuse.rs
index 23a8fec5145..77c226538f7 100644
--- a/rust/lance/src/index/frag_reuse.rs
+++ b/rust/lance/src/index/frag_reuse.rs
@@ -50,16 +50,41 @@ pub async fn load_frag_reuse_index_details(
                 .join(index.uuid.to_string())
                 .join(external_file.path.clone());
 
+            // Use checked arithmetic and bounds validation so that malformed
+            // metadata surfaces as a proper `Error` (and therefore a Python
+            // exception) instead of panicking or building an invalid range.
+            let offset = usize::try_from(external_file.offset).map_err(|_| {
+                Error::index(format!(
+                    "Fragment reuse external file offset {} does not fit in usize",
+                    external_file.offset
+                ))
+            })?;
+            let size = usize::try_from(external_file.size).map_err(|_| {
+                Error::index(format!(
+                    "Fragment reuse external file size {} does not fit in usize",
+                    external_file.size
+                ))
+            })?;
+            let end = offset.checked_add(size).ok_or_else(|| {
+                Error::index(format!(
+                    "Fragment reuse external file range overflows: offset {offset} + size {size}"
+                ))
+            })?;
+
             // the file content will be cached in the index cache later
             // so we do not put it to the file cache
-            let range = external_file.offset as usize
-                ..(external_file.offset as usize + external_file.size as usize);
-            let data = dataset
-                .object_store
-                .open(&file_path)
-                .await?
-                .get_range(range)
-                .await?;
+            let reader = dataset.object_store.open(&file_path).await?;
+            let file_size = reader.size().await.map_err(|e| {
+                Error::index(format!(
+                    "Failed to read size of fragment reuse external file {file_path}: {e}"
+                ))
+            })?;
+            if end > file_size {
+                return Err(Error::index(format!(
+                    "Fragment reuse external file range {offset}..{end} is out of bounds for file {file_path} of size {file_size}"
+                )));
+            }
+            let data = reader.get_range(offset..end).await?;
 
             let pb_sequence = InlineContent::decode(data)?;
             Ok(Arc::new(FragReuseIndexDetails::try_from(pb_sequence)?))
@@ -176,3 +201,86 @@ pub(crate) async fn build_frag_reuse_index_metadata(
         files: None,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::utils::test::{DatagenExt, FragmentCount, FragmentRowCount};
+    use arrow_array::types::Int32Type;
+
+    /// Build index metadata pointing at an external details file with the given
+    /// `offset`/`size`, so we can exercise the offset/size arithmetic in
+    /// `load_frag_reuse_index_details` with crafted (potentially malformed) values.
+    fn external_details_meta(uuid: Uuid, offset: u64, size: u64) -> IndexMetadata {
+        let proto = FragmentReuseIndexDetails {
+            content: Some(Content::External(ExternalFile {
+                path: FRAG_REUSE_DETAILS_FILE_NAME.to_owned(),
+                offset,
+                size,
+            })),
+        };
+        IndexMetadata {
+            uuid,
+            name: FRAG_REUSE_INDEX_NAME.to_string(),
+            fields: vec![],
+            dataset_version: 0,
+            fragment_bitmap: Some(RoaringBitmap::new()),
+            index_details: Some(Arc::new(prost_types::Any::from_msg(&proto).unwrap())),
+            index_version: 0,
+            created_at: None,
+            base_id: None,
+            files: None,
+        }
+    }
+
+    /// Malformed external metadata (offset/size pointing past the end of the file,
+    /// or arithmetic that overflows `usize`) must surface as a normal `Error`
+    /// instead of panicking or building an invalid object-store range.
+    #[tokio::test]
+    async fn test_load_external_details_malformed_offset_size_errors() {
+        let dataset = lance_datagen::gen_batch()
+            .col("i", lance_datagen::array::step::<Int32Type>())
+            .into_ram_dataset(FragmentCount::from(1), FragmentRowCount::from(10))
+            .await
+            .unwrap();
+
+        // Write a small, valid-looking external details file.
+        let uuid = Uuid::new_v4();
+        let file_path = dataset
+            .indices_dir()
+            .join(uuid.to_string())
+            .join(FRAG_REUSE_DETAILS_FILE_NAME);
+        let payload = vec![0u8; 16];
+        let mut writer = dataset.object_store.create(&file_path).await.unwrap();
+        writer.write_all(&payload).await.unwrap();
+        writer.shutdown().await.unwrap();
+
+        // Out-of-bounds range (offset + size > file length) must error, not panic.
+        let meta = external_details_meta(uuid, 0, payload.len() as u64 + 1);
+        let err = load_frag_reuse_index_details(&dataset, &meta)
+            .await
+            .expect_err("out-of-bounds external range must error");
+        assert!(
+            err.to_string().contains("out of bounds"),
+            "expected an out-of-bounds error, got: {err}"
+        );
+
+        // Offset past the end of the file must error too.
+        let meta = external_details_meta(uuid, payload.len() as u64 + 100, 1);
+        assert!(
+            load_frag_reuse_index_details(&dataset, &meta)
+                .await
+                .is_err(),
+            "offset past end of file must error"
+        );
+
+        // Overflowing offset + size must error rather than panic on add.
+        let meta = external_details_meta(uuid, u64::MAX, u64::MAX);
+        assert!(
+            load_frag_reuse_index_details(&dataset, &meta)
+                .await
+                .is_err(),
+            "overflowing offset + size must error"
+        );
+    }
+}

From 53ac7156b044264981058787e725bd78595b700f Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:12:22 +0300
Subject: [PATCH 5/9] test(frag-reuse): assert error variant and message in
 cleanup tests

---
 rust/lance/src/dataset/index/frag_reuse.rs | 11 ++++-
 rust/lance/src/index/frag_reuse.rs         | 50 +++++++++++-----------
 2 files changed, 33 insertions(+), 28 deletions(-)

diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index 1c0e31a6a32..6d82e33f67c 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -333,9 +333,16 @@ mod tests {
             .await
             .unwrap();
 
+        let err = cleanup_frag_reuse_index(&mut dataset)
+            .await
+            .expect_err("corrupt frag-reuse metadata must surface as an error, not a panic");
+        assert!(
+            matches!(err, Error::Index { .. }),
+            "expected an Error::Index for corrupt frag-reuse metadata, got: {err}"
+        );
         assert!(
-            cleanup_frag_reuse_index(&mut dataset).await.is_err(),
-            "corrupt frag-reuse metadata must surface as an error, not a panic"
+            err.to_string().contains("fragment reuse index"),
+            "error message should identify the fragment reuse index, got: {err}"
         );
     }
 
diff --git a/rust/lance/src/index/frag_reuse.rs b/rust/lance/src/index/frag_reuse.rs
index 77c226538f7..000f4e2751d 100644
--- a/rust/lance/src/index/frag_reuse.rs
+++ b/rust/lance/src/index/frag_reuse.rs
@@ -207,6 +207,10 @@ mod tests {
     use super::*;
     use crate::utils::test::{DatagenExt, FragmentCount, FragmentRowCount};
     use arrow_array::types::Int32Type;
+    use rstest::rstest;
+
+    /// Size of the valid external details file written by the malformed-input test.
+    const PAYLOAD_LEN: usize = 16;
 
     /// Build index metadata pointing at an external details file with the given
     /// `offset`/`size`, so we can exercise the offset/size arithmetic in
@@ -233,11 +237,20 @@ mod tests {
         }
     }
 
-    /// Malformed external metadata (offset/size pointing past the end of the file,
-    /// or arithmetic that overflows `usize`) must surface as a normal `Error`
-    /// instead of panicking or building an invalid object-store range.
+    /// Malformed external metadata must surface as a normal Error instead of
+    /// panicking or building an invalid object-store range. Each case crafts a
+    /// different bad offset/size against a valid PAYLOAD_LEN-byte details file
+    /// and asserts both the error variant and an identifying message fragment.
+    #[rstest]
+    #[case::out_of_bounds(0, PAYLOAD_LEN as u64 + 1, "out of bounds")]
+    #[case::offset_past_eof(PAYLOAD_LEN as u64 + 100, 1, "out of bounds")]
+    #[case::overflow(u64::MAX, u64::MAX, "overflow")]
     #[tokio::test]
-    async fn test_load_external_details_malformed_offset_size_errors() {
+    async fn test_load_external_details_malformed_offset_size_errors(
+        #[case] offset: u64,
+        #[case] size: u64,
+        #[case] expected_msg: &str,
+    ) {
         let dataset = lance_datagen::gen_batch()
             .col("i", lance_datagen::array::step::<Int32Type>())
             .into_ram_dataset(FragmentCount::from(1), FragmentRowCount::from(10))
@@ -250,37 +263,22 @@ mod tests {
             .indices_dir()
             .join(uuid.to_string())
             .join(FRAG_REUSE_DETAILS_FILE_NAME);
-        let payload = vec![0u8; 16];
+        let payload = vec![0u8; PAYLOAD_LEN];
         let mut writer = dataset.object_store.create(&file_path).await.unwrap();
         writer.write_all(&payload).await.unwrap();
         writer.shutdown().await.unwrap();
 
-        // Out-of-bounds range (offset + size > file length) must error, not panic.
-        let meta = external_details_meta(uuid, 0, payload.len() as u64 + 1);
+        let meta = external_details_meta(uuid, offset, size);
         let err = load_frag_reuse_index_details(&dataset, &meta)
             .await
-            .expect_err("out-of-bounds external range must error");
+            .expect_err("malformed external metadata must surface as an error, not a panic");
         assert!(
-            err.to_string().contains("out of bounds"),
-            "expected an out-of-bounds error, got: {err}"
+            matches!(err, Error::Index { .. }),
+            "expected an Error::Index, got: {err:?}"
         );
-
-        // Offset past the end of the file must error too.
-        let meta = external_details_meta(uuid, payload.len() as u64 + 100, 1);
-        assert!(
-            load_frag_reuse_index_details(&dataset, &meta)
-                .await
-                .is_err(),
-            "offset past end of file must error"
-        );
-
-        // Overflowing offset + size must error rather than panic on add.
-        let meta = external_details_meta(uuid, u64::MAX, u64::MAX);
         assert!(
-            load_frag_reuse_index_details(&dataset, &meta)
-                .await
-                .is_err(),
-            "overflowing offset + size must error"
+            err.to_string().to_lowercase().contains(expected_msg),
+            "expected error message to contain {expected_msg:?}, got: {err}"
         );
     }
 }

From b26e9ac000e40679bee5060045ae7d34c19b11e6 Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:15:08 +0300
Subject: [PATCH 6/9] simplified documentation

---
 rust/lance/src/dataset/index/frag_reuse.rs | 39 +++++++++++-----------
 1 file changed, 19 insertions(+), 20 deletions(-)

diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index 6d82e33f67c..74ba4d2cf52 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -85,13 +85,13 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
 
 /// Decide whether a reuse version can be safely cleaned up given the current indices.
 ///
-/// A reuse version is only droppable once *every* index has caught up to it. This
-/// mirrors the load/remap path: `recalculate_fragment_bitmap` treats a rewrite group
+/// A reuse version is only droppable once every index has caught up to it. This
+/// mirrors the load/remap path. `recalculate_fragment_bitmap` treats a rewrite group
 /// that straddles indexed and non-indexed fragments as a recoverable state that aborts
-/// the remap (preserving the reuse generation) rather than as corruption to discard.
-/// We therefore treat the same `InvalidInput` straddling state as "not caught up" so the
-/// reuse generation is retained — dropping it would leave the affected indexes unable to
-/// remap their stale fragments. Any other (unexpected) error is surfaced to the caller.
+/// the remap and preserves the reuse generation, rather than as corruption to discard.
+/// We treat the same `InvalidInput` straddling state as "not caught up" so the reuse
+/// generation is kept. Dropping it would leave the affected indexes unable to remap
+/// their stale fragments. Any other error is surfaced to the caller.
 fn is_reuse_version_caught_up(
     frag_reuse_version: &FragReuseVersion,
     indices: &[IndexMetadata],
@@ -99,11 +99,10 @@ fn is_reuse_version_caught_up(
     for idx in indices.iter() {
         match is_index_remap_caught_up(frag_reuse_version, idx) {
             Ok(true) => {}
-            // The index is behind (or straddles indexed/non-indexed data); the reuse
-            // version is still needed and must be retained, exactly as the load path
-            // would preserve it.
+            // The index is behind or straddles indexed and non-indexed data, so the
+            // reuse version is still needed and must be kept, as the load path does.
             Ok(false) | Err(Error::InvalidInput { .. }) => return Ok(false),
-            // Any other error is unexpected; surface it instead of panicking.
+            // Any other error is unexpected, so surface it instead of panicking.
             Err(e) => return Err(e),
         }
     }
@@ -382,19 +381,19 @@ mod tests {
         }
     }
 
-    /// A rewrite group that straddles indexed and non-indexed fragments must NOT be
+    /// A rewrite group that straddles indexed and non-indexed fragments must not be
     /// treated as corrupt-and-droppable by cleanup. The load/remap path
-    /// (`recalculate_fragment_bitmap`) aborts on this state and thereby *preserves*
-    /// the reuse generation, so cleanup must report the version as "not caught up"
-    /// (retain it). A buggy "drop on straddle" implementation would report it as
-    /// caught up and silently prune the still-needed generation.
+    /// (`recalculate_fragment_bitmap`) aborts on this state and preserves the reuse
+    /// generation, so cleanup must report the version as "not caught up" and retain
+    /// it. A buggy "drop on straddle" implementation would report it as caught up and
+    /// silently prune the still-needed generation.
     #[test]
     fn test_straddling_group_is_not_caught_up() {
         // Reuse version at dataset version 5 rewrites old fragments [1, 2] -> [10].
         let version = reuse_version(5, &[1, 2], &[10]);
 
-        // Index is at a *newer* dataset version (so the version short-circuit does NOT
-        // fire) but its bitmap only covers fragment 1 of the group — i.e. it straddles
+        // Index is at a newer dataset version so the version short-circuit does not
+        // fire, but its bitmap only covers fragment 1 of the group. It straddles
         // indexed (1) and non-indexed (2) data, which makes
         // `is_index_remap_caught_up` return `Err(InvalidInput)`.
         let index = index_meta_with_bitmap(7, &[1]);
@@ -422,13 +421,13 @@ mod tests {
     #[test]
     fn test_remapped_group_is_caught_up() {
         let version = reuse_version(5, &[1, 2], &[10]);
-        // Bitmap points at the new fragment only; the old fragments [1, 2] have
+        // Bitmap points at the new fragment only. The old fragments [1, 2] have
         // already been remapped away, so this index is caught up.
         let index = index_meta_with_bitmap(7, &[10]);
         assert!(is_reuse_version_caught_up(&version, std::slice::from_ref(&index)).unwrap());
 
-        // An index still covering all old fragments has NOT remapped yet, so the
-        // version is still needed (not caught up).
+        // An index still covering all old fragments has not remapped yet, so the
+        // version is still needed and not caught up.
         let stale_index = index_meta_with_bitmap(7, &[1, 2]);
         assert!(!is_reuse_version_caught_up(&version, std::slice::from_ref(&stale_index)).unwrap());
     }

From 1ec8936160d6c7e81bdb897808fb0fac727fa1c1 Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:47:35 +0300
Subject: [PATCH 7/9] perf(frag-reuse): pre-allocate retained-version buffer in
 cleanup

---
 rust/lance/src/dataset/index/frag_reuse.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/rust/lance/src/dataset/index/frag_reuse.rs b/rust/lance/src/dataset/index/frag_reuse.rs
index 74ba4d2cf52..975d11f054c 100644
--- a/rust/lance/src/dataset/index/frag_reuse.rs
+++ b/rust/lance/src/dataset/index/frag_reuse.rs
@@ -41,7 +41,7 @@ pub async fn cleanup_frag_reuse_index(dataset: &mut Dataset) -> lance_core::Resu
     // Surface corrupt/stale metadata as a normal error instead of panicking on unwrap.
     let frag_reuse_details = load_frag_reuse_index_details(dataset, frag_reuse_index_meta).await?;
 
-    let mut retained_versions = Vec::new();
+    let mut retained_versions = Vec::with_capacity(frag_reuse_details.versions.len());
     let mut fragment_bitmaps = RoaringBitmap::new();
     for version in frag_reuse_details.versions.iter() {
         if !is_reuse_version_caught_up(version, &indices)? {

From c63a7cba8176eaedfd50af4a46c80ea690930b55 Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Sat, 18 Jul 2026 19:26:47 +0200
Subject: [PATCH 8/9] fix(python): validate batch readahead type

---
 python/python/lance/dataset.py      |  9 ++++++---
 python/python/tests/test_dataset.py | 10 ++++++++++
 2 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py
index 7077d2d6972..348689554ee 100644
--- a/python/python/lance/dataset.py
+++ b/python/python/lance/dataset.py
@@ -6283,10 +6283,13 @@ def batch_readahead(self, nbatches: Optional[int] = None) -> ScannerBuilder:
         """
         Set the maximum number of batches to decode concurrently.
 
-        This parameter must be greater than zero.
+        This parameter must be a positive integer.
         """
-        if nbatches is not None and int(nbatches) <= 0:
-            raise ValueError("batch_readahead must be greater than 0")
+        if nbatches is not None:
+            if not isinstance(nbatches, int) or isinstance(nbatches, bool):
+                raise ValueError("batch_readahead must be an integer")
+            if nbatches <= 0:
+                raise ValueError("batch_readahead must be greater than 0")
         self._batch_readahead = nbatches
         return self
 
diff --git a/python/python/tests/test_dataset.py b/python/python/tests/test_dataset.py
index 898bdf93069..d7cf998c4ff 100644
--- a/python/python/tests/test_dataset.py
+++ b/python/python/tests/test_dataset.py
@@ -1166,6 +1166,16 @@ def test_to_batches(tmp_path: Path):
     assert sorted_batches == table
 
 
+@pytest.mark.parametrize("batch_readahead", [1.5, "1", True])
+def test_batch_readahead_requires_integer(tmp_path: Path, batch_readahead):
+    table = pa.Table.from_pydict({"a": range(10)})
+    base_dir = tmp_path / "test"
+    lance.write_dataset(table, base_dir)
+
+    with pytest.raises(ValueError, match="batch_readahead must be an integer"):
+        lance.dataset(base_dir).scanner().batch_readahead(batch_readahead)
+
+
 def test_list_from_parquet(tmp_path: Path):
     # This is a regression for GH-1482, the parquet reader creates
     # list fields with the name 'element' instead of 'item'.  We should

From d89a6e04208606585b6798357c9bf13e7a27ee8b Mon Sep 17 00:00:00 2001
From: gstamatakis95 <126914070+gstamatakis95@users.noreply.github.com>
Date: Sat, 18 Jul 2026 19:41:13 +0200
Subject: [PATCH 9/9] test(python): exercise batch readahead validation

---
 python/python/tests/test_dataset.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/python/python/tests/test_dataset.py b/python/python/tests/test_dataset.py
index d7cf998c4ff..ddd0783a49a 100644
--- a/python/python/tests/test_dataset.py
+++ b/python/python/tests/test_dataset.py
@@ -1173,7 +1173,7 @@ def test_batch_readahead_requires_integer(tmp_path: Path, batch_readahead):
     lance.write_dataset(table, base_dir)
 
     with pytest.raises(ValueError, match="batch_readahead must be an integer"):
-        lance.dataset(base_dir).scanner().batch_readahead(batch_readahead)
+        lance.dataset(base_dir).scanner(batch_readahead=batch_readahead)
 
 
 def test_list_from_parquet(tmp_path: Path):