fix(python): clarify compaction limits

python/python/lancedb/table.py

@@ -224,26 +224,36 @@ class CompactionOptions(TypedDict, total=False):
 
     Unspecified options use Lance's defaults.
 
+    Compaction planning is row based. Lowering ``target_rows_per_fragment``
+    based on the expected row size can bound later compaction passes once
+    oversized fragments have been rewritten. It does not split an existing
+    fragment, so the first pass over an oversized fragment is not subject to
+    that bound. ``max_bytes_per_file`` limits output file size, not compaction
+    memory. Source budgets keep whole planned tasks; if the first task exceeds
+    a budget, that run performs no compaction work.
+
     Examples
     --------
-    Limit fragment size and compaction concurrency for tables with large rows:
+    Derive a steady-state row target from the expected row size:
 
+    >>> desired_fragment_bytes = 750 * 1024 * 1024
+    >>> average_row_bytes = 1_500_000
     >>> options: CompactionOptions = {
-    ...     "target_rows_per_fragment": 500,
-    ...     "batch_size": 1024,
-    ...     "num_threads": 1,
+    ...     "target_rows_per_fragment": max(
+    ...         1, desired_fragment_bytes // average_row_bytes
+    ...     ),
     ... }
     >>> await table.optimize(compaction_options=options)  # doctest: +SKIP
     """
 
     target_rows_per_fragment: int
-    """Target number of rows per fragment (default: 1,048,576)."""
+    """Target rows per fragment; existing oversized fragments are not split."""
 
     max_rows_per_group: int
     """Maximum number of rows per row group (default: 1,024)."""
 
     max_bytes_per_file: Optional[int]
-    """Maximum number of bytes per data file."""
+    """Maximum output data-file size; this does not bound compaction memory."""
 
     materialize_deletions: bool
     """Whether to rewrite fragments containing deleted rows (default: True)."""
@@ -278,10 +288,13 @@ class CompactionOptions(TypedDict, total=False):
     """Maximum number of source fragments compacted in one run."""
 
     max_source_rows: Optional[int]
-    """Maximum number of live source rows compacted in one run."""
+    """Maximum live source rows per run, applied to whole planned tasks."""
 
     max_source_bytes: Optional[int]
-    """Maximum source data and overlay bytes compacted in one run."""
+    """Maximum source bytes per run, applied to whole planned tasks."""
+
+    excluded_fragment_ids: List[int]
+    """Fragment IDs to leave unchanged and use as planning boundaries."""
 
     max_overlays_per_fragment: Optional[int]
     """Maximum overlays before a fragment is fully compacted."""
@@ -1936,9 +1949,10 @@ class Table(ABC):
         retrain: bool, default False
             This parameter is no longer used and is deprecated.
         compaction_options: CompactionOptions, optional
-            Options that control file compaction. This can be used to bound
-            memory usage by reducing ``target_rows_per_fragment``, ``batch_size``,
-            or ``num_threads``.
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].
 
         Notes
         -----
@@ -4048,9 +4062,10 @@ class LanceTable(Table):
         retrain: bool, default False
             This parameter is no longer used and is deprecated.
         compaction_options: CompactionOptions, optional
-            Options that control file compaction. This can be used to bound
-            memory usage by reducing ``target_rows_per_fragment``, ``batch_size``,
-            or ``num_threads``.
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].
 
         Notes
         -----
@@ -6440,9 +6455,10 @@ class AsyncTable:
         retrain: bool, default False
             This parameter is no longer used and is deprecated.
         compaction_options: CompactionOptions, optional
-            Options that control file compaction. This can be used to bound
-            memory usage by reducing ``target_rows_per_fragment``, ``batch_size``,
-            or ``num_threads``.
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].
 
         Notes
         -----

python/python/tests/test_table.py

@@ -12,7 +12,7 @@ import weakref
 from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta
 from time import sleep
-from typing import List
+from typing import Any, List
 from unittest.mock import patch
 
 import lancedb
@@ -3709,6 +3709,25 @@ async def test_optimize_compaction_source_limits(
     assert stats.compaction.fragments_added == 0
 
 
+@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}])
+    await table.add([{"x": 2}])
+
+    stats = await table.optimize(
+        compaction_options={
+            "target_rows_per_fragment": 3,
+            "excluded_fragment_ids": [0],
+        }
+    )
+    assert stats.compaction.fragments_removed == 0
+    assert stats.compaction.fragments_added == 0
+
+    stats = await table.optimize(compaction_options={"target_rows_per_fragment": 3})
+    assert stats.compaction.fragments_removed == 2
+    assert stats.compaction.fragments_added == 1
+
+
 @pytest.mark.parametrize(
     ("option", "value", "message"),
     [
@@ -3722,11 +3741,21 @@ async def test_optimize_compaction_source_limits(
         ("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"),
+        (
+            "excluded_fragment_ids",
+            [-1],
+            "must contain values between 0 and 4294967295",
+        ),
+        (
+            "excluded_fragment_ids",
+            [2**32],
+            "must contain values between 0 and 4294967295",
+        ),
     ],
 )
 @pytest.mark.asyncio
 async def test_optimize_compaction_options_validation(
-    mem_db_async: AsyncConnection, option: str, value: int, message: str
+    mem_db_async: AsyncConnection, option: str, value: Any, message: str
 ):
     table = await mem_db_async.create_table("test", data=[{"x": 1}])
     with pytest.raises(ValueError, match=message):

python/src/table.rs

@@ -142,6 +142,21 @@ fn optional_positive_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Optio
     Ok(value)
 }
 
+fn u32_list(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Vec<u32>> {
+    value
+        .extract::<Vec<i64>>()?
+        .into_iter()
+        .map(|value| {
+            u32::try_from(value).map_err(|_| {
+                PyValueError::new_err(format!(
+                    "{name} must contain values between 0 and {}",
+                    u32::MAX
+                ))
+            })
+        })
+        .collect()
+}
+
 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) {
@@ -195,6 +210,7 @@ fn parse_compaction_options(options: Option<&Bound<'_, PyDict>>) -> PyResult<Com
             "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)?,
+            "excluded_fragment_ids" => parsed.excluded_fragment_ids = u32_list(&value, &key)?,
             "max_overlays_per_fragment" => parsed.max_overlays_per_fragment = value.extract()?,
             _ => {
                 return Err(PyValueError::new_err(format!(