feat: add lance_dataset_compact_files for fragment compaction (#40)

## Summary

Adds `lance_dataset_compact_files`, exposing upstream's `compact_files`
so callers can merge small fragments and materialize pending deletion
files in one pass. It's the operational follow-on to `_delete` /
`_update` / `_merge_insert`: heavy mutation users accumulate small
fragments and deletion files, and this is the cleanup step.

A clean dataset (no fragment under the target size, no deletions worth
materializing) is a no-op — the call succeeds with all-zero metrics and
the version is unchanged.

## Surface

`LanceCompactionOptions` carries the five common numeric knobs:

| field | upstream default |
|---|---|
| `target_rows_per_fragment` | ~1Mi rows |
| `max_rows_per_group` | 1024 |
| `max_bytes_per_file` | writer cap |
| `num_threads` | num compute-intensive CPUs |
| `batch_size` | scanner default |

Each uses `0` to mean "keep the upstream default", so `NULL` and a
zero-initialized struct behave identically. Non-zero values go through a
`usize` range check so the API doesn't silently truncate on 32-bit
hosts.

`LanceCompactionMetrics` mirrors upstream's four counts:
`fragments_removed`, `fragments_added`, `files_removed`, `files_added`.

The C++ surface is just `Dataset::compact_files(options=nullptr)`.

## Out of scope

The knobs that don't fit the 0-sentinel convention:
`materialize_deletions` (bool, default `true`),
`materialize_deletions_threshold` (f32, default `0.1`),
`compaction_mode` (3-variant enum), and the niche `defer_index_remap` /
`binary_copy_read_batch_bytes` / `max_source_fragments` /
`transaction_properties` fields. The struct is `repr(C)` and grows by
appending, so any of these can land in a later PR without breaking
existing callers. The `remap_options` argument to upstream is fixed at
`None` for now (uses the default remapper).

## Test plan

- 12 Rust integration tests: basic merge of small neighbors, data
round-trip, no-op on a clean single fragment, deletion-file
materialization after a partial-row delete on each fragment, each of the
five numeric overrides flowing through, zero-init/NULL equivalence,
optional `out_metrics`, NULL-dataset rejection, and untouched-on-error
semantics.
- C and C++ smoke tests in `compile_and_run_test` exercise the no-op
path on the existing single-fragment fixture; the C side also pins the
NULL-dataset rejection.
- `cargo fmt`, `cargo clippy --all-targets -- -D warnings`, the full
Rust test suite, and the C/C++ smoke compile are green locally.

Author: LuciferYang

include/lance/lance.h

@@ -390,6 +390,69 @@ int32_t lance_dataset_merge_insert(
     LanceMergeInsertResult* out_result
 );
 
+/* ─── lance_dataset_compact_files ─────────────────────────────────────────── */
+
+/**
+ * Tunable parameters for lance_dataset_compact_files. Pass NULL to use the
+ * upstream defaults. Each numeric field uses 0 as a "keep upstream default"
+ * sentinel; non-zero values are forwarded after a usize range check so the
+ * API does not silently truncate on 32-bit hosts.
+ */
+typedef struct LanceCompactionOptions {
+    /* Target row count per output fragment. Fragments below this size are
+       candidates for being merged with neighbors. 0 = default (~1Mi rows). */
+    uint64_t target_rows_per_fragment;
+    /* Soft cap on rows per row group within an output fragment. 0 = default. */
+    uint64_t max_rows_per_group;
+    /* Soft cap on bytes per output fragment file. 0 = default (writer cap). */
+    uint64_t max_bytes_per_file;
+    /* Compute parallelism for compaction tasks. 0 = default
+       (number of compute-intensive CPUs). */
+    uint64_t num_threads;
+    /* Scanner batch size for reading input fragments. 0 = default. */
+    uint64_t batch_size;
+} LanceCompactionOptions;
+
+/** Per-call compaction metrics returned via the optional out parameter. */
+typedef struct LanceCompactionMetrics {
+    /* Number of input fragments that were rewritten and dropped. */
+    uint64_t fragments_removed;
+    /* Number of new fragments produced by the rewrite. */
+    uint64_t fragments_added;
+    /* Total files removed across the operation, including deletion files. */
+    uint64_t files_removed;
+    /* Total files added across the operation; one per new fragment. */
+    uint64_t files_added;
+} LanceCompactionMetrics;
+
+/**
+ * Compact the dataset's fragments, committing a new manifest if anything
+ * changed. Each compaction task merges adjacent small fragments and
+ * materializes any deletion files in the process. A clean dataset (no
+ * fragment under the target size, no deletions worth materializing) is a
+ * no-op: the function returns success with all-zero metrics and the
+ * dataset's version is unchanged.
+ *
+ * Mutates `dataset` in place — the same handle remains valid afterward and
+ * sees the new version. Scanners already in flight against this dataset
+ * keep their pre-compaction snapshot view.
+ *
+ * @param dataset      Open dataset (not consumed). Must not be NULL.
+ * @param options      Tunable parameters. Pass NULL for upstream defaults.
+ * @param out_metrics  Optional. If non-NULL, on success receives the per-call
+ *                     compaction metrics. On error the slot is left unchanged
+ *                     — do not read it.
+ * @return 0 on success, -1 on error. Error codes:
+ *         LANCE_ERR_INVALID_ARGUMENT for NULL `dataset` or for numeric
+ *         overrides that exceed usize::MAX on the running target;
+ *         LANCE_ERR_COMMIT_CONFLICT for a concurrent writer.
+ */
+int32_t lance_dataset_compact_files(
+    LanceDataset* dataset,
+    const LanceCompactionOptions* options,
+    LanceCompactionMetrics* out_metrics
+);
+
 /**
  * Export the dataset schema via Arrow C Data Interface.
  * @param out  Pointer to caller-allocated ArrowSchema struct

include/lance/lance.hpp

@@ -423,6 +423,19 @@ class Dataset {
         return merge_insert(on_columns, source, &params);
     }
 
+    /// Compact small or deleted-heavy fragments into larger ones, committing
+    /// a new manifest. A clean dataset is a no-op — all-zero metrics and the
+    /// version is unchanged. Pass `nullptr` for upstream defaults.
+    /// Throws lance::Error on failure (commit conflict, ...).
+    LanceCompactionMetrics compact_files(
+        const LanceCompactionOptions* options = nullptr) {
+        LanceCompactionMetrics metrics{};
+        if (lance_dataset_compact_files(handle_.get(), options, &metrics) != 0) {
+            check_error();
+        }
+        return metrics;
+    }
+
     /// Export the schema as an Arrow C Data Interface struct.
     void schema(ArrowSchema* out) const {
         if (lance_dataset_schema(handle_.get(), out) != 0) {

src/compact.rs

@@ -0,0 +1,169 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Compaction C API: rewrite small/deleted-heavy fragments into larger ones,
+//! committing a new manifest. Operates as a no-op (no version bump) when no
+//! fragments need compacting.
+//!
+//! Mutates the dataset in place under an exclusive write lock; existing
+//! scanners that already cloned the inner Arc keep their pre-compaction
+//! snapshot view.
+
+use lance::dataset::optimize::{CompactionOptions, compact_files};
+use lance_core::Result;
+
+use crate::dataset::LanceDataset;
+use crate::error::ffi_try;
+use crate::runtime::block_on;
+
+/// Tunable parameters for `lance_dataset_compact_files`. Pass NULL to use the
+/// upstream defaults. Each numeric field uses `0` as a "keep upstream default"
+/// sentinel; explicit overrides are forwarded after a `usize` range check.
+///
+/// The struct is `#[repr(C)]` and ABI-stable within a minor version; new
+/// fields can be appended without breaking existing callers.
+#[repr(C)]
+pub struct LanceCompactionOptions {
+    /// Target row count per output fragment. Fragments below this size are
+    /// candidates for being merged with neighbors. `0` uses upstream's
+    /// default (~1Mi rows).
+    pub target_rows_per_fragment: u64,
+    /// Soft cap on rows per row group within an output fragment. `0` uses
+    /// upstream's default.
+    pub max_rows_per_group: u64,
+    /// Soft cap on bytes per output fragment file. `0` uses upstream's
+    /// default (the writer's per-file cap).
+    pub max_bytes_per_file: u64,
+    /// Compute parallelism for compaction tasks. `0` uses upstream's default
+    /// (the number of compute-intensive CPUs).
+    pub num_threads: u64,
+    /// Scanner batch size for reading input fragments. `0` uses upstream's
+    /// default.
+    pub batch_size: u64,
+}
+
+/// Per-call compaction metrics returned via the optional out parameter.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct LanceCompactionMetrics {
+    /// Number of input fragments that were rewritten and dropped.
+    pub fragments_removed: u64,
+    /// Number of new fragments produced by the rewrite.
+    pub fragments_added: u64,
+    /// Total files removed across the operation, including deletion files.
+    pub files_removed: u64,
+    /// Total files added across the operation; one per new fragment.
+    pub files_added: u64,
+}
+
+/// Compact the dataset's fragments, committing a new manifest if anything
+/// changed.
+///
+/// Each compaction task merges adjacent small fragments and materializes any
+/// deletion files in the process. A clean dataset (no fragment under the
+/// target size, no deletions worth materializing) is a no-op: the function
+/// returns success with all-zero metrics and the dataset's version is
+/// unchanged.
+///
+/// - `dataset`: Open dataset (mutated; same handle remains valid afterward).
+///   Must not be NULL.
+/// - `options`: Optional. NULL uses upstream defaults; otherwise each field
+///   is treated as an override (`0` keeps the default for that field).
+/// - `out_metrics`: Optional. If non-NULL, on success receives the
+///   `LanceCompactionMetrics` for this call. On error the slot is untouched.
+///
+/// Returns 0 on success, -1 on error. Error codes:
+/// `LANCE_ERR_INVALID_ARGUMENT` for NULL `dataset` or for numeric overrides
+/// that exceed `usize::MAX` on the running target;
+/// `LANCE_ERR_COMMIT_CONFLICT` for a concurrent writer.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lance_dataset_compact_files(
+    dataset: *mut LanceDataset,
+    options: *const LanceCompactionOptions,
+    out_metrics: *mut LanceCompactionMetrics,
+) -> i32 {
+    ffi_try!(unsafe { compact_inner(dataset, options, out_metrics) }, neg)
+}
+
+unsafe fn compact_inner(
+    dataset: *mut LanceDataset,
+    options: *const LanceCompactionOptions,
+    out_metrics: *mut LanceCompactionMetrics,
+) -> Result<i32> {
+    if dataset.is_null() {
+        return Err(lance_core::Error::InvalidInput {
+            source: "dataset must not be NULL".into(),
+            location: snafu::location!(),
+        });
+    }
+
+    // SAFETY: `options` is either NULL (use defaults) or points to a valid
+    // `LanceCompactionOptions` for the duration of this call. `resolve` only
+    // reads through the pointer.
+    let resolved = unsafe { resolve_options(options)? };
+
+    // SAFETY: `dataset` is non-NULL (checked above) and the caller guarantees
+    // it points to a live `LanceDataset` not aliased mutably elsewhere.
+    let ds = unsafe { &*dataset };
+    let metrics = ds.with_mut(|d| block_on(compact_files(d, resolved, None)))?;
+
+    if !out_metrics.is_null() {
+        // SAFETY: caller guarantees `out_metrics` (when non-NULL) points to
+        // caller-owned, writable storage of size `sizeof(LanceCompactionMetrics)`.
+        // We only write on success; on the error paths above the slot stays
+        // untouched per the documented contract.
+        unsafe {
+            *out_metrics = LanceCompactionMetrics {
+                fragments_removed: metrics.fragments_removed as u64,
+                fragments_added: metrics.fragments_added as u64,
+                files_removed: metrics.files_removed as u64,
+                files_added: metrics.files_added as u64,
+            };
+        }
+    }
+    Ok(0)
+}
+
+/// Apply caller-provided overrides onto a default `CompactionOptions`. NULL
+/// means "no overrides"; the per-field sentinel and overflow contract are
+/// documented on `LanceCompactionOptions`.
+unsafe fn resolve_options(options: *const LanceCompactionOptions) -> Result<CompactionOptions> {
+    let mut resolved = CompactionOptions::default();
+    if options.is_null() {
+        return Ok(resolved);
+    }
+
+    // SAFETY: `options` is non-NULL (checked above) and the caller guarantees
+    // it points to a properly-initialized `LanceCompactionOptions` valid for
+    // the duration of this call. We read by shared reference.
+    let opts = unsafe { &*options };
+
+    if opts.target_rows_per_fragment > 0 {
+        resolved.target_rows_per_fragment =
+            u64_to_usize(opts.target_rows_per_fragment, "target_rows_per_fragment")?;
+    }
+    if opts.max_rows_per_group > 0 {
+        resolved.max_rows_per_group = u64_to_usize(opts.max_rows_per_group, "max_rows_per_group")?;
+    }
+    if opts.max_bytes_per_file > 0 {
+        resolved.max_bytes_per_file =
+            Some(u64_to_usize(opts.max_bytes_per_file, "max_bytes_per_file")?);
+    }
+    if opts.num_threads > 0 {
+        resolved.num_threads = Some(u64_to_usize(opts.num_threads, "num_threads")?);
+    }
+    if opts.batch_size > 0 {
+        resolved.batch_size = Some(u64_to_usize(opts.batch_size, "batch_size")?);
+    }
+    Ok(resolved)
+}
+
+/// Narrow `u64 -> usize` with an explicit error on overflow (32-bit hosts).
+/// Realistic compaction tunings fit in `usize` on every supported target,
+/// but a silent `as` cast would wrap on a 32-bit host.
+fn u64_to_usize(v: u64, field: &'static str) -> Result<usize> {
+    usize::try_from(v).map_err(|_| lance_core::Error::InvalidInput {
+        source: format!("{field}={v} exceeds usize::MAX on this target").into(),
+        location: snafu::location!(),
+    })
+}

src/lib.rs

@@ -17,6 +17,7 @@
 
 mod async_dispatcher;
 mod batch;
+mod compact;
 mod dataset;
 mod delete;
 mod error;
@@ -33,6 +34,7 @@ mod writer;
 
 // Re-export all extern "C" symbols so they appear in the cdylib.
 pub use batch::*;
+pub use compact::*;
 pub use dataset::*;
 pub use delete::*;
 pub use error::{