fix(master): compact and version-prune _stats so it stops growing unbounded

crates/lance-context-master/src/config.rs

@@ -27,6 +27,20 @@ pub struct MasterConfig {
     #[arg(long, env = "SCAN_CONCURRENCY", default_value_t = 8)]
     pub scan_concurrency: usize,
 
+    /// Run maintenance (compaction + old-version cleanup) on the `_stats`
+    /// dataset every Nth stats-scan round. `_stats` is written delete-then-
+    /// append, so each scan adds versions and fragments per experiment; without
+    /// maintenance the manifest chain grows without bound and slows cold start
+    /// and `/experiments`. `0` disables stats maintenance.
+    #[arg(long, env = "STATS_MAINTENANCE_EVERY_N_SCANS", default_value_t = 12)]
+    pub stats_maintenance_every_n_scans: u64,
+
+    /// Grace window, in seconds, for `_stats` old-version cleanup. Versions
+    /// newer than this are never removed, so in-flight readers on another
+    /// replica keep working.
+    #[arg(long, env = "STATS_HISTORY_TTL_SECS", default_value_t = 3_600)]
+    pub stats_history_ttl_secs: u64,
+
     /// Interval, in seconds, between automatic compaction sweeps. `0` disables
     /// automatic compaction (manual triggers still work).
     #[arg(long, env = "COMPACTION_INTERVAL_SECS", default_value_t = 600)]

crates/lance-context-master/src/routes.rs

@@ -538,6 +538,8 @@ mod tests {
             port: 0,
             stats_scan_interval_secs: 0,
             scan_concurrency: 4,
+            stats_maintenance_every_n_scans: 0,
+            stats_history_ttl_secs: 3_600,
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,

crates/lance-context-master/src/scanner.rs

@@ -22,6 +22,42 @@ use crate::stats_store::StatRow;
 /// scan round.
 const OBSERVE_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// Bound on one `_stats` maintenance pass (compaction + version cleanup) so a
+/// slow object store cannot wedge the scanner loop.
+const MAINTENANCE_TIMEOUT: Duration = Duration::from_secs(300);
+
+/// Compact `_stats` and prune its old manifest versions.
+///
+/// `_stats` is written delete-then-append (one upsert per experiment per
+/// round), so its version chain and fragment count grow every scan and Lance
+/// never reclaims them on its own. Callers must hold the `stats-writer`
+/// coordination lock so only one replica ever rewrites the dataset.
+pub async fn maintain_stats(state: &Arc<MasterState>) -> lance::Result<()> {
+    let ttl = Duration::from_secs(state.config.stats_history_ttl_secs);
+    let start = std::time::Instant::now();
+    let mut stats = state.stats.lock().await;
+    match tokio::time::timeout(MAINTENANCE_TIMEOUT, stats.maintain(ttl)).await {
+        Ok(Ok((compaction, removal))) => {
+            metrics::histogram!("master_stats_maintenance_duration_seconds")
+                .record(start.elapsed().as_secs_f64());
+            metrics::counter!("master_stats_versions_removed_total")
+                .increment(removal.old_versions);
+            metrics::gauge!("master_stats_version").set(stats.version() as f64);
+            tracing::info!(
+                fragments_removed = compaction.fragments_removed,
+                fragments_added = compaction.fragments_added,
+                old_versions_removed = removal.old_versions,
+                bytes_removed = removal.bytes_removed,
+                version = stats.version(),
+                "stats maintenance complete"
+            );
+            Ok(())
+        }
+        Ok(Err(e)) => Err(e),
+        Err(_) => Err(lance::Error::io("stats maintenance timed out")),
+    }
+}
+
 /// Run a single scan pass: refresh every experiment's stats row and drop rows
 /// for experiments no longer in the registry. Returns the number of
 /// experiments successfully observed.
@@ -36,7 +72,7 @@ pub async fn scan_once(state: &Arc<MasterState>) -> lance::Result<usize> {
     }
 }
 
-async fn try_scan_once(state: &Arc<MasterState>) -> lance::Result<Option<usize>> {
+async fn try_scan_once(state: &Arc<MasterState>, maintain: bool) -> lance::Result<Option<usize>> {
     let Some(guard) = state
         .task_store
         .try_coordination_lock("stats-writer")
@@ -45,6 +81,13 @@ async fn try_scan_once(state: &Arc<MasterState>) -> lance::Result<Option<usize>>
         return Ok(None);
     };
     let result = scan_once_inner(state).await;
+    // Maintenance runs under the same writer lock as the scan that just
+    // created the versions, and its failure never fails the scan round.
+    if maintain {
+        if let Err(e) = maintain_stats(state).await {
+            tracing::warn!(error = %e, "stats maintenance failed");
+        }
+    }
     let release = state.task_store.release_coordination_lock(guard).await;
     match (result, release) {
         (Ok(count), Ok(())) => Ok(Some(count)),
@@ -179,9 +222,16 @@ pub fn spawn_scanner(state: &Arc<MasterState>) -> Option<JoinHandle<()>> {
     let state = state.clone();
     Some(tokio::spawn(async move {
         let mut ticker = tokio::time::interval(Duration::from_secs(interval_secs));
+        let every_n = state.config.stats_maintenance_every_n_scans;
+        let mut round: u64 = 0;
         loop {
             ticker.tick().await;
-            match try_scan_once(&state).await {
+            round += 1;
+            // Round 1 included: an existing deployment may start with a very
+            // long version chain, and waiting N rounds to reclaim it would
+            // leave cold start slow for another N intervals.
+            let maintain = every_n > 0 && (round == 1 || round.is_multiple_of(every_n));
+            match try_scan_once(&state, maintain).await {
                 Ok(Some(n)) => tracing::info!(experiments = n, "stats scan complete"),
                 Ok(None) => {
                     tracing::debug!("stats scan skipped; another master owns the writer lock")

crates/lance-context-master/src/scheduler.rs

@@ -430,6 +430,8 @@ mod tests {
             port: 0,
             stats_scan_interval_secs: 0,
             scan_concurrency: 4,
+            stats_maintenance_every_n_scans: 0,
+            stats_history_ttl_secs: 3_600,
             compaction_interval_secs: 0,
             // Low threshold so a handful of appends crosses it.
             min_fragments: 2,

crates/lance-context-master/src/state.rs

@@ -131,6 +131,8 @@ mod tests {
             port: 0,
             stats_scan_interval_secs: 0,
             scan_concurrency: 4,
+            stats_maintenance_every_n_scans: 0,
+            stats_history_ttl_secs: 3_600,
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,

crates/lance-context-master/src/stats_store.rs

@@ -13,9 +13,13 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use std::time::Duration;
+
 use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray};
 use arrow_schema::{ArrowError, DataType, Field, Schema};
 use futures::TryStreamExt;
+use lance::dataset::cleanup::RemovalStats;
+use lance::dataset::optimize::{compact_files, CompactionMetrics, CompactionOptions};
 use lance::dataset::{builder::DatasetBuilder, Dataset, WriteMode, WriteParams};
 use lance::io::{ObjectStoreParams, StorageOptionsAccessor};
 use lance::{Error as LanceError, Result as LanceResult};
@@ -312,6 +316,56 @@ impl StatsStore {
     pub fn uri(&self) -> &str {
         &self.uri
     }
+
+    /// Current dataset version (manifest chain head). Exposed for metrics and
+    /// tests asserting that maintenance actually bounds growth.
+    pub fn version(&self) -> u64 {
+        self.dataset.version().version
+    }
+
+    /// Fold the many tiny append fragments produced by [`Self::upsert`] into a
+    /// few, then drop manifest versions older than `older_than`.
+    ///
+    /// `_stats` is written delete-then-append, so *every* upsert adds one or
+    /// two dataset versions and at least one fragment. Lance retains every
+    /// historical manifest until explicitly cleaned, so without this the
+    /// version chain grows without bound (observed: 170k+ versions), and any
+    /// open/checkout/history traversal pays for the whole chain.
+    ///
+    /// Both halves are best-effort in the sense that a failure is returned to
+    /// the caller to log; neither is required for correctness of reads. The
+    /// cleanup never touches versions newer than the grace window, so a
+    /// concurrent reader holding a recent version is unaffected.
+    ///
+    /// Callers must serialize this with other mutations (it takes `&mut self`)
+    /// and, across master replicas, hold the `stats-writer` coordination lock.
+    pub async fn maintain(
+        &mut self,
+        older_than: Duration,
+    ) -> LanceResult<(CompactionMetrics, RemovalStats)> {
+        self.dataset.checkout_latest().await?;
+
+        let options = CompactionOptions {
+            // One small row per experiment: everything belongs in one fragment.
+            target_rows_per_fragment: 1_048_576,
+            // Deletions are the whole point here — every upsert leaves one.
+            materialize_deletions: true,
+            materialize_deletions_threshold: 0.0,
+            ..Default::default()
+        };
+        let compaction = compact_files(&mut self.dataset, options, None).await?;
+
+        // Re-open so the handle (and the cleanup below) sees the rewritten
+        // version rather than the pre-compaction manifest.
+        self.dataset = Self::load(&self.uri, self.storage_options.clone()).await?;
+
+        let grace = chrono::TimeDelta::from_std(older_than)
+            .map_err(|e| LanceError::io(format!("invalid stats history TTL: {e}")))?;
+        let removal = self.dataset.cleanup_old_versions(grace, None, None).await?;
+        self.dataset = Self::load(&self.uri, self.storage_options.clone()).await?;
+
+        Ok((compaction, removal))
+    }
 }
 
 #[cfg(test)]
@@ -394,6 +448,47 @@ mod tests {
         assert_eq!(s.get("persist").await.unwrap().unwrap().row_count, 7);
     }
 
+    #[tokio::test]
+    async fn maintain_bounds_versions_and_preserves_rows() {
+        let dir = TempDir::new().unwrap();
+        let mut s = new_store(&dir).await;
+        for i in 0..20 {
+            s.upsert(&sample("exp-a", i)).await.unwrap();
+            s.upsert(&sample("exp-b", i)).await.unwrap();
+        }
+        let before = s.version();
+        assert!(before > 20, "expected version churn, got {before}");
+
+        // Zero grace window: everything older than "now" is prunable.
+        let (compaction, removal) = s.maintain(Duration::from_secs(0)).await.unwrap();
+        assert!(removal.old_versions > 0, "no versions reclaimed");
+        assert!(compaction.fragments_removed > 0, "nothing compacted");
+
+        // Data survives maintenance unchanged.
+        let rows = s.list(None, 100, 0).await.unwrap();
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0].name, "exp-a");
+        assert_eq!(rows[0].row_count, 19);
+        assert_eq!(rows[1].row_count, 19);
+
+        // And the store keeps working afterwards.
+        s.upsert(&sample("exp-c", 1)).await.unwrap();
+        assert_eq!(s.count(None).await.unwrap(), 3);
+    }
+
+    #[tokio::test]
+    async fn maintain_respects_grace_window() {
+        let dir = TempDir::new().unwrap();
+        let mut s = new_store(&dir).await;
+        for i in 0..5 {
+            s.upsert(&sample("exp-a", i)).await.unwrap();
+        }
+        // A wide grace window must leave the recent history alone.
+        let (_, removal) = s.maintain(Duration::from_secs(86_400)).await.unwrap();
+        assert_eq!(removal.old_versions, 0);
+        assert_eq!(s.list(None, 10, 0).await.unwrap().len(), 1);
+    }
+
     #[tokio::test]
     async fn no_compaction_sentinel_maps_to_none() {
         let row = sample("x", 1);

crates/lance-context-master/src/task_store.rs

@@ -910,6 +910,8 @@ mod tests {
             port: 0,
             stats_scan_interval_secs: 0,
             scan_concurrency: 4,
+            stats_maintenance_every_n_scans: 0,
+            stats_history_ttl_secs: 3_600,
             compaction_interval_secs: 0,
             min_fragments: 16,
             target_rows_per_fragment: 1_048_576,