From 73c731d9deb7772853833cc2cc452f01f98ce00c Mon Sep 17 00:00:00 2001 From: Beinan Date: Sat, 25 Jul 2026 08:41:28 +0000 Subject: [PATCH] fix(master): compact and version-prune _stats so it stops growing unbounded _stats.rollout.lance is written delete-then-append on every stats upsert, so each scan round adds versions and fragments per experiment, and Lance retains every historical manifest until explicitly cleaned. On a long-running master the chain reached 170k+ versions, making cold start and GET /api/v1/experiments progressively slower. Add StatsStore::maintain(): compact_files with materialize_deletions, then cleanup_old_versions with a grace window, reloading the handle around both. The scanner runs it under the existing stats-writer coordination lock on the first round and every Nth round after (STATS_MAINTENANCE_EVERY_N_SCANS, default 12; STATS_HISTORY_TTL_SECS, default 3600), bounded by a timeout and never failing the scan round. Co-Authored-By: Claude --- crates/lance-context-master/src/config.rs | 14 +++ crates/lance-context-master/src/routes.rs | 2 + crates/lance-context-master/src/scanner.rs | 54 ++++++++++- crates/lance-context-master/src/scheduler.rs | 2 + crates/lance-context-master/src/state.rs | 2 + .../lance-context-master/src/stats_store.rs | 95 +++++++++++++++++++ crates/lance-context-master/src/task_store.rs | 2 + 7 files changed, 169 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index eb62a8a..e127068 100644 --- a/crates/lance-context-master/src/config.rs +++ b/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)] diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 3e4b5d6..ad4a3b7 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/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, diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs index 024faf4..cf40f6a 100644 --- a/crates/lance-context-master/src/scanner.rs +++ b/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) -> 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) -> lance::Result { } } -async fn try_scan_once(state: &Arc) -> lance::Result> { +async fn try_scan_once(state: &Arc, maintain: bool) -> lance::Result> { let Some(guard) = state .task_store .try_coordination_lock("stats-writer") @@ -45,6 +81,13 @@ async fn try_scan_once(state: &Arc) -> lance::Result> 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) -> Option> { 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") diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 28044e7..a6e6bc9 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/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, diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index f3b6ac3..b10ff41 100644 --- a/crates/lance-context-master/src/state.rs +++ b/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, diff --git a/crates/lance-context-master/src/stats_store.rs b/crates/lance-context-master/src/stats_store.rs index a8b27e9..c10d39f 100644 --- a/crates/lance-context-master/src/stats_store.rs +++ b/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); diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 09680a0..1924397 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/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,