feat: implement background compaction for Lance fragments

Implements issue #16 with comprehensive compaction functionality:

**Core Features:**
- Manual compaction via `compact()` method
- Optional background compaction with configurable intervals
- Comprehensive configuration (thresholds, quiet hours, intervals)
- Advanced observability (stats API, metrics, logging)

**Implementation Details:**
- Rust: Added CompactionConfig, CompactionStats types to store.rs
- Rust: Implemented compact(), should_compact(), compaction_stats()
- Rust: Background task with Tokio interval timer and graceful shutdown
- Python: PyO3 bindings for all compaction methods
- Python: High-level API with full docstrings
- Tests: 10 comprehensive tests (all passing)

**Configuration Options:**
- enable_background_compaction: Enable auto-compaction
- compaction_interval_secs: Check interval (default: 300s)
- compaction_min_fragments: Trigger threshold (default: 5)
- compaction_target_rows: Target rows per fragment (default: 1M)
- quiet_hours: Skip compaction during specified hours

**Metrics Returned:**
- fragments_removed/added
- files_removed/added
- is_compacting status
- last_compaction timestamp
- total_compactions count

All tests pass. Documentation updated with usage examples.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: jja725

Cargo.lock

@@ -17,6 +17,7 @@ Key motivations inspired by the broader Lance roadmap<sup>[1](https://github.com
 
 - Unified schema for agent messages (`ContextRecord`) with optional embeddings and metadata.
 - Automatic versioning via Lance manifests with `checkout(version)` support.
+- Background compaction to optimize storage and read performance.
 - Remote persistence: point the store at `s3://` URIs with either AWS environment variables or explicit credentials/endpoint overrides.
 - Python API (`lance_context.api.Context`) aligned with the Rust implementation.
 - Integration tests that exercise real persistence, image serialization, and version rollbacks.
@@ -77,6 +78,27 @@ ctx = Context.create(
     allow_http=True,
 )
 # AWS_* environment variables work too—pass overrides only when you need custom endpoints.
+
+# Background Compaction - optimize storage and read performance
+ctx = Context.create(
+    "context.lance",
+    enable_background_compaction=True,  # Enable automatic compaction
+    compaction_interval_secs=300,       # Check every 5 minutes
+    compaction_min_fragments=10,        # Trigger when 10+ fragments exist
+    quiet_hours=[(22, 6)],              # Skip compaction 10pm-6am
+)
+
+# Manual compaction control
+for i in range(100):
+    ctx.add("user", f"message {i}")  # Creates many small fragments
+
+# Check compaction status
+stats = ctx.compaction_stats()
+print(f"Fragments: {stats['total_fragments']}")
+
+# Manually trigger compaction
+metrics = ctx.compact()
+print(f"Compaction removed {metrics['fragments_removed']} fragments")
 ```
 
 ### Rust
@@ -121,7 +143,7 @@ We are tracking future enhancements as GitHub issues:
 
 - [Support S3-backed context stores](https://github.com/lance-format/lance-context/issues/14)
 - [Add relationship column for GraphRAG workflows](https://github.com/lance-format/lance-context/issues/15)
-- [Background compaction for Lance fragments](https://github.com/lance-format/lance-context/issues/16)
+- ~~[Background compaction for Lance fragments](https://github.com/lance-format/lance-context/issues/16)~~ ✅ **Implemented**
 
 Contributions are welcome—feel free to comment on the issues above or open your own proposals.
 

README.md

@@ -21,6 +21,8 @@ lance-namespace = "1.0.1"
 lance-graph = "0.4.0"
 serde = { version = "1", features = ["derive"] }
 futures = "0.3"
+tokio = { version = "1", features = ["sync", "time"] }
+tracing = "0.1"
 
 [dev-dependencies]
 tempfile = "3"

crates/lance-context-core/Cargo.toml

@@ -7,4 +7,7 @@ mod store;
 
 pub use context::{Context, ContextEntry, Snapshot};
 pub use record::{ContextRecord, SearchResult, StateMetadata};
-pub use store::{ContextStore, ContextStoreOptions};
+pub use store::{CompactionConfig, CompactionStats, ContextStore, ContextStoreOptions};
+
+// Re-export CompactionMetrics from lance for Python bindings
+pub use lance::dataset::optimize::CompactionMetrics;

crates/lance-context-core/src/lib.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
 use arrow_array::builder::{
     FixedSizeListBuilder, Float32Builder, Int32Builder, LargeBinaryBuilder, LargeStringBuilder,
@@ -12,28 +13,98 @@ use arrow_array::{
     TimestampMicrosecondArray,
 };
 use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema, TimeUnit};
-use chrono::DateTime;
+use chrono::{DateTime, Timelike, Utc};
 use futures::TryStreamExt;
+use lance::dataset::optimize::{compact_files, CompactionMetrics, CompactionOptions};
 use lance::dataset::{builder::DatasetBuilder, Dataset, WriteMode, WriteParams};
 use lance::io::ObjectStoreParams;
 use lance::{Error as LanceError, Result as LanceResult};
+use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
+use tracing::{error, info, warn};
 
 use crate::record::{ContextRecord, SearchResult, StateMetadata};
 
 /// Embedding length used for the semantic index column.
 const DEFAULT_EMBEDDING_DIM: i32 = 1536;
 const DEFAULT_SEARCH_LIMIT: usize = 10;
 
+/// Configuration for background compaction.
+#[derive(Debug, Clone)]
+pub struct CompactionConfig {
+    /// Whether background compaction is enabled.
+    pub enabled: bool,
+    /// Minimum number of fragments to trigger compaction.
+    pub min_fragments: usize,
+    /// Target rows per fragment after compaction.
+    pub target_rows_per_fragment: usize,
+    /// Maximum rows per row group.
+    pub max_rows_per_group: usize,
+    /// Whether to materialize (remove) deleted rows during compaction.
+    pub materialize_deletions: bool,
+    /// Deletion threshold (0.0-1.0) to trigger materialization.
+    pub materialize_deletions_threshold: f32,
+    /// Number of threads for compaction (None = auto).
+    pub num_threads: Option<usize>,
+    /// Interval in seconds between compaction checks.
+    pub check_interval_secs: u64,
+    /// Quiet hours during which compaction is skipped [(start_hour, end_hour)].
+    pub quiet_hours: Vec<(u8, u8)>,
+}
+
+impl Default for CompactionConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            min_fragments: 5,
+            target_rows_per_fragment: 1_000_000,
+            max_rows_per_group: 1024,
+            materialize_deletions: true,
+            materialize_deletions_threshold: 0.1,
+            num_threads: None,
+            check_interval_secs: 300,
+            quiet_hours: vec![],
+        }
+    }
+}
+
+/// Statistics about compaction status and history.
+#[derive(Debug, Clone)]
+pub struct CompactionStats {
+    /// Current number of fragments in the dataset.
+    pub total_fragments: usize,
+    /// Whether a compaction is currently in progress.
+    pub is_compacting: bool,
+    /// Timestamp of the last successful compaction.
+    pub last_compaction: Option<DateTime<Utc>>,
+    /// Error message from the last failed compaction.
+    pub last_error: Option<String>,
+    /// Total number of successful compactions performed.
+    pub total_compactions: u64,
+}
+
+/// Internal state for tracking background compaction.
+struct CompactionState {
+    background_task: Option<JoinHandle<()>>,
+    is_compacting: bool,
+    last_compaction: Option<DateTime<Utc>>,
+    last_error: Option<String>,
+    total_compactions: u64,
+}
+
 /// Persistent Lance-backed context store.
 #[derive(Clone)]
 pub struct ContextStore {
     dataset: Dataset,
+    compaction_state: Arc<Mutex<CompactionState>>,
+    pub compaction_config: CompactionConfig,
 }
 
 /// Additional configuration when opening a [`ContextStore`].
 #[derive(Debug, Clone, Default)]
 pub struct ContextStoreOptions {
     pub storage_options: Option<HashMap<String, String>>,
+    pub compaction: CompactionConfig,
 }
 
 impl ContextStoreOptions {
@@ -52,14 +123,30 @@ impl ContextStore {
     /// Open a dataset with explicit object store configuration (e.g. S3 credentials).
     pub async fn open_with_options(uri: &str, options: ContextStoreOptions) -> LanceResult<Self> {
         let storage_options = options.storage_options();
-        match Self::load_with_options(uri, storage_options.clone()).await {
-            Ok(dataset) => Ok(Self { dataset }),
+        let dataset = match Self::load_with_options(uri, storage_options.clone()).await {
+            Ok(dataset) => dataset,
             Err(LanceError::DatasetNotFound { .. }) => {
-                let dataset = Self::create_with_options(uri, storage_options).await?;
-                Ok(Self { dataset })
+                Self::create_with_options(uri, storage_options).await?
             }
-            Err(err) => Err(err),
-        }
+            Err(err) => return Err(err),
+        };
+
+        let mut store = Self {
+            dataset,
+            compaction_state: Arc::new(Mutex::new(CompactionState {
+                background_task: None,
+                is_compacting: false,
+                last_compaction: None,
+                last_error: None,
+                total_compactions: 0,
+            })),
+            compaction_config: options.compaction,
+        };
+
+        // Start background compaction if enabled
+        store.start_background_compaction().await?;
+
+        Ok(store)
     }
 
     /// Append context records to the store and return the new dataset version.
@@ -146,6 +233,166 @@ impl ContextStore {
         Ok(results)
     }
 
+    /// Manually trigger compaction to merge small fragments.
+    pub async fn compact(&mut self, options: Option<CompactionConfig>) -> LanceResult<CompactionMetrics> {
+        let config = options.unwrap_or_else(|| self.compaction_config.clone());
+
+        info!("Starting compaction: {} fragments", self.dataset.count_fragments());
+        let start = std::time::Instant::now();
+
+        // Mark as compacting
+        {
+            let mut state = self.compaction_state.lock().await;
+            if state.is_compacting {
+                warn!("Compaction already in progress, skipping");
+                return Err(LanceError::from(ArrowError::InvalidArgumentError(
+                    "Compaction already in progress".to_string(),
+                )));
+            }
+            state.is_compacting = true;
+        }
+
+        // Build Lance CompactionOptions
+        let lance_options = CompactionOptions {
+            target_rows_per_fragment: config.target_rows_per_fragment,
+            max_rows_per_group: config.max_rows_per_group,
+            materialize_deletions: config.materialize_deletions,
+            materialize_deletions_threshold: config.materialize_deletions_threshold,
+            num_threads: config.num_threads,
+            ..Default::default()
+        };
+
+        // Run compaction
+        let result = compact_files(&mut self.dataset, lance_options, None).await;
+
+        // Update state
+        let mut state = self.compaction_state.lock().await;
+        state.is_compacting = false;
+
+        match result {
+            Ok(metrics) => {
+                state.last_compaction = Some(Utc::now());
+                state.total_compactions += 1;
+                state.last_error = None;
+
+                info!(
+                    "Compaction completed in {:?}: removed {} fragments ({}files), added {} fragments ({} files)",
+                    start.elapsed(),
+                    metrics.fragments_removed,
+                    metrics.files_removed,
+                    metrics.fragments_added,
+                    metrics.files_added
+                );
+
+                // Reload dataset to see new version
+                self.dataset = Dataset::open(self.dataset.uri()).await?;
+
+                Ok(metrics)
+            }
+            Err(e) => {
+                error!("Compaction failed: {}", e);
+                state.last_error = Some(e.to_string());
+                Err(e)
+            }
+        }
+    }
+
+    /// Check if compaction should run based on configuration thresholds.
+    pub async fn should_compact(&self) -> LanceResult<bool> {
+        let fragment_count = self.dataset.count_fragments();
+
+        if fragment_count < self.compaction_config.min_fragments {
+            return Ok(false);
+        }
+
+        // Check quiet hours
+        if !self.compaction_config.quiet_hours.is_empty() {
+            let now = Utc::now();
+            let current_hour = now.hour() as u8;
+
+            for (start, end) in &self.compaction_config.quiet_hours {
+                if current_hour >= *start && current_hour < *end {
+                    info!("Skipping compaction during quiet hours ({}-{})", start, end);
+                    return Ok(false);
+                }
+            }
+        }
+
+        Ok(true)
+    }
+
+    /// Get current compaction statistics.
+    pub async fn compaction_stats(&self) -> LanceResult<CompactionStats> {
+        let state = self.compaction_state.lock().await;
+
+        Ok(CompactionStats {
+            total_fragments: self.dataset.count_fragments(),
+            is_compacting: state.is_compacting,
+            last_compaction: state.last_compaction,
+            last_error: state.last_error.clone(),
+            total_compactions: state.total_compactions,
+        })
+    }
+
+    /// Start background compaction task if enabled.
+    async fn start_background_compaction(&mut self) -> LanceResult<()> {
+        if !self.compaction_config.enabled {
+            return Ok(());
+        }
+
+        let mut state = self.compaction_state.lock().await;
+        if state.background_task.is_some() {
+            warn!("Background compaction already running");
+            return Ok(());
+        }
+
+        info!(
+            "Starting background compaction (interval: {}s, min fragments: {})",
+            self.compaction_config.check_interval_secs, self.compaction_config.min_fragments
+        );
+
+        let mut store_clone = self.clone();
+        let interval_secs = self.compaction_config.check_interval_secs;
+
+        let task = tokio::spawn(async move {
+            let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
+
+            loop {
+                interval.tick().await;
+
+                match store_clone.should_compact().await {
+                    Ok(true) => {
+                        info!("Background compaction triggered");
+                        if let Err(e) = store_clone.compact(None).await {
+                            error!("Background compaction failed: {}", e);
+                        }
+                    }
+                    Ok(false) => {
+                        // Not needed or in quiet hours
+                    }
+                    Err(e) => {
+                        error!("Error checking compaction need: {}", e);
+                    }
+                }
+            }
+        });
+
+        state.background_task = Some(task);
+        Ok(())
+    }
+
+    /// Stop background compaction task.
+    pub async fn stop_background_compaction(&mut self) -> LanceResult<()> {
+        let mut state = self.compaction_state.lock().await;
+
+        if let Some(task) = state.background_task.take() {
+            info!("Stopping background compaction");
+            task.abort();
+        }
+
+        Ok(())
+    }
+
     /// Lance schema for the context store.
     pub fn schema() -> Schema {
         Schema::new(vec![
@@ -368,6 +615,17 @@ impl ContextStore {
     }
 }
 
+impl Drop for ContextStore {
+    fn drop(&mut self) {
+        // Best-effort cleanup of background task
+        if let Ok(mut state) = self.compaction_state.try_lock() {
+            if let Some(task) = state.background_task.take() {
+                task.abort();
+            }
+        }
+    }
+}
+
 fn batch_to_search_results(batch: &RecordBatch) -> LanceResult<Vec<SearchResult>> {
     let records = batch_to_records(batch)?;