fix(python): expose optimize compaction options (#3870)

## Summary
- expose typed compaction options on synchronous and asynchronous table
optimization
- translate validated Python options into native Lance compaction
settings
- document the row-based steady-state limit and first-pass/source-budget
caveats
- carry the Lance beta.15 fragment-exclusion option through the typed
Python API

## Root cause
The native Python optimize binding always constructed default compaction
settings, and the public AsyncTable API offered no way to lower the
fragment row target. Wide-row fragments therefore remained compaction
candidates on every later optimize call, repeatedly rewriting the
accumulated table.

## Limitations
This is an explicit row-based mitigation, not byte-native task sizing or
a hard memory cap. Callers must derive target_rows_per_fragment from
expected row size. Existing oversized fragments are rewritten whole on
the first pass; max_bytes_per_file limits output file size rather than
compaction memory; and source budgets admit whole planned tasks, so a
budget below the first task performs no compaction. Splitting tasks by a
byte budget or returning a planner error requires a Lance-side change.

## Validation
- cargo check --release -p lancedb-python
- cargo clippy --quiet -p lancedb-python
- cargo fmt --all --check
- ruff format --check python tests
- ruff check python tests
- uv run --no-sync pytest python/tests/test_table.py -k
optimize_compaction -q (16 passed)

Fixes #2325

<!-- lance-gatekeeper-fix:v1 agent=9b58eee0c79040b19540e259af0e676c
generation=1 -->

---------

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
Co-authored-by: Xuanwo <github@xuanwo.io>

rust/lancedb/src/table.rs

@@ -107,7 +107,9 @@ use futures::future::join_all;
 pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags};
 pub use lance::dataset::scanner::DatasetRecordBatchStream;
 pub use lance_index::optimize::OptimizeOptions;
-pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
+pub use optimize::{
+    CompactionMode, CompactionOptions, IndexRemapMode, OptimizeAction, OptimizeStats,
+};
 pub use refresh::RefreshColumnResult;
 pub use rte_stats::{MemtableStats, RTEStats, SsTableStats, TableShardStats};
 pub use schema_evolution::{

rust/lancedb/src/table/optimize.rs

@@ -16,7 +16,7 @@ use lance_index::optimize::OptimizeOptions;
 use log::info;
 
 pub use chrono::Duration;
-pub use lance::dataset::optimize::CompactionOptions;
+pub use lance::dataset::optimize::{CompactionMode, CompactionOptions, IndexRemapMode};
 
 use super::NativeTable;
 use crate::error::Result;

rust/lancedb/tests/doc_storage_examples.rs

@@ -0,0 +1,76 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use std::ops::Range;
+use std::sync::Arc;
+
+use arrow_array::{
+    ArrayRef, BinaryArray, Int64Array, RecordBatch, RecordBatchIterator, RecordBatchReader,
+};
+
+fn wide_rows(ids: Range<i64>) -> anyhow::Result<Box<dyn RecordBatchReader + Send>> {
+    let payload = [b'x'; 1016];
+    let batch = RecordBatch::try_from_iter(vec![
+        (
+            "id",
+            Arc::new(Int64Array::from_iter_values(ids.clone())) as ArrayRef,
+        ),
+        (
+            "payload",
+            Arc::new(BinaryArray::from_iter_values(
+                ids.map(|_| payload.as_slice()),
+            )) as ArrayRef,
+        ),
+    ])?;
+    let schema = batch.schema();
+    Ok(Box::new(RecordBatchIterator::new(
+        std::iter::once(Ok(batch)),
+        schema,
+    )))
+}
+
+async fn tune_compaction(table: &lancedb::Table) -> lancedb::Result<()> {
+    // --8<-- [start:storage_compaction_tuning]
+    use lancedb::table::{CompactionOptions, OptimizeAction};
+
+    let desired_fragment_bytes: usize = 64 * 1024;
+    let average_row_bytes: usize = 1024; // Estimate from representative rows.
+    let options = CompactionOptions {
+        target_rows_per_fragment: (desired_fragment_bytes / average_row_bytes).max(1),
+        batch_size: Some(16),
+        num_threads: Some(1),
+        ..Default::default()
+    };
+    table
+        .optimize(OptimizeAction::Compact {
+            options,
+            remap_options: None,
+        })
+        .await?;
+    // --8<-- [end:storage_compaction_tuning]
+    Ok(())
+}
+
+#[tokio::test]
+async fn compaction_tuning() -> anyhow::Result<()> {
+    let directory = tempfile::tempdir()?;
+    let db = lancedb::connect(directory.path().to_str().unwrap())
+        .execute()
+        .await?;
+    let table = db
+        .create_table("compaction_tuning", wide_rows(0..32)?)
+        .execute()
+        .await?;
+    table.add(wide_rows(32..64)?).execute().await?;
+    assert_eq!(table.stats().await?.fragment_stats.num_fragments, 2);
+
+    tune_compaction(&table).await?;
+    assert_eq!(table.stats().await?.fragment_stats.num_fragments, 1);
+    assert_eq!(table.count_rows(None).await?, 64);
+
+    table.add(wide_rows(64..65)?).execute().await?;
+    tune_compaction(&table).await?;
+    assert_eq!(table.stats().await?.fragment_stats.num_fragments, 2);
+    assert_eq!(table.count_rows(None).await?, 65);
+    Ok(())
+}