fix(python): expose compaction source limits

python/python/lancedb/table.py

@@ -277,6 +277,12 @@ class CompactionOptions(TypedDict, total=False):
     max_source_fragments: Optional[int]
     """Maximum number of source fragments compacted in one run."""
 
+    max_source_rows: Optional[int]
+    """Maximum number of live source rows compacted in one run."""
+
+    max_source_bytes: Optional[int]
+    """Maximum source data and overlay bytes compacted in one run."""
+
     max_overlays_per_fragment: Optional[int]
     """Maximum overlays before a fragment is fully compacted."""
 

python/python/tests/test_table.py

@@ -3691,6 +3691,24 @@ async def test_optimize_compaction_options(mem_db_async: AsyncConnection):
         await table.optimize(compaction_options={"unknown": 1})
 
 
+@pytest.mark.parametrize("option", ["max_source_rows", "max_source_bytes"])
+@pytest.mark.asyncio
+async def test_optimize_compaction_source_limits(
+    mem_db_async: AsyncConnection, option: str
+):
+    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,
+            option: 1,
+        }
+    )
+    assert stats.compaction.fragments_removed == 0
+    assert stats.compaction.fragments_added == 0
+
+
 @pytest.mark.parametrize(
     ("option", "value", "message"),
     [
@@ -3702,6 +3720,8 @@ async def test_optimize_compaction_options(mem_db_async: AsyncConnection):
         ("max_rows_per_group", 2**32, "must be between 1 and 4294967295"),
         ("batch_size", 2**32, "must be between 1 and 4294967295"),
         ("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"),
     ],
 )
 @pytest.mark.asyncio

python/src/table.rs

@@ -132,6 +132,16 @@ fn optional_positive_usize(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Opt
     Ok(value)
 }
 
+fn optional_positive_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<u64>> {
+    let value: Option<u64> = value.extract()?;
+    if value == Some(0) {
+        return Err(PyValueError::new_err(format!(
+            "{name} must be greater than 0"
+        )));
+    }
+    Ok(value)
+}
+
 fn optional_i64_bounded_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<u64>> {
     let value: Option<u64> = value.extract()?;
     if value.is_some_and(|value| value > i64::MAX as u64) {
@@ -183,6 +193,8 @@ fn parse_compaction_options(options: Option<&Bound<'_, PyDict>>) -> PyResult<Com
                 parsed.binary_copy_read_batch_bytes = value.extract()?
             }
             "max_source_fragments" => parsed.max_source_fragments = value.extract()?,
+            "max_source_rows" => parsed.max_source_rows = optional_positive_usize(&value, &key)?,
+            "max_source_bytes" => parsed.max_source_bytes = optional_positive_u64(&value, &key)?,
             "max_overlays_per_fragment" => parsed.max_overlays_per_fragment = value.extract()?,
             _ => {
                 return Err(PyValueError::new_err(format!(