fix(python): expose compaction storage version

python/python/lancedb/table.py

@@ -309,6 +309,9 @@ class CompactionOptions(TypedDict, total=False):
     max_overlays_per_fragment: Optional[int]
     """Maximum overlays before a fragment is fully compacted."""
 
+    data_storage_version: str
+    """Data-file version or release selector for compacted output."""
+
 
 # Known distance metrics for legacy API detection
 KNOWN_METRICS = {"l2", "cosine", "dot", "hamming"}

python/python/tests/test_table.py

@@ -4123,6 +4123,33 @@ async def test_optimize_compaction_source_limits(
     assert stats.compaction.fragments_added == 0
 
 
+@pytest.mark.asyncio
+async def test_optimize_compaction_data_storage_version(
+    mem_db_async: AsyncConnection,
+):
+    table = await mem_db_async.create_table("test", data=[{"x": 1}])
+    await table.add([{"x": 2}])
+
+    stats = await table.optimize(
+        compaction_options={
+            "target_rows_per_fragment": 3,
+            "data_storage_version": "2.2",
+        }
+    )
+    assert stats.compaction.fragments_removed == 2
+    assert stats.compaction.fragments_added == 1
+
+    table = await mem_db_async.create_table("cross_family", data=[{"x": 1}])
+    await table.add([{"x": 2}])
+    with pytest.raises(ValueError, match="V1 and V2 storage versions cannot be mixed"):
+        await table.optimize(
+            compaction_options={
+                "target_rows_per_fragment": 3,
+                "data_storage_version": "legacy",
+            }
+        )
+
+
 @pytest.mark.asyncio
 async def test_optimize_compaction_excluded_fragments(mem_db_async: AsyncConnection):
     table = await mem_db_async.create_table("test", data=[{"x": 1}])
@@ -4155,6 +4182,7 @@ async def test_optimize_compaction_excluded_fragments(mem_db_async: AsyncConnect
         ("io_buffer_size", 2**63, "must be at most 9223372036854775807"),
         ("max_source_rows", 0, "must be greater than 0"),
         ("max_source_bytes", 0, "must be greater than 0"),
+        ("data_storage_version", "invalid", "Unknown Lance storage version: invalid"),
         (
             "excluded_fragment_ids",
             [-1],

python/src/table.rs

@@ -18,6 +18,7 @@ use arrow::{
     pyarrow::{FromPyArrow, PyArrowType, ToPyArrow},
 };
 use lancedb::blob::{BlobFile, BlobRangeRequest};
+use lancedb::connection::LanceFileVersion;
 use lancedb::index::scalar::FtsIndexBuilder;
 use lancedb::table::{
     AddDataMode, ColumnAlteration, CompactionMode, CompactionOptions, Duration,
@@ -212,6 +213,14 @@ fn parse_compaction_options(options: Option<&Bound<'_, PyDict>>) -> PyResult<Com
             "max_source_bytes" => parsed.max_source_bytes = optional_positive_u64(&value, &key)?,
             "excluded_fragment_ids" => parsed.excluded_fragment_ids = u32_list(&value, &key)?,
             "max_overlays_per_fragment" => parsed.max_overlays_per_fragment = value.extract()?,
+            "data_storage_version" => {
+                let version: String = value.extract()?;
+                parsed.data_storage_version = Some(
+                    version
+                        .parse::<LanceFileVersion>()
+                        .map_err(|err| PyValueError::new_err(err.to_string()))?,
+                );
+            }
             _ => {
                 return Err(PyValueError::new_err(format!(
                     "Invalid compaction option: {key}"