diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index c364ab0d3..afa38dc96 100644 --- a/python/python/lancedb/table.py +++ b/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 ----- diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 4cdf39797..4e4450e13 100644 --- a/python/python/tests/test_table.py +++ b/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): diff --git a/python/src/table.rs b/python/src/table.rs index ffb20f377..ba86f7c3d 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -142,6 +142,21 @@ fn optional_positive_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult, name: &str) -> PyResult> { + value + .extract::>()? + .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> { let value: Option = 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 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!(